From 2f87c24e9d1d50173ab51524a6d13eca7a1fd3ce Mon Sep 17 00:00:00 2001
From: anlicheng <244108715@qq.com>
Date: Thu, 13 Nov 2025 17:41:07 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=96=87=E4=BB=B6=E4=B8=8A?=
 =?UTF-8?q?=E4=BC=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 apps/efka/src/channel/ws_channel.erl |  2 +-
 apps/efka/src/efka_util.erl          | 21 +++++++++++++++++++--
 2 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/apps/efka/src/channel/ws_channel.erl b/apps/efka/src/channel/ws_channel.erl
index dcb6b2b..d1fbdb3 100644
--- a/apps/efka/src/channel/ws_channel.erl
+++ b/apps/efka/src/channel/ws_channel.erl
@@ -181,7 +181,7 @@ handle_request(#{<<"method">> := <<"stream_chunk">>,
     case maps:find(StreamId, StreamMap) of
         error ->
             {ok, State};
-        {ok, StreamPid} ->
+        {ok, {StreamPid, _}} ->
             case ChunkData =:= <<>> of
                 true ->
                     efka_stream:finish(StreamPid);
diff --git a/apps/efka/src/efka_util.erl b/apps/efka/src/efka_util.erl
index eae5318..1de8e78 100644
--- a/apps/efka/src/efka_util.erl
+++ b/apps/efka/src/efka_util.erl
@@ -14,7 +14,7 @@
 -export([timestamp/0, number_format/2, timestamp_ms/0, float_to_binary/2, int_format/2]).
 -export([chunks/2, rand_bytes/1, uuid/0, md5/1, sha_uuid/0]).
 -export([json_data/1, json_error/2]).
--export([starts_with/2]).
+-export([starts_with/2, file_md5/1]).
 
 get_file_md5(FilePath) when is_list(FilePath) ->
     {ok, FileData} = file:read_file(FilePath),
@@ -111,4 +111,21 @@ starts_with(Binary, Prefix) when is_binary(Binary), is_binary(Prefix) ->
     case Binary of
         <<Prefix:PrefixSize/binary, _Rest/binary>> -> true;
         _ -> false
-    end.
\ No newline at end of file
+    end.
+
+-spec file_md5(FilePath :: string()) -> Md5 :: string().
+file_md5(FilePath) when is_list(FilePath) ->
+    {ok, F} = file:open(FilePath, [read, binary]),
+    Digest = md5_loop(F, crypto:hash_init(md5)),
+    file:close(F),
+    lists:flatten(io_lib:format("~32.16.0b", [binary:decode_unsigned(Digest)])).
+
+md5_loop(F, Context) ->
+    %% 每次读取 1MB，可调整块大小
+    case file:read(F, 1024 * 1024) of
+        eof ->
+            crypto:hash_final(Context);
+        {ok, Bin} ->
+            md5_loop(F, crypto:hash_update(Context, Bin))
+    end.
+