Merge pull request #1094 from emqtt/develop

Add a new listener for HTTP REST API

Makefile

@@ -14,6 +14,7 @@ dep_pbkdf2       = git https://github.com/emqtt/pbkdf2 2.0.1
 dep_lager_syslog = git https://github.com/basho/lager_syslog
 dep_bcrypt       = git https://github.com/smarkets/erlang-bcrypt master
 
+ERLC_OPTS += +debug_info
 ERLC_OPTS += +'{parse_transform, lager_transform}'
 
 NO_AUTOPATCH = cuttlefish

etc/emq.conf

@@ -457,6 +457,17 @@ listener.wss.external.certfile = {{ platform_etc_dir }}/certs/cert.pem
 
 ## listener.wss.external.fail_if_no_peer_cert = true
 
+##--------------------------------------------------------------------
+## HTTP Management API Listener
+
+listener.api.mgmt = 127.0.0.1:8080
+
+listener.api.mgmt.acceptors = 4
+
+listener.api.mgmt.max_clients = 64
+
+listener.api.mgmt.access.1 = allow all
+
 ##-------------------------------------------------------------------
 ## System Monitor
 ##-------------------------------------------------------------------

priv/emq.schema

@@ -1027,9 +1027,89 @@ end}.
                   ++
                   [SslListeners(Type, Name) || {["listener", Type, Name], ListenOn}
                                                <- cuttlefish_variable:filter_by_prefix("listener.ssl", Conf)
-                                               ++ cuttlefish_variable:filter_by_prefix("listener.wss", Conf)])
+                                               ++ cuttlefish_variable:filter_by_prefix("listener.wss", Conf)
+                                               ++ cuttlefish_variable:filter_by_prefix("listener.api", Conf)])
 end}.
 
+%%--------------------------------------------------------------------
+%% MQTT REST API Listeners
+
+{mapping, "listener.api.$name", "emqttd.listeners", [
+  {datatype, [integer, ip]}
+]}.
+
+{mapping, "listener.api.$name.acceptors", "emqttd.listeners", [
+  {default, 8},
+  {datatype, integer}
+]}.
+
+{mapping, "listener.api.$name.max_clients", "emqttd.listeners", [
+  {default, 1024},
+  {datatype, integer}
+]}.
+
+{mapping, "listener.api.$name.rate_limit", "emqttd.listeners", [
+  {datatype, string}
+]}.
+
+{mapping, "listener.api.$name.access.$id", "emqttd.listeners", [
+  {datatype, string}
+]}.
+
+{mapping, "listener.api.$name.backlog", "emqttd.listeners", [
+  {default, 1024},
+  {datatype, integer}
+]}.
+
+{mapping, "listener.api.$name.recbuf", "emqttd.listeners", [
+  {datatype, bytesize},
+  hidden
+]}.
+
+{mapping, "listener.api.$name.sndbuf", "emqttd.listeners", [
+  {datatype, bytesize},
+  hidden
+]}.
+
+{mapping, "listener.api.$name.buffer", "emqttd.listeners", [
+  {datatype, bytesize},
+  hidden
+]}.
+
+{mapping, "listener.api.$name.tune_buffer", "emqttd.listeners", [
+  {datatype, flag},
+  hidden
+]}.
+
+{mapping, "listener.api.$name.nodelay", "emqttd.listeners", [
+  {datatype, {enum, [true, false]}},
+  hidden
+]}.
+
+{mapping, "listener.api.$name.handshake_timeout", "emqttd.listeners", [
+  {datatype, {duration, ms}}
+]}.
+
+{mapping, "listener.api.$name.keyfile", "emqttd.listeners", [
+  {datatype, string}
+]}.
+
+{mapping, "listener.api.$name.certfile", "emqttd.listeners", [
+  {datatype, string}
+]}.
+
+{mapping, "listener.api.$name.cacertfile", "emqttd.listeners", [
+  {datatype, string}
+]}.
+
+{mapping, "listener.api.$name.verify", "emqttd.listeners", [
+  {datatype, atom}
+]}.
+
+{mapping, "listener.api.$name.fail_if_no_peer_cert", "emqttd.listeners", [
+  {datatype, {enum, [true, false]}}
+]}.
+
 %%--------------------------------------------------------------------
 %% System Monitor
 %%--------------------------------------------------------------------

src/emqttd_app.erl

@@ -165,11 +165,14 @@ start_listener({ssl, ListenOn, Opts}) ->
 
 %% Start http listener
 start_listener({Proto, ListenOn, Opts}) when Proto == http; Proto == ws ->
-    mochiweb:start_http('mqtt:ws', ListenOn, Opts, {emqttd_http, handle_request, []});
+    mochiweb:start_http('mqtt:ws', ListenOn, Opts, {emqttd_ws, handle_request, []});
 
 %% Start https listener
 start_listener({Proto, ListenOn, Opts}) when Proto == https; Proto == wss ->
-    mochiweb:start_http('mqtt:wss', ListenOn, Opts, {emqttd_http, handle_request, []}).
+    mochiweb:start_http('mqtt:wss', ListenOn, Opts, {emqttd_ws, handle_request, []});
+
+start_listener({Proto, ListenOn, Opts}) when Proto == api ->
+    mochiweb:start_http('mqtt:api', ListenOn, Opts, {emqttd_http, handle_request, []}).
 
 start_listener(Proto, ListenOn, Opts) ->
     Env = lists:append(emqttd:env(client, []), emqttd:env(protocol, [])),

src/emqttd_bridge.erl

@@ -37,7 +37,6 @@
 
 -record(state, {pool, id,
                 node, subtopic,
-                qos                = ?QOS_0,
                 topic_suffix       = <<>>,
                 topic_prefix       = <<>>,
                 mqueue             :: emqttd_mqueue:mqueue(),
@@ -45,8 +44,7 @@
                 ping_down_interval = ?PING_DOWN_INTERVAL,
                 status             = up}).
 
--type(option() :: {qos, mqtt_qos()} |
-                  {topic_suffix, binary()} |
+-type(option() :: {topic_suffix, binary()} |
                   {topic_prefix, binary()} |
                   {max_queue_len, pos_integer()} |
                   {ping_down_interval, pos_integer()}).
@@ -87,8 +85,6 @@ init([Pool, Id, Node, Topic, Options]) ->
 
 parse_opts([], State) ->
     State;
-parse_opts([{qos, Qos} | Opts], State) ->
-    parse_opts(Opts, State#state{qos = Qos});
 parse_opts([{topic_suffix, Suffix} | Opts], State) ->
     parse_opts(Opts, State#state{topic_suffix= Suffix});
 parse_opts([{topic_prefix, Prefix} | Opts], State) ->

src/emqttd_cli.erl

@@ -326,7 +326,6 @@ bridges(["list"]) ->
 
 bridges(["options"]) ->
     ?PRINT_MSG("Options:~n"),
-    ?PRINT_MSG("  qos     = 0 | 1 | 2~n"),
     ?PRINT_MSG("  prefix  = string~n"),
     ?PRINT_MSG("  suffix  = string~n"),
     ?PRINT_MSG("  queue   = integer~n"),
@@ -363,8 +362,6 @@ parse_opts(Cmd, OptStr) ->
     Tokens = string:tokens(OptStr, ","),
     [parse_opt(Cmd, list_to_atom(Opt), Val)
         || [Opt, Val] <- [string:tokens(S, "=") || S <- Tokens]].
-parse_opt(bridge, qos, Qos) ->
-    {qos, list_to_integer(Qos)};
 parse_opt(bridge, suffix, Suffix) ->
     {topic_suffix, bin(Suffix)};
 parse_opt(bridge, prefix, Prefix) ->

src/emqttd_http.erl

@@ -52,25 +52,6 @@ handle_request('POST', "/mqtt/publish", Req) ->
         false -> Req:respond({401, [], <<"Unauthorized">>})
     end;
 
-%%--------------------------------------------------------------------
-%% MQTT Over WebSocket
-%%--------------------------------------------------------------------
-
-handle_request('GET', "/mqtt", Req) ->
-    lager:info("WebSocket Connection from: ~s", [Req:get(peer)]),
-    Upgrade = Req:get_header_value("Upgrade"),
-    Proto   = check_protocol_header(Req),
-    case {is_websocket(Upgrade), Proto} of
-        {true, "mqtt" ++ _Vsn} ->
-            emqttd_ws:handle_request(Req);
-        {false, _} ->
-            lager:error("Not WebSocket: Upgrade = ~s", [Upgrade]),
-            Req:respond({400, [], <<"Bad Request">>});
-        {_, Proto} ->
-            lager:error("WebSocket with error Protocol: ~s", [Proto]),
-            Req:respond({400, [], <<"Bad WebSocket Protocol">>})
-    end;
-
 %%--------------------------------------------------------------------
 %% Get static files
 %%--------------------------------------------------------------------
@@ -83,18 +64,6 @@ handle_request(Method, Path, Req) ->
     lager:error("Unexpected HTTP Request: ~s ~s", [Method, Path]),
     Req:not_found().
 
-check_protocol_header(Req) ->
-    case emqttd:env(websocket_protocol_header, false) of
-        true  -> get_protocol_header(Req);
-        false -> "mqtt-v3.1.1"
-    end.
-
-get_protocol_header(Req) ->
-    case Req:get_header_value("EMQ-WebSocket-Protocol") of
-        undefined -> Req:get_header_value("Sec-WebSocket-Protocol");
-        Proto     -> Proto
-    end.
-
 %%--------------------------------------------------------------------
 %% HTTP Publish
 %%--------------------------------------------------------------------
@@ -174,9 +143,6 @@ bool("1") -> true;
 bool(<<"0">>) -> false;
 bool(<<"1">>) -> true.
 
-is_websocket(Upgrade) ->
-    Upgrade =/= undefined andalso string:to_lower(Upgrade) =:= "websocket".
-
 docroot() ->
     {file, Here} = code:is_loaded(?MODULE),
     Dir = filename:dirname(filename:dirname(Here)),

src/emqttd_ws.erl

@@ -31,20 +31,53 @@
               lager:Level("WsClient(~s): " ++ Format,
                           [esockd_net:format(State#wsocket_state.peername) | Args])).
 
+
+handle_request(Req) ->
+    handle_request(Req:get(method), Req:get(path), Req).
+
 %%--------------------------------------------------------------------
-%% Handle WebSocket Request
+%% MQTT Over WebSocket
 %%--------------------------------------------------------------------
+handle_request('GET', "/mqtt", Req) ->
+    lager:info("WebSocket Connection from: ~s", [Req:get(peer)]),
+    Upgrade = Req:get_header_value("Upgrade"),
+    Proto   = check_protocol_header(Req),
+    case {is_websocket(Upgrade), Proto} of
+        {true, "mqtt" ++ _Vsn} ->
+            {ok, ProtoEnv} = emqttd:env(protocol),
+            PacketSize = get_value(max_packet_size, ProtoEnv, ?MAX_PACKET_SIZE),
+            Parser = emqttd_parser:initial_state(PacketSize),
+            %% Upgrade WebSocket.
+            {ReentryWs, ReplyChannel} = mochiweb_websocket:upgrade_connection(Req, fun ?MODULE:ws_loop/3),
+            {ok, ClientPid} = emqttd_ws_client_sup:start_client(self(), Req, ReplyChannel),
+            ReentryWs(#wsocket_state{peername = Req:get(peername), parser = Parser,
+                                     max_packet_size = PacketSize, client_pid = ClientPid});
+        {false, _} ->
+            lager:error("Not WebSocket: Upgrade = ~s", [Upgrade]),
+            Req:respond({400, [], <<"Bad Request">>});
+        {_, Proto} ->
+            lager:error("WebSocket with error Protocol: ~s", [Proto]),
+            Req:respond({400, [], <<"Bad WebSocket Protocol">>})
+    end;
 
-%% @doc Handle WebSocket Request.
-handle_request(Req) ->
-    {ok, ProtoEnv} = emqttd:env(protocol),
-    PacketSize = get_value(max_packet_size, ProtoEnv, ?MAX_PACKET_SIZE),
-    Parser = emqttd_parser:initial_state(PacketSize),
-    %% Upgrade WebSocket.
-    {ReentryWs, ReplyChannel} = mochiweb_websocket:upgrade_connection(Req, fun ?MODULE:ws_loop/3),
-    {ok, ClientPid} = emqttd_ws_client_sup:start_client(self(), Req, ReplyChannel),
-    ReentryWs(#wsocket_state{peername = Req:get(peername), parser = Parser,
-                             max_packet_size = PacketSize, client_pid = ClientPid}).
+handle_request(Method, Path, Req) ->
+    lager:error("Unexpected WS Request: ~s ~s", [Method, Path]),
+    Req:not_found().
+
+is_websocket(Upgrade) ->
+    Upgrade =/= undefined andalso string:to_lower(Upgrade) =:= "websocket".
+
+check_protocol_header(Req) ->
+    case emqttd:env(websocket_protocol_header, false) of
+        true  -> get_protocol_header(Req);
+        false -> "mqtt-v3.1.1"
+    end.
+
+get_protocol_header(Req) ->
+    case Req:get_header_value("EMQ-WebSocket-Protocol") of
+        undefined -> Req:get_header_value("Sec-WebSocket-Protocol");
+        Proto     -> Proto
+    end.
 
 %%--------------------------------------------------------------------
 %% Receive Loop