Add a REST API to be able to configure devices (#8)

* Add API skeleton

* Add another route

* Remove useless callback

* Rename handler

* Add configuration parser

* Add configuration parser

* Fix configuration parser

* Fix another bug in configuration parser

* Add body decoding

* Fix possible errors

* Add configuration parser tests

* Add a property based test

* Add property based tests

* Add missing tests

* Use internal events instead of message

* Feature/update config (#4)

* Add update configuration

* Add fetch configs

* Change the README to reStructuredText

* Fix command

* Fix formatting

* Add configurable synchronization interval

* Modify response and validation

* Fix tests

* Add OTP 22 to CI

* Change minimum coverage

* Add an encoder

* Fix type spec

* Add function to iterate over config fields

* Add more tests

* Fix devices handler

* Fix a bug in encoder

* Add property based tests for the encoder

* Stop breaking config API

* Remove trx plugin

* Add tests for the encoder

* Use a macro for the UUID

* Remove unused variable

* Raise code coverage

* Add appveyor configuration

* Add appveyor badge

* Fix appveyor

* Fix erlang version in appveyor

.travis.yml

@@ -1,6 +1,7 @@
 language: erlang
 otp_release:
     - 21.3
+    - 22.1
 
 script:
     - rebar3 check

README.rst

@@ -0,0 +1,76 @@
+==============
+mqtt-simulator
+==============
+
+.. image:: https://travis-ci.org/AntoineGagne/mqtt-simulator.svg?branch=master
+    :target: https://travis-ci.org/AntoineGagne/mqtt-simulator
+
+.. image:: https://ci.appveyor.com/api/projects/status/glyeekdu4vum33ht/branch/master?svg=true
+    :target: https://ci.appveyor.com/api/projects/status/glyeekdu4vum33ht/branch/master
+
+:Author: `Antoine Gagné <[email protected]>`_
+
+.. contents::
+    :backlinks: none
+
+.. sectnum::
+
+Installation
+============
+
+Local Build
+-----------
+
+To build the runnable release, you need to have Erlang with OTP 21 and above.
+You also need ``rebar3``. Then, you can run the following command:
+
+.. code-block:: sh
+
+    rebar3 as prod release
+
+Docker Image
+------------
+
+To build this image, you can use the following command:
+
+.. code-block:: sh
+
+    docker build -f Dockerfile -t "${name_of_the_image}" .
+
+Usage
+=====
+
+From Local Build
+----------------
+
+If you built the release, you can run it with:
+
+.. code-block:: sh
+
+    ./_build/prod/rel/mqtt_simulator/bin/mqtt_simulator foreground
+
+
+Docker
+------
+
+After building the image, you can run the image by using the following command:
+
+.. code-block:: sh
+
+    docker run \
+        --detach \
+        --name "${name_of_the_running_container}" \
+        --publish "${port_on_host}:${port_of_simulator:-8000}" \
+        "${name_of_the_image}"
+
+Development
+===========
+
+Running all the tests and linters
+---------------------------------
+
+You can run all the tests and linters with the ``rebar3`` alias:
+
+.. code-block:: sh
+
+    rebar3 check

apps/mqtt_simulator/src/mqtt_simulator.app.src

@@ -7,7 +7,10 @@
    [kernel,
     stdlib,
     gproc,
-    emqttc
+    emqttc,
+    cowboy,
+    jsone,
+    uuid
    ]},
   {env,[]},
   {modules, []},

apps/mqtt_simulator/src/mqtt_simulator_api.erl

@@ -0,0 +1,58 @@
+-module(mqtt_simulator_api).
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+
+%% gen_server callbacks
+-export([init/1,
+         handle_call/3,
+         handle_cast/2,
+         handle_info/2,
+         terminate/2]).
+
+-define(SERVER, ?MODULE).
+-define(DEFAULT_PORT, 8000).
+
+-record(state, {}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+init([]) ->
+    process_flag(trap_exit, true),
+    Port = application:get_env(mqtt_simulator, api_port, ?DEFAULT_PORT),
+    Routes = [{"/devices/", mqtt_simulator_devices_handler, []},
+              {"/devices/:id", [{id, nonempty}], mqtt_simulator_devices_handler, []}],
+    Dispatch = cowboy_router:compile([{'_', Routes}]),
+    {ok, _} = cowboy:start_clear(?SERVER,
+                                 [{port, Port}],
+                                 #{env => #{dispatch => Dispatch}}
+                                ),
+    {ok, #state{}}.
+
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+terminate(_Reason, _State) ->
+    cowboy:stop_listener(?SERVER).
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/mqtt_simulator/src/mqtt_simulator_api_sup.erl

@@ -0,0 +1,36 @@
+-module(mqtt_simulator_api_sup).
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+%%====================================================================
+%% API functions
+%%====================================================================
+
+-spec start_link() ->
+    {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+%%====================================================================
+%% Supervisor callbacks
+%%====================================================================
+
+init([]) ->
+    {ok, {#{strategy => one_for_one,
+            intensity => 5,
+            period => 10},
+          [#{id => mqtt_simulator_api,
+             start => {mqtt_simulator_api, start_link, []},
+             restart => permanent,
+             shutdown => 5000,
+             type => worker,
+             modules => [mqtt_simulator_api]}
+          ]}}.

apps/mqtt_simulator/src/mqtt_simulator_client.erl

@@ -5,6 +5,7 @@
 -behavior(gen_statem).
 
 -export([start_link/3,
+         update_config/2,
          publish/3]).
 
 -export([callback_mode/0,
@@ -30,6 +31,10 @@
 start_link(Id, ConfigId, Config) ->
     gen_statem:start_link(?VIA_GPROC(Id), ?MODULE, [ConfigId, Config], []).
 
+-spec update_config(term(), mqtt_simulator_client_config:config()) -> ok.
+update_config(Id, Config) ->
+    gen_statem:cast(?VIA_GPROC(Id), {update_config, Config}).
+
 -spec publish(term(), binary(), binary()) -> ok.
 publish(Id, Topic, Payload) ->
     gen_statem:cast(?VIA_GPROC(Id), {publish, Topic, Payload}).
@@ -44,10 +49,18 @@ callback_mode() ->
 init([ConfigId, Config]) ->
     process_flag(trap_exit, true),
     ReconnectTimeout = mqtt_simulator_client_config:reconnect_timeout(Config),
-    self() ! connect,
+    NextEvent = {next_event, internal, connect},
     {ok, disconnected, #data{config_id = ConfigId,
                              config = Config,
-                             reconnect_timeout = ReconnectTimeout}}.
+                             reconnect_timeout = ReconnectTimeout}, NextEvent}.
+
+handle_event(cast, {update_config, Config}, State, Data) ->
+    ?LOG_INFO(#{what => update_config, id => mqtt_simulator_client_config:id(Config),
+                state => State}),
+    DataPoints = mqtt_simulator_client_config:data(Config),
+    ok = mqtt_simulator_data_simulators_config:update_config(Data#data.config_id, DataPoints),
+    UpdatedData = maybe_close_connection(Data),
+    try_connect(UpdatedData#data{config = Config});
 
 handle_event(cast, {publish, Topic, Payload}, connected, #data{client = Client}) ->
     emqttc:publish(Client, Topic, Payload),
@@ -57,7 +70,7 @@ handle_event(cast, {publish, Topic, Payload}, disconnected, _Data) ->
                    reason => disconnected}),
     keep_state_and_data;
 
-handle_event(info, connect, disconnected, Data=#data{config = Config, config_id = ConfigId}) ->
+handle_event(internal, connect, disconnected, Data=#data{config = Config, config_id = ConfigId}) ->
     DataPoints = mqtt_simulator_client_config:data(Config),
     ok = mqtt_simulator_data_simulators_config:update_config(ConfigId, DataPoints),
     try_connect(Data);
@@ -100,6 +113,12 @@ terminate(Reason, _, _) ->
 %% Internal functions
 %%====================================================================
 
+maybe_close_connection(Data=#data{client = undefined}) ->
+    Data;
+maybe_close_connection(Data=#data{client = Client}) ->
+    ok = emqttc:disconnect(Client),
+    Data#data{client = undefined}.
+
 try_connect(Data=#data{config = Config}) ->
     DefaultConfig = default_config(),
     AdaptedConfig = to_mqttc_config(Config),

apps/mqtt_simulator/src/mqtt_simulator_client_config.erl

@@ -2,6 +2,7 @@
 
 %% API
 -export([init/0,
+         fold/3,
          id/1,
          id/2,
          host/1,
@@ -33,6 +34,7 @@
 -export_type([config/0,
               data/0]).
 
+-define(UUID, uuid:uuid_to_string(uuid:get_v4(), binary_standard)).
 -define(DEFAULT_RECONNECT_TIMEOUT, 5000).
 
 %%%===================================================================
@@ -41,12 +43,16 @@
 
 -spec init() -> config().
 init() ->
-    #{id => <<"">>,
+    #{id => ?UUID,
       host => <<"">>,
       port => 0,
       reconnect_timeout => ?DEFAULT_RECONNECT_TIMEOUT,
       data => []}.
 
+-spec fold(fun ((Key :: term(), Value :: term(), Acc) -> Acc), Acc, config()) -> Acc.
+fold(Fun, Acc, Config) ->
+    maps:fold(Fun, Acc, Config).
+
 -spec id(binary(), config()) -> config().
 id(Id, Config) ->
     Config#{id := Id}.

apps/mqtt_simulator/src/mqtt_simulator_client_sup.erl

@@ -3,13 +3,15 @@
 -behaviour(supervisor).
 
 %% API
--export([start_link/2]).
+-export([start_link/2,
+         update_config/1]).
 
 %% Supervisor callbacks
 -export([init/1]).
 
 -define(VIA_GPROC(Id), {via, gproc, {n, l, Id}}).
 -define(CLIENT_ID(Id), {client, Id}).
+-define(WHERE(Id), gproc:where({n, l, Id})).
 -define(SUP_ID(Id), {data_simulator_sup_id, Id}).
 -define(CONFIG_ID(Id), {data_simulator_config_id, Id}).
 -define(DEFAULT_SYNCHRONIZATION_INTERVAL, 60000).
@@ -23,6 +25,15 @@
 start_link(Id, Config) ->
     supervisor:start_link(?VIA_GPROC(Id), ?MODULE, [Config]).
 
+-spec update_config(mqtt_simulator_client_config:config()) -> ok | {error, not_found}.
+update_config(Config) ->
+    Id = mqtt_simulator_client_config:id(Config),
+    ClientId = ?CLIENT_ID(Id),
+    case ?WHERE(ClientId) of
+        undefined -> {error, {not_found, Id}};
+        _ -> mqtt_simulator_client:update_config(ClientId, Config)
+    end.
+
 %%====================================================================
 %% Supervisor callbacks
 %%====================================================================
@@ -32,6 +43,9 @@ init([Config]) ->
     ClientId = ?CLIENT_ID(Id),
     ConfigId = ?CONFIG_ID(Id),
     SupId = ?SUP_ID(Id),
+    SynchronizationInterval = application:get_env(mqtt_simulator,
+                                                  synchronization_interval,
+                                                  ?DEFAULT_SYNCHRONIZATION_INTERVAL),
     {ok, {#{strategy => one_for_all,
             intensity => 5,
             period => 10},
@@ -44,7 +58,7 @@ init([Config]) ->
              modules => [mqtt_simulator_data_simulators_sup]},
            #{id => mqtt_simulator_data_simulators_config,
              start => {mqtt_simulator_data_simulators_config, start_link,
-                       [ConfigId, SupId, ?DEFAULT_SYNCHRONIZATION_INTERVAL]},
+                       [ConfigId, SupId, SynchronizationInterval]},
              restart => permanent,
              shutdown => 5000,
              type => worker,