fix channel

apps/efka/src/docker/efka_container.erl

@@ -69,9 +69,9 @@ metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(Devic
 send_event(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_binary(Params) ->
     gen_server:cast(Pid, {send_event, EventType, Params}).
 
--spec attach_channel(Pid :: pid(), ChannelPid :: pid(), Meta :: binary()) -> ok | {error, Reason :: binary()}.
+-spec attach_channel(Pid :: pid(), ChannelPid :: pid(), MetaTag :: binary()) -> ok | {error, Reason :: binary()}.
-attach_channel(Pid, ChannelPid, Meta) when is_pid(Pid), is_pid(ChannelPid), is_binary(Meta) ->
+attach_channel(Pid, ChannelPid, MetaTag) when is_pid(Pid), is_pid(ChannelPid), is_binary(MetaTag) ->
-    gen_server:call(Pid, {attach_channel, ChannelPid, Meta}).
+    gen_server:call(Pid, {attach_channel, ChannelPid, MetaTag}).
 
 %% @doc Spawns the server and registers the local name (unique)
 -spec(start_link(Name :: atom(), ContainerId :: binary()) ->

apps/efka/src/websocket/ws_channel.erl

@@ -13,25 +13,20 @@
 %% API
 -export([init/2]).
 -export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
--export([push_config/4, invoke/4, channel_reply/3, register/2]).
+-export([invoke/4, channel_reply/3]).
 
 %% 最大的等待时间
 -define(PENDING_TIMEOUT, 10 * 1000).
 
 -record(state, {
     packet_id = 1,
-    service_id :: undefined | binary(),
+    container_pid :: undefined | pid(),
-    service_pid :: undefined | pid(),
     is_registered = false :: boolean(),
 
     %% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}; 自身的inflight需要超时逻辑处理
     inflight = #{}
 }).
 
--spec push_config(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), ConfigJson :: binary()) -> no_return().
-push_config(ChannelPid, Ref, ReceiverPid, ConfigJson) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(ConfigJson), is_reference(Ref) ->
-    ChannelPid ! {push_config, Ref, ReceiverPid, ConfigJson}.
-
 -spec invoke(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), Payload :: binary()) -> no_return().
 invoke(ChannelPid, Ref, ReceiverPid, Payload) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(Payload), is_reference(Ref) ->
     ChannelPid ! {invoke, Ref, ReceiverPid, Payload}.
@@ -51,27 +46,6 @@ channel_reply(Id, Reply, Inflight) ->
             NInflight
     end.
 
-%% 注册
--spec register(Id :: integer(), ServiceId :: binary()) -> {error, Reply :: binary()} | {ok, Reply :: binary(), ServicePid :: pid()}.
-register(Id, ServiceId) when is_integer(Id), is_binary(ServiceId) ->
-    case efka_service:get_pid(ServiceId) of
-        undefined ->
-            lager:warning("[ws_channel] service_id: ~p, not running", [ServiceId]),
-            Reply = json_error(Id, -1, <<"service not running">>),
-            {error, Reply};
-        ServicePid when is_pid(ServicePid) ->
-            case efka_service:attach_channel(ServicePid, self()) of
-                ok ->
-                    Reply = json_result(Id, <<"ok">>),
-                    erlang:monitor(process, ServicePid),
-                    {ok, Reply, ServicePid};
-                {error, Error} ->
-                    lager:warning("[ws_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
-                    Reply = json_error(Id, -1, Error),
-                    {error, Reply}
-            end
-    end.
-
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% 逻辑处理方法
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
@@ -103,14 +77,6 @@ websocket_handle(Info, State) ->
     lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
     {stop, State}.
 
-websocket_info({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
-    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson}},
-    Packet = jiffy:encode(PushConfig, [force_utf8]),
-    Reply = <<?PACKET_PUSH:8, Packet/binary>>,
-    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
-
-    {reply, {binary, Reply}, State#state{packet_id = ws_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
-
 %% 远程调用
 websocket_info({invoke, Ref, ReceiverPid, Payload}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
     PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}},
@@ -132,9 +98,9 @@ websocket_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = In
     {ok, State#state{inflight = NInflight}};
 
 %% service进程关闭
-websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State = #state{service_pid = ServicePid}) ->
+websocket_info({'DOWN', _Ref, process, ContainerPid, Reason}, State = #state{container_pid = ContainerPid}) ->
-    lager:debug("[tcp_channel] service_pid: ~p, exited: ~p", [ServicePid, Reason]),
+    lager:debug("[tcp_channel] container_pid: ~p, exited: ~p", [ContainerPid, Reason]),
-    {stop, State#state{service_pid = undefined}};
+    {stop, State#state{container_pid = undefined}};
 
 websocket_info({timeout, _, {stop, Reason}}, State) ->
     lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
@@ -158,32 +124,38 @@ terminate(Reason, _Req, State) ->
 %% helper methods
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
-%% 注册
+%% 注册, 要建立程序和容器之间的关系
-handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State) ->
+handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"container_name">> := ContainerName, <<"meta_tag">> := MetaTag}}, State) ->
-    case ws_channel:register(Id, ServiceId) of
+    case efka_container:get_pid(ContainerName) of
-        {error, Reply} ->
+        undefined ->
+            lager:warning("[ws_channel] container_name: ~p, not running", [ContainerName]),
+            Reply = json_error(Id, -1, <<"container not running">>),
             delay_stop(10, normal),
             {reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State};
-        {ok, Reply, ServicePid} ->
+        ContainerPid when is_pid(ContainerPid) ->
+            case efka_container:attach_channel(ContainerPid, self(), MetaTag) of
+                ok ->
+                    Reply = json_result(Id, <<"ok">>),
+                    erlang:monitor(process, ContainerPid),
+
+                    {reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{container_pid = ContainerPid, is_registered = true}};
+                {error, Error} ->
+                    lager:warning("[ws_channel] container_name: ~p, attach_channel get error: ~p", [ContainerName, Error]),
+                    Reply = json_error(Id, -1, Error),
                     delay_stop(10, normal),
-            {reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}
+                    {reply, {binary,  <<?PACKET_RESPONSE:8, Reply/binary>>}, State}
+            end
     end;
-%% 请求参数
-handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{service_pid = ServicePid, is_registered = true}) ->
-    {ok, ConfigJson} = efka_service:request_config(ServicePid),
-    Packet = ws_channel:json_result(Id, ConfigJson),
-    Reply = <<?PACKET_RESPONSE:8, Packet/binary>>,
-    {reply, {binary, Reply}, State};
 
 %% 数据项
 handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>,
-    <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+    <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{container_pid = ContainerPid, is_registered = true}) ->
-    efka_service:metric_data(ServicePid, DeviceUUID, RouteKey, Metric),
+    efka_container:metric_data(ContainerPid, DeviceUUID, RouteKey, Metric),
     {ok, State};
 
 %% Event事件
-handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{container_pid = ContainerPid, is_registered = true}) ->
-    efka_service:send_event(ServicePid, EventType, Body),
+    efka_container:send_event(ContainerPid, EventType, Body),
     {ok, State};
 
 %% 订阅事件