fix

src/host/iot_host.erl

@@ -10,7 +10,6 @@
 -author("aresei").
 -include("iot.hrl").
 -include("domain_model.hrl").
--include("message_pb.hrl").
 
 -behaviour(gen_statem).
 
@@ -130,10 +129,10 @@ remove_container(Pid, ContainerName) when is_pid(Pid), is_binary(ContainerName)
     {ok, Result :: binary()} | {error, Code :: integer(), Reason :: binary()}.
 await_reply(Pid, Ref, Timeout) when is_pid(Pid), is_reference(Ref), is_integer(Timeout) ->
     receive
-        {request_reply, Ref, #'ReplyFrame'{reply = {result, ResultBin}}} ->
-            {ok, iolist_to_binary(ResultBin)};
-        {request_reply, Ref, #'ReplyFrame'{reply = {error, #'ReplyFrame.Error'{code = Code, message = Message}}}} ->
-            {error, Code, unicode:characters_to_binary(Message)}
+        {request_reply, Ref, {ok, ResultBin}} when is_binary(ResultBin) ->
+            {ok, ResultBin};
+        {request_reply, Ref, {error, Code, Reason}} when is_integer(Code), is_binary(Reason) ->
+            {error, Code, Reason}
     after Timeout ->
         ok = gen_statem:call(Pid, {cancel_request_call, Ref}),
         flush_reply(Ref),
@@ -248,10 +247,7 @@ handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, chan
     case is_pid(ChannelPid) of
         true ->
             logger:debug("[iot_host] uuid: ~p, activate: true", [UUID]),
-            Command = #'CastFrame.Command'{
-                command = {auth, #'CastFrame.Command.Authorization'{cmd = 'ACTIVATE'}}
-            },
-            ssl_channel:command(ChannelPid, Command);
+            ssl_channel:activate(ChannelPid, true);
         false ->
             logger:debug("[iot_host] uuid: ~p, activate: true, no channel", [UUID])
     end,
@@ -261,10 +257,7 @@ handle_event({call, From}, {activate, true}, _, State = #state{uuid = UUID, chan
 handle_event({call, From}, {activate, false}, _, State = #state{uuid = UUID, channel_pid = ChannelPid}) ->
     case is_pid(ChannelPid) of
         true ->
-            Command = #'CastFrame.Command' {
-                command = {auth, #'CastFrame.Command.Authorization'{cmd = 'DEACTIVATE'}}
-            },
-            ssl_channel:command(ChannelPid, Command),
+            ssl_channel:activate(ChannelPid, false),
             logger:debug("[iot_host] uuid: ~p, activate: false", [UUID]),
             ssl_channel:stop(ChannelPid, closed);
         false ->
@@ -293,10 +286,8 @@ handle_event({call, From}, {attach_channel, _}, _, State = #state{uuid = UUID, c
     {keep_state, State, [{reply, From, {error, <<"channel existed">>}}]};
 
 %% 数据分发
-handle_event(cast, {handle, {data, #'CastFrame.Data'{route_key = RouteKey0, metric = Metric}}}, ?STATE_ACTIVATED,
+handle_event(cast, {handle, {data, RouteKey, MetricBin}}, ?STATE_ACTIVATED,
     State = #state{uuid = UUID, has_session = true}) ->
-    RouteKey = iolist_to_binary(RouteKey0),
-    MetricBin = iolist_to_binary(Metric),
     logger:debug("[iot_host] metric_data host: ~p, route_key: ~p, metric: ~p", [UUID, RouteKey, MetricBin]),
     endpoint_subscription:publish(get_route_key(RouteKey), MetricBin),
     {keep_state, State};
@@ -365,9 +356,9 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec container_call(Pid :: pid(), Request :: message_pb:'ContainerRequest'()) ->
+-spec container_call(Pid :: pid(), Request :: term()) ->
     {ok, Ref :: reference()} | {error, Reason :: any()}.
-container_call(Pid, Request) when is_pid(Pid), is_record(Request, 'ContainerRequest') ->
+container_call(Pid, Request) when is_pid(Pid) ->
     gen_statem:call(Pid, {container_call, self(), Request}).
 
 -spec get_route_key(binary()) -> binary().

src/transport/tcp/ssl_channel.erl

@@ -16,7 +16,7 @@
 -define(INFLIGHT_TIMEOUT, 60000).
 
 %% API
--export([pub/4, container_call/3, cancel_request_call/2, command/2]).
+-export([pub/4, container_call/3, cancel_request_call/2, command/2, activate/2]).
 
 -export([start_link/3, stop/2]).
 %% gen_server callbacks
@@ -52,6 +52,14 @@ pub(Pid, Topic, Qos, Content) when is_pid(Pid), is_binary(Topic), is_integer(Qos
 command(Pid, Command) when is_pid(Pid) ->
     gen_server:cast(Pid, {command, Command}).
 
+-spec activate(Pid :: pid(), Auth :: boolean()) -> no_return().
+activate(Pid, Auth) when is_pid(Pid), is_boolean(Auth) ->
+    Cmd = case Auth of true -> 'ACTIVATE'; false -> 'DEACTIVATE' end,
+    Command =  #'CastFrame.Command'{
+        command = {auth, #'CastFrame.Command.Authorization'{cmd = Cmd}}
+    },
+    command(Pid, Command).
+
 -spec container_call(Pid :: pid(), ReceiverPid :: pid(), Request :: message_pb:'ContainerRequest'()) -> Ref :: reference().
 container_call(Pid, ReceiverPid, Request) when is_pid(Pid), is_pid(ReceiverPid), is_record(Request, 'ContainerRequest') ->
     Ref = make_ref(),
@@ -255,7 +263,7 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _T
 -spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) ->
     {noreply, #state{}}.
 handle_cast_frame(#'CastFrame'{body = {data, Data}}, HostPid, State) when is_pid(HostPid) ->
-    iot_host:handle(HostPid, {data, Data}),
+    iot_host:handle(HostPid, decode_data_frame(Data)),
     {noreply, State};
 handle_cast_frame(#'CastFrame'{body = {task_event, CastMessage}}, HostPid, State) when is_pid(HostPid) ->
     handle_event_stream_frame(CastMessage),
@@ -282,7 +290,7 @@ handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, Inflight,
             erlang:cancel_timer(TimerRef),
             case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
                 true ->
-                    ReceiverPid ! {request_reply, Ref, #'ReplyFrame'{packet_id = PacketId, reply = Reply}};
+                    ReceiverPid ! {request_reply, Ref, decode_reply(Reply)};
                 false ->
                     logger:warning("[ws_channel] get reply message: ~p, packet_id: ~p, but receiver_pid is deaded", [Reply, PacketId])
             end,
@@ -299,3 +307,16 @@ handle_reply_frame(#'ReplyFrame'{packet_id = PacketId, reply = Reply}, _Inflight
 send_reply_frame(Transport, Socket, PacketId, Reply) ->
     Encoded = message_pb:encode_msg(#'ReplyFrame'{packet_id = PacketId, reply = Reply}),
     Transport:send(Socket, <<?FRAME_REPLY, Encoded/binary>>).
+
+-spec decode_data_frame(message_pb:'CastFrame.Data'()) -> {data, binary(), binary()}.
+decode_data_frame(#'CastFrame.Data'{route_key = RouteKey0, metric = Metric0}) ->
+    {data, iolist_to_binary(RouteKey0), iolist_to_binary(Metric0)}.
+
+-spec decode_reply({result, iodata()} | {error, message_pb:'ReplyFrame.Error'()} | undefined) ->
+    {ok, binary()} | {error, integer(), binary()} | undefined.
+decode_reply({result, ResultBin}) ->
+    {ok, iolist_to_binary(ResultBin)};
+decode_reply({error, #'ReplyFrame.Error'{code = Code, message = Message}}) ->
+    {error, Code, unicode:characters_to_binary(Message)};
+decode_reply(undefined) ->
+    undefined.