支持文件上传

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);

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),
@@ -112,3 +112,20 @@ starts_with(Binary, Prefix) when is_binary(Binary), is_binary(Prefix) ->
         <<Prefix:PrefixSize/binary, _Rest/binary>> -> true;
         _ -> false
     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.
+