fix httpc

apps/efka/src/efka.app.src

@@ -6,13 +6,11 @@
   {applications,
    [
     sync,
-    hackney,
-    lager,
-    esockd,
     %jiffy,
     %gpb,
     %mnesia,
     crypto,
+    inets,
     ssl,
     public_key,
     kernel,

apps/efka/src/efka_agent.erl

@@ -92,7 +92,7 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
             efka_transport:auth_request(TransportPid, 5000),
             {next_state, ?STATE_AUTH, State};
         {error, Reason} ->
-            lager:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
+            efka_logger:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
             efka_transport:stop(TransportPid),
             {next_state, ?STATE_DENIED, State}
     end;
@@ -101,23 +101,23 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
 handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pid = TransportPid}) ->
     case Reply of
         {ok, #auth_reply{code = 1, message = Message, repository_url = RepositoryUrl}} ->
-            lager:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
+            efka_logger:debug("[efka_agent] auth failed, message: ~p, repository_url: ~p", [Message, RepositoryUrl]),
             {next_state, ?STATE_ACTIVATED, State};
 
         %% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
         %% socket的连接状态需要维持
         {ok, #auth_reply{code = -1, message = Message}} ->
-            lager:debug("[efka_agent] auth denied, message: ~p", [Message]),
+            efka_logger:debug("[efka_agent] auth denied, message: ~p", [Message]),
             {next_state, ?STATE_RESTRICTED, State};
 
         %% 其他类型的错误，需要间隔时间重试
         {ok, #auth_reply{code = -2, message = Message}} ->
-            lager:debug("[efka_agent] auth failed, message: ~p", [Message]),
+            efka_logger:debug("[efka_agent] auth failed, message: ~p", [Message]),
             efka_transport:stop(TransportPid),
             {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
 
         {error, Reason} ->
-            lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
+            efka_logger:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
             efka_transport:stop(TransportPid),
             {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
     end;
@@ -143,7 +143,7 @@ handle_event(info, {server_push_message, <<8:8, ActivatePush>>}, StateName, Stat
 %% 收到需要回复的指令
 handle_event(info, {server_push_message, PacketId, <<16:8, Directive>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     #topic_message{topic = Topic, content = Content} = message_pb:decode_msg(Directive, directive),
-    lager:debug("[efka_agent] get directive with packet_id: ~p, to device_uuid: ~p, content: ~p", [PacketId, Topic, Content]),
+    efka_logger:debug("[efka_agent] get directive with packet_id: ~p, to device_uuid: ~p, content: ~p", [PacketId, Topic, Content]),
 
     %% 消息发送到订阅系统
     case PacketId > 0 of
@@ -158,8 +158,8 @@ handle_event(info, {server_push_message, PacketId, <<16:8, Directive>>}, ?STATE_
 
 %% transport进程退出
 handle_event(info, {'EXIT', TransportPid, Reason}, _StateName, State = #state{transport_pid = TransportPid}) ->
-    lager:warning("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
+    efka_logger:warning("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
-    erlang:start_timer(5000, self(), create_transport),
+    erlang:start_timer(500000, self(), create_transport),
     {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
 
 handle_event(_EventType, _EventContent, _StateName, State = #state{}) ->

apps/efka/src/efka_app.erl

@@ -17,7 +17,7 @@ start(_StartType, _StartArgs) ->
     erlang:system_flag(fullsweep_after, 16),
 
     %% 启动tcp的服务
-    start_tcp_server(),
+    % start_tcp_server(),
 
     %% 仓库的基础url
     application:set_env(efka, repository_url, "http://118.178.229.213:3000/anlicheng/ekfa/"),
@@ -46,4 +46,4 @@ start_tcp_server() ->
     ],
     {ok, _} = esockd:open('efka/tcp_server', Port, TransOpts, {tcp_channel, start_link, []}),
 
-    lager:debug("[efka_app] the tcp server start at: ~p", [Port]).
+    efka_loggeefka_logger:debug("[efka_app] the tcp server start at: ~p", [Port]).

apps/efka/src/efka_downloader.erl

@@ -33,7 +33,7 @@ test() ->
     Ref = download(Pid, Url, TargetDir),
     receive
         {download_response, Ref, Info} ->
-            lager:debug("info is: ~p", [Info])
+            efka_logger:debug("info is: ~p", [Info])
     end,
     ok.
 
@@ -82,23 +82,28 @@ handle_call(_Request, _From, State = #state{}) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_cast({download, ReceiverPid, Ref, Url, TargetDir}, State = #state{}) ->
-    SSLOpts = {ssl_options, [
+    SslOpts = [
+        {ssl, [
             % 完全禁用证书验证
             {verify, verify_none}
-    ]},
+        ]}
+    ],
 
     TargetFile = get_filename_from_url(Url),
+    FullFilename = TargetDir ++ TargetFile,
+
     StartTs = os:timestamp(),
-    case hackney:request(get, Url, [], <<>>, [async, {stream_to, self()}, {pool, false}, SSLOpts]) of
+    case httpc:request(get, {Url, []}, SslOpts, [{sync, false}, {stream, self}]) of
-        {ok, ClientRef} ->
+        {ok, RequestId} ->
-            case receive_data(ClientRef, TargetDir, TargetFile) of
+            case receive_data(RequestId, FullFilename) of
                 ok ->
                     EndTs = os:timestamp(),
                     %% 计算操作的时间，单位为毫秒
                     CostMs = timer:now_diff(EndTs, StartTs) div 1000,
-                    lager:debug("download cost: ~p", [CostMs]),
                     ReceiverPid ! {download_response, Ref, {ok, CostMs}};
                 {error, Reason} ->
+                    %% 出错需要删除掉文件
+                    file:delete(FullFilename),
                     ReceiverPid ! {download_response, Ref, {error, Reason}}
             end;
         {error, Reason} ->
@@ -137,37 +142,25 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
-%% 读取请求 ResponseLine
-receive_data(ClientRef, TargetDir, DefaultFile) when is_list(TargetDir), is_list(DefaultFile) ->
-    receive
-        {hackney_response, ClientRef, {status, 200, _Reason}} ->
-            receive_data0(ClientRef, TargetDir, DefaultFile);
-        {hackney_response, ClientRef, {status, StatusCode, _Reason}} ->
-            {error, {http_status, StatusCode}}
-    end.
 %% 处理头部信息, 解析可能的文件名
-receive_data0(ClientRef, TargetDir, DefaultFile) ->
+receive_data(RequestId, FullFilename) ->
     receive
-        {hackney_response, ClientRef, {headers, Headers}} ->
+        {http, {RequestId, stream_start, _Headers}} ->
-            TargetFilename = extra_filename(Headers, DefaultFile),
-            FullFilename = TargetDir ++ TargetFilename,
-            lager:debug("full name: ~p", [FullFilename]),
             {ok, File} = file:open(FullFilename, [write, binary]),
-            receive_data1(ClientRef, File)
+            receive_data1(RequestId, File)
     end.
 %% 接受文件数据
-receive_data1(ClientRef, File) ->
+receive_data1(RequestId, File) ->
     receive
-        {hackney_response, ClientRef, {error, Reason}} ->
+        {http, {RequestId, {error, Reason}}} ->
             ok = file:close(File),
             {error, Reason};
-        {hackney_response, ClientRef, done} ->
+        {http, {RequestId, stream_end, _Headers}} ->
             ok = file:close(File),
-            hackney:close(ClientRef),
             ok;
-        {hackney_response, ClientRef, Data} ->
+        {http, {RequestId, stream, Data}} ->
             file:write(File, Data),
-            receive_data1(ClientRef, File)
+            receive_data1(RequestId, File)
     end.
 
 -spec extra_filename(Headers :: list(), Default :: string()) -> Filename :: string().
@@ -192,4 +185,3 @@ get_filename_from_url(Url) when is_list(Url) ->
     URIMap = uri_string:parse(Url),
     Path = maps:get(path, URIMap),
     filename:basename(Path).
-

apps/efka/src/efka_logger.erl

@@ -0,0 +1,19 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 29. 4月 2025 22:39
+%%%-------------------------------------------------------------------
+-module(efka_logger).
+-author("anlicheng").
+
+%% API
+-export([debug/2, notice/2]).
+
+debug(Format, Args) ->
+    io:format(Format ++ "~n", Args).
+
+notice(Format, Args) ->
+    io:format(Format ++ "~n", Args).

apps/efka/src/efka_subscription.erl

@@ -119,7 +119,7 @@ handle_cast({publish, Topic, Content}, State = #state{subscribers = Subscribers}
         erlang:send(SubscriberPid, {topic_broadcast, Content})
                   end, MatchedSubscribers),
 
-    lager:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
+    efka_logger:debug("[efka_subscription] topic: ~p, content: ~p, match subscribers: ~p", [Topic, Content, MatchedSubscribers]),
     {noreply, State};
 
 %% 需要响应的推送
@@ -133,7 +133,7 @@ handle_cast({publish, PacketId, Topic, Content, CallbackFun}, State = #state{sub
         [] ->
             ok;
         [S = #subscriber{subscriber_pid = SubscriberPid} | _] ->
-            lager:debug("[efka_subscription] topic_sink topic: ~p, content: ~p, match subscriber: ~p", [Topic, Content, S]),
+            efka_logger:debug("[efka_subscription] topic_sink topic: ~p, content: ~p, match subscriber: ~p", [Topic, Content, S]),
             erlang:send(SubscriberPid, {topic_sink, Content})
     end,
 
@@ -156,12 +156,12 @@ handle_cast({response, ShaUUID, Response}, State = #state{callbacks = Callbacks}
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({'DOWN', _Ref, process, SubscriberPid, Reason}, State = #state{subscribers = Subscribers}) ->
-    lager:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
+    efka_logger:debug("[efka_subscription] subscriber: ~p, down with reason: ~p", [SubscriberPid, Reason]),
     NSubscribers = lists:filter(fun(#subscriber{subscriber_pid = Pid0}) -> SubscriberPid /= Pid0 end, Subscribers),
     {noreply, State#state{subscribers = NSubscribers}};
 
 handle_info(Info, State = #state{}) ->
-    lager:debug("[efka_subscription] get unknown info: ~p", [Info]),
+    efka_logger:debug("[efka_subscription] get unknown info: ~p", [Info]),
     {noreply, State}.
 
 %% @private

apps/efka/src/efka_sup.erl

@@ -28,14 +28,14 @@ start_link() ->
 init([]) ->
     SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
     ChildSpecs = [
-        #{
+        %#{
-            id => 'efka_agent',
+        %    id => 'efka_agent',
-            start => {'efka_agent', start_link, []},
+        %    start => {'efka_agent', start_link, []},
-            restart => permanent,
+        %    restart => permanent,
-            shutdown => 2000,
+        %    shutdown => 2000,
-            type => worker,
+        %    type => worker,
-            modules => ['efka_agent']
+        %    modules => ['efka_agent']
-        },
+        %},
 
         #{
             id => 'efka_server_sup',

apps/efka/src/efka_transport.erl

@@ -148,13 +148,13 @@ handle_cast({response, PacketId, Response}, State = #state{socket = Socket}) ->
     {stop, Reason :: term(), NewState :: #state{}}).
 %% 服务器主动推送的数据，有packetId的是要求返回的；为0的表示不需要返回值
 handle_info({ssl, Socket, <<?PACKET_PUBLISH, 0:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
-    lager:debug("[efka_agent] socket get message: ~p", [Msg]),
+    efka_logger:debug("[efka_agent] socket get message: ~p", [Msg]),
     ParentPid ! {server_push_message, Msg},
     {noreply, State};
 
 %% 目前推送的消息包括:  <<CommandType:8, Command/binary>>, <<16:8, Directive/binary>>
 handle_info({ssl, Socket, <<?PACKET_PUBLISH, PacketId:32, Msg/binary>>}, State = #state{socket = Socket, parent_pid = ParentPid}) ->
-    lager:debug("[efka_agent] socket get message: ~p", [Msg]),
+    efka_logger:debug("[efka_agent] socket get message: ~p", [Msg]),
     ParentPid ! {server_push_message, PacketId, Msg},
     {noreply, State#state{}};
 
@@ -165,7 +165,7 @@ handle_info({ssl_closed, Socket}, State = #state{socket = Socket}) ->
     {stop, normal, State};
 
 handle_info(Info, State = #state{}) ->
-    lager:debug("[efka_transport] info: ~p", [Info]),
+    efka_logger:debug("[efka_transport] info: ~p", [Info]),
     {noreply, State}.
 
 %% @private
@@ -176,7 +176,7 @@ handle_info(Info, State = #state{}) ->
 -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
         State :: #state{}) -> term()).
 terminate(Reason, #state{}) ->
-    lager:debug("[efka_transport] terminate with reason: ~p", [Reason]),
+    efka_logger:debug("[efka_transport] terminate with reason: ~p", [Reason]),
     ok.
 
 %% @private

apps/efka/src/tcp_channel.erl

@@ -66,7 +66,7 @@ start_link(Transport, Sock) ->
     {ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Sock]])}.
 
 init([Transport, Sock]) ->
-    lager:debug("[sdlan_channel] get a new connection: ~p", [Sock]),
+    efka_logger:debug("[sdlan_channel] get a new connection: ~p", [Sock]),
     case Transport:wait(Sock) of
         {ok, NewSock} ->
             Transport:setopts(Sock, [{active, true}]),
@@ -85,15 +85,15 @@ handle_cast(_Msg, State) ->
 %% 网络流量统计
 handle_info({tcp, _Sock, <<Body/binary>>}, State = #state{is_registered = true}) ->
     % #sdl_flows{forward_num = ForwardNum, p2p_num = P2PNum, inbound_num = InboundNum} = sdlan_pb:decode_msg(Body, sdl_flows),
-    lager:debug("[sdlan_channel] body: ~p", [Body]),
+    efka_logger:debug("[sdlan_channel] body: ~p", [Body]),
     {noreply, State};
 
 handle_info({tcp_error, Sock, Reason}, State = #state{socket = Sock}) ->
-    lager:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
+    efka_logger:notice("[sdlan_channel] tcp_error: ~p", [Reason]),
     {stop, normal, State};
 
 handle_info({tcp_closed, Sock}, State = #state{socket = Sock}) ->
-    lager:notice("[sdlan_channel] tcp_closed", []),
+    efka_logger:notice("[sdlan_channel] tcp_closed", []),
     {stop, normal, State};
 
 %% 关闭当前通道
@@ -101,11 +101,11 @@ handle_info({stop, Reason}, State) ->
     {stop, Reason, State};
 
 handle_info(Info, State) ->
-    lager:warning("[sdlan_channel] get a unknown message: ~p, channel will closed", [Info]),
+    efka_logger:warning("[sdlan_channel] get a unknown message: ~p, channel will closed", [Info]),
     {noreply, State}.
 
 terminate(Reason, #state{}) ->
-    lager:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
+    efka_logger:warning("[sdlan_channel] stop with reason: ~p", [Reason]),
     ok.
 
 code_change(_OldVsn, State, _Extra) ->

rebar.config

@@ -1,12 +1,8 @@
 {erl_opts, [debug_info]}.
 {deps, [
-    {hackney, ".*", {git, "https://github.com/benoitc/hackney.git", {tag, "1.17.0"}}},
     {sync, ".*", {git, "https://github.com/rustyio/sync.git", {branch, "master"}}},
-    {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"}}},
+    {gpb, ".*", {git, "https://github.com/tomas-abrahamsson/gpb.git", {tag, "4.20.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"}}}
 ]}.
 
 {relx, [{release, {efka, "0.1.0"},
@@ -39,8 +35,6 @@
                      ]
             }]}]}.
 
-{erl_opts, [{parse_transform,lager_transform}]}.
-
 {project_plugins, [
     %% 或从 Git 仓库拉取最新版本
     {pc, {git, "https://github.com/blt/port_compiler.git", {tag, "v1.15.0"}}}

rebar.lock

@@ -1,57 +0,0 @@
-{"1.2.0",
-[{<<"certifi">>,{pkg,<<"certifi">>,<<"2.5.3">>},1},
- {<<"esockd">>,
-  {git,"https://github.com/emqx/esockd.git",
-       {ref,"9b959fc11a1c398a589892f335235be6c5b4a454"}},
-  0},
- {<<"fs">>,{pkg,<<"fs">>,<<"6.1.1">>},1},
- {<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
- {<<"gpb">>,
-  {git,"https://github.com/tomas-abrahamsson/gpb.git",
-       {ref,"edda1006d863a09509673778c455d33d88e6edbc"}},
-  0},
- {<<"hackney">>,
-  {git,"https://github.com/benoitc/hackney.git",
-       {ref,"f642ee0eccaff9aa15707ae3a3effa29f2920e41"}},
-  0},
- {<<"idna">>,{pkg,<<"idna">>,<<"6.1.1">>},1},
- {<<"jiffy">>,
-  {git,"https://github.com/davisp/jiffy.git",
-       {ref,"50daa80a62a97ffb6dd46ea9cb8ccb4930cbf1ae"}},
-  0},
- {<<"lager">>,
-  {git,"https://github.com/erlang-lager/lager.git",
-       {ref,"459a3b2cdd9eadd29e5a7ce5c43932f5ccd6eb88"}},
-  0},
- {<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},1},
- {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.2.0">>},1},
- {<<"parse_trans">>,
-  {git,"https://github.com/uwiger/parse_trans",
-       {ref,"6f3645afb43c7c57d61b54ef59aecab288ce1013"}},
-  0},
- {<<"ssl_verify_fun">>,{pkg,<<"ssl_verify_fun">>,<<"1.1.6">>},1},
- {<<"sync">>,
-  {git,"https://github.com/rustyio/sync.git",
-       {ref,"7dc303ed4ce8d26db82e171dbbd7c41067852c65"}},
-  0},
- {<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.7.0">>},1}]}.
-[
-{pkg_hash,[
- {<<"certifi">>, <<"70BDD7E7188C804F3A30EE0E7C99655BC35D8AC41C23E12325F36AB449B70651">>},
- {<<"fs">>, <<"9D147B944D60CFA48A349F12D06C8EE71128F610C90870BDF9A6773206452ED0">>},
- {<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
- {<<"idna">>, <<"8A63070E9F7D0C62EB9D9FCB360A7DE382448200FBBD1B106CC96D3D8099DF8D">>},
- {<<"metrics">>, <<"25F094DEA2CDA98213CECC3AEFF09E940299D950904393B2A29D191C346A8486">>},
- {<<"mimerl">>, <<"67E2D3F571088D5CFD3E550C383094B47159F3EEE8FFA08E64106CDF5E981BE3">>},
- {<<"ssl_verify_fun">>, <<"CF344F5692C82D2CD7554F5EC8FD961548D4FD09E7D22F5B62482E5AEAEBD4B0">>},
- {<<"unicode_util_compat">>, <<"BC84380C9AB48177092F43AC89E4DFA2C6D62B40B8BD132B1059ECC7232F9A78">>}]},
-{pkg_hash_ext,[
- {<<"certifi">>, <<"ED516ACB3929B101208A9D700062D520F3953DA3B6B918D866106FFA980E1C10">>},
- {<<"fs">>, <<"EF94E95FFE79916860649FED80AC62B04C322B0BB70F5128144C026B4D171F8B">>},
- {<<"goldrush">>, <<"99CB4128CFFCB3227581E5D4D803D5413FA643F4EB96523F77D9E6937D994CEB">>},
- {<<"idna">>, <<"92376EB7894412ED19AC475E4A86F7B413C1B9FBB5BD16DCCD57934157944CEA">>},
- {<<"metrics">>, <<"69B09ADDDC4F74A40716AE54D140F93BEB0FB8978D8636EADED0C31B6F099F16">>},
- {<<"mimerl">>, <<"F278585650AA581986264638EBF698F8BB19DF297F66AD91B18910DFC6E19323">>},
- {<<"ssl_verify_fun">>, <<"BDB0D2471F453C88FF3908E7686F86F9BE327D065CC1EC16FA4540197EA04680">>},
- {<<"unicode_util_compat">>, <<"25EEE6D67DF61960CF6A794239566599B09E17E668D3700247BC498638152521">>}]}
-].