Merge branch 'dev' into dev-hd

CHANGELOG.md

@@ -5,6 +5,8 @@ emqttd ChangeLog
 0.9.0-alpha (2015-07-xx)
 -------------------------
 
+TODO:......
+
 Session and Queue
 
 Merge emqtt, emqttd apps

src/emqttd_guid.erl

@@ -31,7 +31,7 @@
 %%% 1. Timestamp: erlang:system_time if Erlang >= R18, otherwise os:timestamp
 %%% 2. NodeId:    encode node() to 2 bytes integer
 %%% 3. Pid:       encode pid to 4 bytes integer
-%%% 4. Sequence:  2 bytes sequence no per pid
+%%% 4. Sequence:  2 bytes sequence per pid
 %%%
 %%% @end
 %%%-----------------------------------------------------------------------------
@@ -44,6 +44,10 @@
 
 -type guid() :: <<_:128>>.
 
+%%------------------------------------------------------------------------------
+%% @doc Generate a global unique id.
+%% @end
+%%------------------------------------------------------------------------------
 -spec gen() -> guid().
 gen() ->
     Guid = case get(guid) of

src/emqttd_session.erl

@@ -159,7 +159,7 @@ destroy(Session, ClientId) ->
 %%------------------------------------------------------------------------------
 -spec subscribe(pid(), [{binary(), mqtt_qos()}]) -> {ok, [mqtt_qos()]}.
 subscribe(Session, TopicTable) ->
-    gen_server:call(Session, {subscribe, TopicTable}).
+    gen_server:call(Session, {subscribe, TopicTable}, infinity).
 
 %%------------------------------------------------------------------------------
 %% @doc Publish message
@@ -204,7 +204,7 @@ pubcomp(Session, PktId) ->
 %%------------------------------------------------------------------------------
 -spec unsubscribe(pid(), [binary()]) -> ok.
 unsubscribe(Session, Topics) ->
-    gen_server:call(Session, {unsubscribe, Topics}).
+    gen_server:call(Session, {unsubscribe, Topics}, infinity).
 
 %%%=============================================================================
 %%% gen_server callbacks

src/emqttd_sm.erl

@@ -193,5 +193,5 @@ new_session(CleanSess, ClientId, ClientPid) ->
 
 setstats(State = #state{statsfun = {CFun, SFun}}) ->
     CFun(ets:info(?SESSION_TAB, size)),
-    SFun(ets:select_count(?SESSION_TAB,  [{{true, '_', '_', '_'}, [], [true]}])),
+    SFun(ets:select_count(?SESSION_TAB,  [{{false, '_', '_', '_'}, [], [true]}])),
     State.

src/emqttd_vm.erl

@@ -27,6 +27,27 @@
 
 -module(emqttd_vm).
 
+-export([microsecs/0]).
+
+-export([loads/0, scheduler_usage/1]).
+
+-export([get_memory/0]).
+
+-export([get_process_list/0,
+         get_process_info/0,
+         get_process_gc/0,
+         get_process_group_leader_info/1]).
+	
+-export([get_ets_list/0,
+         get_ets_info/0,
+         get_ets_info/1,
+         get_ets_object/0,
+         get_ets_object/1]).
+
+-export([get_port_types/0,
+         get_port_info/0,
+         get_port_info/1]).
+
 -define(UTIL_ALLOCATORS, [temp_alloc,
                           eheap_alloc,
                           binary_alloc,
@@ -39,41 +60,36 @@
                          ]).
 
 -define(PROCESS_LIST, [initial_call,
-		       reductions,
+                       reductions,
-		       memory, 
+                       memory,
-		       message_queue_len, 
+                       message_queue_len,
-		       current_function]).
+                       current_function]).
 
 -define(PROCESS_INFO, [initial_call,
-		       current_function,
+                       current_function,
-		       registered_name,
+                       registered_name,
-		       status,
+                       status,
-		       message_queue_len,
+                       message_queue_len,
-		       group_leader,
+                       group_leader,
-		       priority,
+                       priority,
-		       trap_exit,
+                       trap_exit,
-		       reductions,
+                       reductions,
-		       binary,
+                       binary,
-		       last_calls,
+                       last_calls,
-		       catchlevel,
+                       catchlevel,
-		       trace,
+                       trace,
-		       suspending,
+                       suspending,
-		       sequential_trace_token,
+                       sequential_trace_token,
-		       error_handler]).
+                       error_handler]).
 
 -define(PROCESS_GC, [memory,
-		     total_heap_size,
+                     total_heap_size,
-		     heap_size,
+                     heap_size,
-		     stack_size,
+                     stack_size,
-		     min_heap_size]).%,
+                     min_heap_size]).
-		     %fullsweep_after]).
+                     %fullsweep_after]).
 
--export([timestamp/0, microsecs/0]).
+-define(SYSTEM_INFO, [allocated_areas,
-
--export([loads/0]).
-
--define(SYSTEM_INFO, [
-                      allocated_areas,
                       allocator,
                       alloc_util_allocators,
                       build_type,
@@ -123,64 +139,30 @@
                       wordsize
                   ]).
 
--define(SOCKET_OPTS, [
+-define(SOCKET_OPTS, [active,
-                      active,
                       broadcast,
-		      buffer,
+                      buffer,
-		      buffer,
-		      buffer,
-		      buffer,
                       delay_send,
                       dontroute,
                       exit_on_close,
                       header,
-		      high_watermark,
+                      high_watermark,
-		      ipv6_v6only,
+                      ipv6_v6only,
                       keepalive,
-		      linger,
+                      linger,
-		      low_watermark,
+                      low_watermark,
-		      mode,
+                      mode,
                       nodelay,
                       packet,
                       packet_size,
-		      priority,
+                      priority,
                       read_packets,
                       recbuf,
                       reuseaddr,
                       send_timeout,
                       send_timeout_close,
                       sndbuf,
-                      tos
+                      tos]).
-                     ]).
-
-
--author("Feng Lee <feng@emqtt.io>").
-
--export([timestamp/0, microsecs/0]).
-
--export([loads/0,
-	scheduler_usage/1]).
-
--export([get_memory/0]).
-
--export([get_process_list/0,
- 	 get_process_info/0,
- 	 get_process_gc/0,
- 	 get_process_group_leader_info/1]).
-	
--export([get_ets_list/0,
- 	 get_ets_info/0,
-	 get_ets_info/1,
- 	 get_ets_object/0,
-	 get_ets_object/1]).
-
--export([get_port_types/0,
-	 get_port_info/0,
- 	 get_port_info/1]).
-
-timestamp() ->
-    {MegaSecs, Secs, _MicroSecs} = os:timestamp(),
-    MegaSecs * 1000000 + Secs.
 
 microsecs() ->
     {Mega, Sec, Micro} = erlang:now(),