Current section
Files
Jump to
Current section
Files
src/quicer_stream.erl
%%--------------------------------------------------------------------
%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(quicer_stream).
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-include("quicer_types.hrl").
-behaviour(gen_server).
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Stream Callbacks
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-callback init_handoff(stream_handle(), stream_opts(), connection_handle(), new_stream_props()) -> cb_ret().
%% Prepare callback state before ownership handoff
-callback post_handoff(stream_handle(), PostInfo::term(), cb_state()) -> cb_ret().
%% Post handoff with PostData if any. Most common action is to set the stream mode to active.
-callback new_stream(stream_handle(), new_stream_props(), connection_handle()) -> cb_ret().
%% Stream accepter is assigned to the owner of the new stream
-callback start_completed(stream_handle(), stream_start_completed_props(), cb_state()) -> cb_ret().
%% Handle local initiated stream start completed
-callback send_complete(stream_handle(), IsCanceled::boolean(), cb_state()) -> cb_ret().
%% Handle send completed.
-callback peer_send_shutdown(stream_handle(), undefined, cb_state()) -> cb_ret().
%% Handle stream peer_send_shutdown.
-callback peer_send_aborted(stream_handle(), error_code(), cb_state()) -> cb_ret().
%% Handle stream peer_send_aborted.
-callback peer_receive_aborted(stream_handle(), error_code(), cb_state()) -> cb_ret().
%% Handle stream peer_receive_aborted
-callback send_shutdown_complete(stream_handle(), IsGraceful::boolean(), cb_state()) -> cb_ret().
%% Handle stream send_shutdown_complete.
%% Happen immediately on an abortive send or after a graceful send has been acknowledged by the peer.
-callback stream_closed(stream_handle(), stream_closed_props(), cb_state()) -> cb_ret().
%% Handle stream closed, Both endpoints of sending and receiving of the stream have been shut down.
-callback peer_accepted(connection_handle(), stream_handle(), cb_state()) -> cb_ret().
%% Handle stream 'peer_accepted'.
%% The stream which **was not accepted** due to peer flow control is now accepted by the peer.
-callback passive(stream_handle(), undefined, cb_state()) -> cb_ret().
%% Stream now in 'passive' mode.
-callback handle_stream_data(stream_handle(), binary(), recv_data_props(), cb_state() ) -> cb_ret().
%% Stream handle data
-callback handle_call(Req::term(), gen_server:from(), cb_state()) -> cb_ret().
%% Handle API call with callback state.
-callback handle_continue(Cont::term(), cb_state()) -> cb_ret().
%% Handle continue from other callbacks with callback state.
-callback handle_info(Info::term(), cb_state()) -> cb_ret().
%% Handle unhandled info with callback state.
-optional_callbacks([post_handoff/3, handle_stream_data/4, handle_call/3, handle_info/2, handle_continue/2]).
-import(quicer_lib, [default_cb_ret/2]).
-export_type([cb_state/0, cb_ret/0]).
-type cb_state() :: term().
-type cb_ret() :: {ok, cb_state()} %% ok and update cb_state
| {error, Reason::term(), cb_state()} %% error handling per callback
| {hibernate, cb_state()} %% ok but also hibernate process
| {{continue, Continue :: term()}, cb_state()} %% split callback work with Continue
| {timeout(), cb_state()} %% ok but also hibernate process
| {stop, Reason :: term(), cb_state()}. %% terminate with reason
%% API
-export([ %% Start before conn handshake, with only Conn handle
start_link/3
%% Start after conn handshake with new Stream Handle
, start_link/5
, send/2
, send/3
]).
%% Helpers
-export([wait_for_handoff/2]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
handle_continue/2,
terminate/2]).
-define(post_init, post_init).
-type state() :: #{ stream := quicer:stream_handle()
, conn := quicer:connection_handle()
, callback := atom()
, callback_state := term()
, is_owner := boolean()
, stream_opts := map()
}.
%%%===================================================================
%%% API
%%%===================================================================
%%--------------------------------------------------------------------
%% @doc
%% Starts the server
%% @end
%%--------------------------------------------------------------------
%% Start acceptor or Start new local stream
-spec start_link(Callback :: module(),
Conn :: quicer:connection_handle(),
StreamOpts :: map()) -> {ok, Pid :: pid()} |
{error, Error :: {already_started, pid()}} |
{error, Error :: term()} |
ignore.
start_link(Callback, Conn, StreamOpts) when is_atom(Callback) ->
gen_server:start_link(?MODULE, [Callback, Conn, StreamOpts], []).
%%--------------------------------------------------------------------
%% @doc Start a new stream owner process and
%% then handoff ownership to this process
%%--------------------------------------------------------------------
-spec start_link(Callback :: module(),
Stream :: quicer:connection_handle(),
Conn :: quicer:connection_handle(),
StreamOpts :: map(),
Props :: new_stream_props()
) -> {ok, Pid :: pid()} |
{error, Error :: {already_started, pid()}} |
{error, Error :: term()} |
ignore.
start_link(Callback, Stream, Conn, StreamOpts, Props)
when Callback =/= undefined
andalso is_atom(Callback)
andalso is_map(Props) ->
gen_server:start_link(?MODULE, [Callback, Stream, Conn, StreamOpts, Props, self()], []).
-spec send(pid(), binary()) -> {ok, Length::non_neg_integer()} | {error, any()}.
send(StreamProc, Data) ->
send(StreamProc, Data, ?QUICER_SEND_FLAG_SYNC).
send(StreamProc, Data, Flag) ->
gen_server:call(StreamProc, {send, Data, Flag}, infinity).
wait_for_handoff(FromOwner, Stream) ->
MRef = erlang:monitor(process, FromOwner),
receive
{handoff_done, Stream, PostInfo} ->
?tp(debug, #{event=>stream_owner_handoff_done,
stream=>Stream,
module=>?MODULE,
post_info=>PostInfo
}),
{ok, PostInfo};
{'DOWN', MRef, process, FromOwner, _Info} ->
?tp(debug, #{event=>handoff_exit_owner_down,
stream=>Stream,
module=>?MODULE
}),
{error, owner_down}
%% For correctness we should never add timeout
end.
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Initializes the server
%% @end
%%--------------------------------------------------------------------
-spec init(Args :: term()) -> {ok, state()} |
{ok, state(), Timeout :: timeout()} |
{ok, state(), hibernate} |
{stop, Reason :: term()} |
ignore.
%% With only Conn handle
%% Stream will be started or accepted.
init([Callback, Conn, StreamOpts]) when is_list(StreamOpts) ->
init([Callback, Conn, maps:from_list(StreamOpts)]);
init([Callback, Conn, StreamOpts]) ->
process_flag(trap_exit, true),
IsLocal = maps:get(is_local, StreamOpts, false),
InitState = #{ stream_opts => StreamOpts
, conn => Conn
, callback => Callback
, callback_state => undefined
, fpbuffer => maybe_buffer(StreamOpts)
},
case IsLocal of
false ->
%% Accept remote stream,
case quicer:async_accept_stream(Conn, StreamOpts) of
{ok, Conn} ->
?tp(new_stream_accept, #{module=>?MODULE, conn=>Conn}),
{ok, InitState#{ stream => undefined
, is_owner => false
, is_local => false
}};
{error, Reason} ->
{stop, Reason}
end;
true ->
%% Initiate local stream
case quicer:start_stream(Conn, StreamOpts) of
{ok, Stream} ->
?tp(new_stream_local_start, #{module=>?MODULE, stream=>Stream}),
IsUni = quicer:is_unidirectional(
maps:get(open_flag, StreamOpts, ?QUIC_STREAM_START_FLAG_NONE)),
{ok, InitState#{ stream => Stream
, is_owner => true
, callback_state :=
#{ conn => Conn
, is_owner => true
, is_local => true
, is_unidir => IsUni
}
}};
{error, Reason, SecReason} ->
{stop, {Reason, SecReason}};
{error, Reason} ->
{stop, Reason}
end
end;
%% After conn handshake, with stream handle for remote stream
init([Callback, Stream, Conn, StreamOpts, Props, PrevOwner]) when is_list(StreamOpts) ->
?tp(new_stream_2, #{module=>?MODULE, stream=>Stream}),
init([Callback, Stream, Conn, maps:from_list(StreamOpts), Props, PrevOwner]);
init([Callback, Stream, Conn, StreamOpts, Props, PrevOwner]) ->
?tp(new_stream_3, #{module=>?MODULE, stream=>Stream, opts => StreamOpts}),
process_flag(trap_exit, true),
case Callback:init_handoff(Stream, StreamOpts, Conn, Props) of
{ok, CBState} ->
State = #{ is_owner => false
, stream_opts => StreamOpts
, conn => Conn
, stream => Stream
, callback => Callback
, callback_state => CBState
, fpbuffer => maybe_buffer(StreamOpts)
},
{ok, State, {continue , {?post_init, PrevOwner}}};
{error, _} = E ->
{stop, E}
end.
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling call messages
%% @end
%%--------------------------------------------------------------------
-spec handle_call(Request :: term(), From :: {pid(), term()}, state()) ->
{reply, Reply :: term(), state()} |
{reply, Reply :: term(), state(), Timeout :: timeout()} |
{reply, Reply :: term(), state(), hibernate} |
{noreply, state()} |
{noreply, state(), Timeout :: timeout()} |
{noreply, state(), hibernate} |
{stop, Reason :: term(), Reply :: term(), state()} |
{stop, Reason :: term(), state()}.
handle_call({send, Data, Flag}, _From,
#{stream := Stream,
stream_opts := _Options, callback_state := _CallbackState} = State) ->
Res = quicer:async_send(Stream, Data, Flag),
{reply, Res, State};
handle_call(Request, From,
#{stream := _Stream,
callback := CallbackModule,
callback_state := CallbackState} = State) ->
default_cb_ret(CallbackModule:handle_call(Request, From, CallbackState), State).
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling cast messages
%% @end
%%--------------------------------------------------------------------
-spec handle_cast(Request :: term(), state()) ->
{noreply, state()} |
{noreply, state(), Timeout :: timeout()} |
{noreply, state(), hibernate} |
{stop, Reason :: term(), state()}.
handle_cast(_Request, State) ->
{noreply, State}.
%%--------------------------------------------------------------------
%% @private
%% @doc
%% Handling all non call/cast messages
%% @end
%%--------------------------------------------------------------------
-spec handle_info(Info :: timeout() | term(), state()) ->
{noreply, state()} |
{noreply, state(), Timeout :: timeout()} |
{noreply, state(), hibernate} |
{stop, Reason :: normal | term(), state()}.
handle_info({quic, closed, undefined, undefined},
#{ is_local := false
, stream := undefined
, conn := _Conn
} = S) ->
?tp_ignore_side_effects_in_prod(acceptor_recv_conn_stop, #{conn => _Conn, module => ?MODULE, pid => self()}),
{stop, normal, S};
%% For acceptor
handle_info({quic, new_stream, Stream, #{flags := _Flags, is_orphan := false} = Props},
#{ stream_opts := Options
, stream := undefined
, conn := Conn
, callback := CallbackModule
, callback_state := undefined
} = State) ->
?tp_ignore_side_effects_in_prod(new_stream, #{module=>?MODULE, stream=>Stream, stream_flags => _Flags}),
try CallbackModule:new_stream(Stream, maps:merge(Options, Props), Conn) of
{ok, CallbackState} ->
{noreply, State#{stream := Stream, callback_state := CallbackState}};
{error, Reason} ->
{stop, Reason, State#{stream := Stream}}
catch
_:Reason:ST ->
maybe_log_stracetrace(ST),
{stop, {new_stream_crash, Reason}, State#{stream := Stream}}
end;
handle_info({quic, Bin, Stream, Props},
#{ fpbuffer := disabled
, stream := Stream, callback := M
, callback_state := CallbackState } = State)
when is_binary(Bin) ->
%% FPbuffer is disabled, callback module should handle out of order delivery
?tp_ignore_side_effects_in_prod(stream_data, #{module=>?MODULE, stream=>Stream}),
default_cb_ret(M:handle_stream_data(Stream, Bin, Props, CallbackState), State);
handle_info({quic, Bin, Stream, Props} = Evt,
#{ stream := Stream, callback := M
, fpbuffer := Buffer
, callback_state := CallbackState } = State)
when is_binary(Bin) andalso Buffer =/= disabled ->
%% FPbuffer is enabled, callback module get ordered data
?tp_ignore_side_effects_in_prod(stream_data, #{module=>?MODULE, stream=>Stream, buffer => Buffer}),
case quicer:update_fpbuffer(quicer:quic_data(Evt), Buffer) of
{[], NewBuffer} ->
{noreply, State#{ fpbuffer := NewBuffer} };
{DataList, NewBuffer} ->
{IoListData, NewSize, NewFlag} = quicer:merge_quic_datalist(DataList),
AppData = iolist_to_binary(IoListData),
default_cb_ret(M:handle_stream_data(Stream, AppData,
Props#{len := NewSize, flags := NewFlag},
CallbackState),
State#{fpbuffer := NewBuffer})
end;
handle_info({quic, start_completed, Stream,
#{ status := _AtomStatus
, stream_id := _StreamId
, is_peer_accepted := _PeerAccepted} = Props
}
, #{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => start_completed, props => Props}),
default_cb_ret(M:start_completed(Stream, Props, CBState), State);
handle_info({quic, send_complete, Stream, IsSendCanceled},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event=>send_complete, is_canceled=>IsSendCanceled}),
default_cb_ret(M:send_complete(Stream, IsSendCanceled, CBState), State);
handle_info({quic, peer_send_shutdown, Stream, undefined},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => peer_send_shutdown}),
default_cb_ret(M:peer_send_shutdown(Stream, undefined, CBState), State);
handle_info({quic, peer_send_aborted, Stream, ErrorCode},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => peer_send_aborted, error_code => ErrorCode}),
default_cb_ret(M:peer_send_aborted(Stream, ErrorCode, CBState), State);
handle_info({quic, peer_receive_aborted, Stream, ErrorCode},
#{ callback := M,
callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => peer_receive_aborted, error_code => ErrorCode}),
default_cb_ret(M:peer_receive_aborted(Stream, ErrorCode, CBState), State);
handle_info({quic, send_shutdown_complete, Stream, IsGraceful},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => send_shutdown_complete, is_graceful => IsGraceful}),
default_cb_ret(M:send_shutdown_complete(Stream, IsGraceful, CBState), State);
handle_info({quic, stream_closed, Stream, Flags},
#{ callback := M
, conn := C
, callback_state := CbState} = State) when C =/= undefined andalso is_map(Flags) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, conn=>C, stream=>Stream, event=>stream_closed, flags=>Flags}),
default_cb_ret(M:stream_closed(Stream, Flags, CbState), State);
handle_info({quic, peer_accepted, Stream, undefined},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => peer_accepted}),
default_cb_ret(M:peer_accepted(Stream, undefined, CBState), State);
handle_info({quic, passive, Stream, undefined},
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => passive}),
default_cb_ret(M:passive(Stream, undefined, CBState), State);
handle_info(Info,
#{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event => info}),
default_cb_ret(M:handle_info(Info, CBState), State).
%% @TODO handle_info({EXIT....
%%--------------------------------------------------------------------
%% @private
%% @doc
%% This callback is optional, so callback modules need to export it only if they return
%% one of the tuples containing {continue,Continue} from another callback.
%% If such a {continue,_} tuple is used and the callback is not implemented,
%% the process will exit with undef error.
-spec handle_continue(Continue::term(), State::term()) ->
{noreply, state()} |
{noreply, state(), Timeout :: timeout()} |
{noreply, state(), hibernate} |
{stop, Reason :: normal | term(), state()}.
handle_continue({?post_init, PrevOwner}, #{ is_owner := false, stream := Stream
, callback_state := CBState
, callback := M} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{event=>?post_init, module=>?MODULE, stream=>Stream}),
case wait_for_handoff(PrevOwner, Stream) of
{ok, PostInfo}->
case erlang:function_exported(M, post_handoff, 3) of
true ->
default_cb_ret(M:post_handoff(Stream, PostInfo, CBState), State#{is_owner => true});
false ->
{noreply, State#{is_owner => true}}
end;
{error, owner_down} ->
{stop, owner_down, State}
end;
handle_continue(Other, #{ callback := M
, callback_state := CBState} = State) ->
?tp_ignore_side_effects_in_prod(debug, #{module=>?MODULE, event=>continue, stream=>maps:get(stream, State)}),
default_cb_ret(M:handle_continue(Other, CBState), 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.
%% @end
%%--------------------------------------------------------------------
-spec terminate(Reason :: normal | shutdown | {shutdown, term()} | term(),
state()) -> any().
terminate(Reason, _State) ->
error_code(Reason),
ok.
%%%===================================================================
%%% Internal functions
%%%===================================================================
error_code(normal) ->
'QUIC_ERROR_NO_ERROR';
error_code(shutdown) ->
'QUIC_ERROR_NO_ERROR';
error_code(_) ->
%% @todo mapping errors to error code
%% for closing stream
'QUIC_ERROR_INTERNAL_ERROR'.
maybe_log_stracetrace(ST) ->
logger:error("~p~n", [ST]),
ok.
maybe_buffer(#{disable_fpbuffer := true}) ->
disabled;
maybe_buffer(_) ->
%% Default enable
quicer:new_fpbuffer().