diff --git a/src/host/iot_host.erl b/src/host/iot_host.erl index b156aba..e7a5d14 100644 --- a/src/host/iot_host.erl +++ b/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(). diff --git a/src/transport/tcp/ssl_channel.erl b/src/transport/tcp/ssl_channel.erl index 6690e43..a0f975d 100644 --- a/src/transport/tcp/ssl_channel.erl +++ b/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, <>). + +-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.