remove efka_stream
This commit is contained in:
parent
0fc1d25219
commit
3f2aa00f35
@ -1,164 +0,0 @@
|
||||
%%%-------------------------------------------------------------------
|
||||
%%% @author anlicheng
|
||||
%%% @copyright (C) 2025, <COMPANY>
|
||||
%%% @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}.
|
||||
Loading…
x
Reference in New Issue
Block a user