diff --git a/src/efka_stream.erl b/src/efka_stream.erl deleted file mode 100644 index 4428243..0000000 --- a/src/efka_stream.erl +++ /dev/null @@ -1,164 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2025, -%%% @doc -%%% -%%% @end -%%% Created : 13. 11月 2025 10:57 -%%%------------------------------------------------------------------- --module(efka_stream). --author("anlicheng"). - --behaviour(gen_server). - -%% API --export([start_monitor/1]). --export([setup/3, data/2, finish/1]). - -%% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). - --define(SERVER, ?MODULE). - --record(state, { - parent_pid :: pid(), - ref :: reference(), - file_size = 0 :: integer(), - acc_size = 0 :: integer(), - real_file :: undefined | string(), - io_device :: undefined | file:fd() -}). - -%%%=================================================================== -%%% API -%%%=================================================================== - --spec setup(StreamPid :: pid(), FileName :: string(), FileSize :: integer()) -> {ok, Path :: string()}. -setup(StreamPid, FileName, FileSize) when is_pid(StreamPid), is_list(FileName), is_integer(FileSize) -> - gen_server:call(StreamPid, {setup, FileName, FileSize}). - --spec data(StreamPid :: pid(), ChunkData :: binary()) -> no_return(). -data(StreamPid, ChunkData) when is_pid(StreamPid), is_binary(ChunkData) -> - gen_server:cast(StreamPid, {data, ChunkData}). - --spec finish(StreamPid :: pid()) -> no_return(). -finish(StreamPid) when is_pid(StreamPid) -> - gen_server:cast(StreamPid, finish). - -%% @doc Spawns the server and registers the local name (unique) --spec(start_monitor(ParentPid :: pid()) -> - {ok, {Pid :: pid(), MonRef :: reference()}} | ignore | {error, Reason :: term()}). -start_monitor(ParentPid) when is_pid(ParentPid) -> - gen_server:start_monitor(?MODULE, [ParentPid], []). - -%%%=================================================================== -%%% gen_server callbacks -%%%=================================================================== - -%% @private -%% @doc Initializes the server --spec(init(Args :: term()) -> - {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | - {stop, Reason :: term()} | ignore). -init([ParentPid]) -> - Ref = erlang:monitor(process, ParentPid), - {ok, #state{parent_pid = ParentPid, ref = Ref}}. - -%% @private -%% @doc Handling call messages --spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, - State :: #state{}) -> - {reply, Reply :: term(), NewState :: #state{}} | - {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_call({setup, FileName, FileSize}, _From, State = #state{}) -> - {RealFileName, Path} = make_file(filename:basename(FileName)), - {ok, IoDevice} = file:open(RealFileName, [write]), - - {reply, {ok, Path}, State#state{io_device = IoDevice, real_file = RealFileName, file_size = FileSize, acc_size = 0}}. - -%% @private -%% @doc Handling cast messages --spec(handle_cast(Request :: term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_cast({data, ChunkData}, State = #state{io_device = IoDevice, acc_size = AccSize}) -> - Data = base64:decode(ChunkData), - Len = byte_size(Data), - - ok = file:write(IoDevice, Data), - {noreply, State#state{acc_size = AccSize + Len}}; -handle_cast(finish, State = #state{parent_pid = ParentPid, io_device = IoDevice, acc_size = AccSize, file_size = FileSize, real_file = RealFile}) -> - case AccSize == FileSize of - true -> - ok = file:close(IoDevice), - ParentPid ! {stream_reply, self(), ok}; - false -> - ok = file:close(IoDevice), - ok = file:delete(RealFile), - ParentPid ! {stream_reply, self(), invalid} - end, - {stop, normal, State}. - -%% @private -%% @doc Handling all non call/cast messages --spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> - {noreply, NewState :: #state{}} | - {noreply, NewState :: #state{}, timeout() | hibernate} | - {stop, Reason :: term(), NewState :: #state{}}). -handle_info({'DOWN', Ref, process, Pid, normal}, State = #state{ref = Ref, parent_pid = Pid}) -> - {noreply, State}; -handle_info({'DOWN', Ref, process, Pid, Reason}, State = #state{ref = Ref, parent_pid = Pid, io_device = IoDevice, real_file = RealFile}) -> - logger:debug("[efka_stream] ws_channel close with reason: ~p", [Reason]), - case IoDevice =:= undefined of - true -> - ok; - false -> - ok = file:close(IoDevice), - RealFile /= undefined andalso file:delete(RealFile) - end, - {stop, normal, State}; -handle_info(_Info, State = #state{}) -> - {noreply, State}. - -%% @private -%% @doc This function is called by a gen_server when it is about to -%% terminate. It should be the opposite of Module:init/1 and do any -%% 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()). -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()) -> - {ok, NewState :: #state{}} | {error, Reason :: term()}). -code_change(_OldVsn, State = #state{}, _Extra) -> - {ok, State}. - -%%%=================================================================== -%%% Internal functions -%%%=================================================================== - --spec make_file(Basename :: string()) -> {string(), string()}. -make_file(Basename) when is_list(Basename) -> - {ok, UploadDir} = application:get_env(efka, upload_dir), - {{Y, M, D}, _} = calendar:local_time(), - DateDir = io_lib:format("~p-~p-~p", [Y, M, D]), - BaseDir = UploadDir ++ DateDir, - case filelib:is_dir(BaseDir) of - true -> - ok; - false -> - ok = file:make_dir(BaseDir) - end, - Path = DateDir ++ "/" ++ Basename, - - {UploadDir ++ Path, Path}.