7 changed files with 80 additions and 33 deletions
--- a/apps/efka/src/efka.app.src
+++ b/apps/efka/src/efka.app.src
@ -5,13 +5,15 @@
  {mod, {efka_app, []}},
  {applications,
   [
-    % sync,
+    sync,
    hackney,
    lager,
    esockd,
-    %jiffy,
-    %gpb,
-    %mnesia,
+    jiffy,
+    gpb,
+    cowlib,
+    gun,
+    mnesia,
    crypto,
    ssl,
    public_key,
--- a/apps/efka/src/efka_agent.erl
+++ b/apps/efka/src/efka_agent.erl
@ -4,7 +4,7 @@
 %%% @doc
 %%%
 %%% @end
-%%% Created : 20. 4月 2025 18:47
+%%% Created : 20. 4月 2025 16:25
 %%%-------------------------------------------------------------------
 -module(efka_agent).
 -author("anlicheng").
@ -17,12 +17,17 @@
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

-define(SERVER, ?MODULE). 
+-define(SERVER, ?MODULE).
+
+-define(DISCONNECTED, 0).
+-define(CONNECTING, 1).
+-define(CONNECTED, 2).

 -record(state, {
+    conn_pid :: pid() | undefined,
    host :: string(),
    port :: integer(),
-    socket
+    status = ?DISCONNECTED
 }).

 %%%===================================================================
@ -45,10 +50,11 @@ start_link() ->
    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
    {stop, Reason :: term()} | ignore).
 init([]) ->
-    {ok, Props} = application:get_env(efka, tls_server),
+    {ok, Props} = application:get_env(efka, wss_server),
    Host = proplists:get_value(host, Props),
    Port = proplists:get_value(port, Props),

+    %% 异步建立到服务器的链接
    erlang:start_timer(0, self(), create_connection),

    {ok, #state{host = Host, port = Port}}.
@ -56,7 +62,7 @@ init([]) ->
 %% @private
 %% @doc Handling call messages
 -spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
-        State :: #state{}) ->
+    State :: #state{}) ->
    {reply, Reply :: term(), NewState :: #state{}} |
    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
    {noreply, NewState :: #state{}} |
@ -83,27 +89,38 @@ handle_cast(_Request, State = #state{}) ->
    {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({timeout, _, create_connection}, State = #state{host = Host, port = Port}) ->
    case connect(Host, Port) of
-        {ok, Socket} ->
-            {noreply, State#state{socket = Socket}};
+        {ok, ConnPid} ->
+            {noreply, State#state{conn_pid = ConnPid, status = ?CONNECTING}};
        {error, Reason} ->
-            lager:debug("[efka_agent] create_connection get error: ~p", [Reason]),
+            lager:debug("[efka_agent] connect host: ~p, port: ~p, get error: ~p", [Host, Port, Reason]),
            retry_connect(),
-            {noreply, State#state{socket = undefined}}
+            {noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}}
    end;

-handle_info({ssl, Socket, Data}, State = #state{socket = Socket}) ->
-    lager:debug("[efka_agent] socket get message: ~p", [Data]),
+%% upgrade逻辑处理
+handle_info({gun_upgrade, ConnPid, _StreamRef, [<<"websocket">>], Headers}, State = #state{conn_pid = ConnPid}) ->
+    lager:debug("[efka_agent] upgrade success, headers: ~p", [Headers]),
+    {noreply, State#state{status = ?CONNECTED}};
+handle_info({gun_response, ConnPid, _, _, Status, Headers}, State = #state{conn_pid = ConnPid}) ->
+    lager:debug("[efka_agent] upgrade failed, status: ~p, headers: ~p", [Status, Headers]),
    retry_connect(),
-    {noreply, State#state{socket = undefined}};
+    gun:close(ConnPid),
+    {noreply, State#state{conn_pid = undefined, status = ?DISCONNECTED}};

-handle_info({ssl_error, Socket, Reason}, State = #state{socket = Socket}) ->
-    lager:warning("[efka_agent] socket close with reason: ~p", [Reason]),
+%% 处理接受到的消息
+handle_info({gun_ws, ConnPid, {binary, Bin}}, State = #state{conn_pid = ConnPid}) ->
+    lager:debug("[efka_agent] get binary message: ~p", [Bin]),
+    {noreply, State};
+
+handle_info({gun_ws, ConnPid, {text, Msg}}, State = #state{conn_pid = ConnPid}) ->
+    lager:debug("[efka_agent] get text message: ~p", [Msg]),
+    {noreply, State};
+
+%% websocket链接关闭
+handle_info({gun_ws, ConnPid, close}, State = #state{conn_pid = ConnPid}) ->
+    lager:debug("[efka_agent] ws close"),
    retry_connect(),
-    {noreply, State#state{socket = undefined}};
-handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
-    lager:warning("[efka_agent] socket closed"),
-    retry_connect(),
-    {noreply, State#state{socket = undefined}};
+    {noreply, State#state{conn_pid = undefined}};

 handle_info(_Info, State = #state{}) ->
    {noreply, State}.
@ -114,14 +131,14 @@ handle_info(_Info, State = #state{}) ->
 %% necessary cleaning up. When it returns, the gen_server terminates
 %% with Reason. The return value is ignored.
 -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
-        State :: #state{}) -> term()).
+    State :: #state{}) -> term()).
 terminate(_Reason, _State = #state{}) ->
    ok.

 %% @private
 %% @doc Convert process state when code is changed
 -spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
-        Extra :: term()) ->
+    Extra :: term()) ->
    {ok, NewState :: #state{}} | {error, Reason :: term()}).
 code_change(_OldVsn, State = #state{}, _Extra) ->
    {ok, State}.
@ -133,9 +150,28 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 retry_connect() ->
    erlang:start_timer(5000, self(), create_connection).

+-spec connect(Host :: string(), Port :: integer()) -> {ok, ConnPid :: pid()} | {error, Reason :: any()}.
 connect(Host, Port) when is_list(Host), is_integer(Port) ->
-    SslOptions = [
-        {verify, verify_none},
-        {active, true}
-    ],
-    ssl:connect(Host, Port, SslOptions, 5000).
+    Opts = #{
+        protocols => [http],
+        transport => tcp
+    },
+    case gun:open(Host, Port, Opts) of
+        {ok, ConnPid} ->
+            %% 等待连接建立
+            receive
+                {gun_up, ConnPid, http} ->
+                    %% 发起 WebSocket 握手请求
+                    gun:ws_upgrade(ConnPid, "/", [
+                        {<<"connection">>, <<"upgrade">>},
+                        {<<"upgrade">>, <<"websocket">>},
+                        {<<"sec-websocket-version">>, <<"13">>},
+                        {<<"sec-websocket-key">>, base64:encode(crypto:strong_rand_bytes(16))}
+                    ]),
+                    {ok, ConnPid}
+            after 5000 ->
+                {error, connect_timeout}
+            end;
+        {error, Reason} ->
+            {error, Reason}
+    end.
--- a/apps/efka/src/efka_app.erl
+++ b/apps/efka/src/efka_app.erl
@ -12,7 +12,7 @@
 start(_StartType, _StartArgs) ->
    io:setopts([{encoding, unicode}]),
    %% 启动mnesia数据库
-    % mnesia:start(),
+    mnesia:start(),
    %% 加速内存的回收
    erlang:system_flag(fullsweep_after, 16),

--- a/config/sys.config
+++ b/config/sys.config
@ -2,7 +2,7 @@
  {efka, [
    {root_dir, "/tmp/efka/"},

-    {tls_server, [
+    {wss_server, [
      {host, "localhost"},
      {port, 18080}
    ]}
--- a/config/vm.args
+++ b/config/vm.args
@ -3,4 +3,4 @@
 -setcookie efka_cookie

 +K true
-+A 5
+A30
--- a/rebar.config
+++ b/rebar.config
@ -5,6 +5,7 @@
    {esockd, ".*", {git, "https://github.com/emqx/esockd.git", {tag, "v5.8.0"}}},
    {jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.2"}}},
    {gpb, ".*", {git, "https://github.com/tomas-abrahamsson/gpb.git", {tag, "4.20.0"}}},
+    {gun, ".*", {git, "https://github.com/ninenines/gun.git", {tag, "2.2.0"}}},
    {parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}},
    {lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}}
 ]}.
--- a/rebar.lock
+++ b/rebar.lock
@ -1,5 +1,9 @@
 {"1.2.0",
 [{<<"certifi">>,{pkg,<<"certifi">>,<<"2.5.3">>},1},
+ {<<"cowlib">>,
+  {git,"https://github.com/ninenines/cowlib",
+       {ref,"f8d0ad7f19b5dddd33cbfb089ebd2e2be2a81a5d"}},
+  1},
 {<<"esockd">>,
  {git,"https://github.com/emqx/esockd.git",
       {ref,"9b959fc11a1c398a589892f335235be6c5b4a454"}},
@ -10,6 +14,10 @@
  {git,"https://github.com/tomas-abrahamsson/gpb.git",
       {ref,"edda1006d863a09509673778c455d33d88e6edbc"}},
  0},
+ {<<"gun">>,
+  {git,"https://github.com/ninenines/gun.git",
+       {ref,"627b8f9ed65da255afaddd166b1b9d102e0fa512"}},
+  0},
 {<<"hackney">>,
  {git,"https://github.com/benoitc/hackney.git",
       {ref,"f642ee0eccaff9aa15707ae3a3effa29f2920e41"}},