Merge branch 'dev'

CHANGELOG.md

@@ -1,6 +1,11 @@
 eMQTT ChangeLog
 ==================
 
+v0.3.1-beta (2015-01-24)
+------------------------
+
+Feature: HTTP POST API to support 'qos', 'retain' parameters
+
 v0.3.0-alpha (2015-01-18)
 ------------------------
 
@@ -28,6 +33,8 @@ Test: passed org.eclipse.paho.mqtt.testing/interoperability
 
 Test: simple cluster test
 
+Closed Issues: #22, #24, #27, #28, #29, #30, #31, #32, #33, #34, #36, #37, #38, #39, #41, #42, #43
+
 
 v0.2.1-beta (2015-01-08)
 ------------------------

README.md

@@ -1,15 +1,15 @@
 # eMQTT
 
-eMQTT is a scalable, fault-tolerant and extensible mqtt broker written in Erlang/OTP.
+eMQTT is a scalable, fault-tolerant and extensible MQTT V3.1.1 broker written in Erlang/OTP.
 
-eMQTT support MQTT V3.1 Protocol Specification.
+eMQTT support MQTT V3.1/V3.1.1 Protocol Specification.
 
 eMQTT requires Erlang R17+.
 
 ## Startup in Five Minutes
 
 ```
-$ git clone git://github.com/slimpp/emqtt.git
+$ git clone git://github.com/emqtt/emqtt.git
 
 $ cd emqtt
 
@@ -52,7 +52,7 @@ cd $INSTALL_DIR/emqtt
             {max_conns, 1024},
             {acceptor_pool, 4}
         ]},
-        {http, 8883, [
+        {http, 8083, [
             {max_conns, 512},
             {acceptor_pool, 1}
         ]}
@@ -65,7 +65,7 @@ cd $INSTALL_DIR/emqtt
 
 ```
 
--sname emqtt
+-name emqtt@127.0.0.1
 
 -setcookie emqtt
 
@@ -124,21 +124,23 @@ eMQTT support http to publish message.
 Example:
 
 ```
-curl -v --basic -u user:passwd -d "topic=/a/b/c&message=hello from http..." -k http://localhost:8883/mqtt/publish
+curl -v --basic -u user:passwd -d "qos=1&retain=0&topic=/a/b/c&message=hello from http..." -k http://localhost:8083/mqtt/publish
 ```
 
 ### URL
 
 ```
-HTTP POST http://host:8883/mqtt/publish
+HTTP POST http://host:8083/mqtt/publish
 ```
 
 ### Parameters
 
-Name | Description
------|-------------
-topic | MQTT Topic
-message | Text Message
+Name    |  Description
+--------|---------------
+qos     |  QoS(0, 1, 2)
+retain  |  Retain(0, 1)
+topic   |  Topic
+message |  Message
 
 ## Design
 
@@ -155,5 +157,6 @@ feng at emqtt.io
 ## Thanks
 
 @hejin1026 (260495915 at qq.com)
+
 @desoulter (assoulter123 at gmail.com)
 

TODO

@@ -27,10 +27,6 @@ fucking stupid..... esockd locked
 0.2.1
 =====
 
-full MQTT 3.1.1 support...
-
-node cluster....
-
 one million connections test...
 
 topic match benchmark tests...
@@ -41,14 +37,4 @@ full test cases...
 
 spawn_link to replace 'spawn' and 'link'
 
-keepalive
-
-retained
-
-QOS
-
-dural sub
-
-packet dump...
-
 

apps/emqtt/src/emqtt_http.erl

@@ -43,15 +43,25 @@ handle(Req) ->
 
 handle('POST', "/mqtt/publish", Req) ->
     Params = mochiweb_request:parse_post(Req),
-	lager:info("~p~n", [Params]),
-	Topic = list_to_binary(get_value("topic", Params)),
-	Message = list_to_binary(get_value("message", Params)),
-	emqtt_pubsub:publish(#mqtt_message {
-				topic      = Topic,
-				payload    = Message
-	}),
-	Req:ok({"text/plan", "ok"});
-
+	lager:info("HTTP Publish: ~p~n", [Params]),
+    Qos = int(get_value("qos", Params, "0")),
+    Retain = bool(get_value("retain", Params,  "0")),
+    Topic = list_to_binary(get_value("topic", Params)),
+    Message = list_to_binary(get_value("message", Params)),
+    case {validate(qos, Qos), validate(topic, Topic)} of
+        {true, true} ->
+            emqtt_router:route(
+                #mqtt_message { qos     = Qos,
+                                retain  = Retain,
+                                topic   = Topic,
+                                payload = Message }),
+            Req:ok({"text/plan", <<"ok\n">>});
+       {false, _} ->
+            Req:respond({400, [], <<"Bad QoS">>});
+        {_, false} ->
+            Req:respond({400, [], <<"Bad Topic">>})
+    end;
+    
 handle(_Method, _Path, Req) ->
 	Req:not_found().
 
@@ -69,3 +79,14 @@ authorized(Req) ->
 user_passwd(BasicAuth) ->
 	list_to_tuple(binary:split(base64:decode(BasicAuth), <<":">>)). 
 
+validate(qos, Qos) ->
+    (Qos >= ?QOS_0) and (Qos =< ?QOS_2); 
+
+validate(topic, Topic) ->
+    emqtt_topic:validate({publish, Topic}).
+
+int(S) -> list_to_integer(S).
+
+bool("0") -> false;
+bool("1") -> true.
+

apps/emqtt/src/emqtt_router.erl

@@ -65,7 +65,7 @@ start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
 route(Msg) ->
-    lager:info("Route message: ~s", [emqtt_message:dump(Msg)]),
+    lager:info("Route ~s", [emqtt_message:dump(Msg)]),
     % need to retain?
     emqtt_server:retain(Msg),
     % unset flag and pubsub