yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore(quic): clean code

This commit is contained in:
William Yang 2023-01-09 09:17:03 +01:00
parent 22dcf5907e
commit 00f615a1e3
8 changed files with 1817 additions and 2037 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apps/emqx/include/emqx_quic.hrl
View File

@ -1,5 +1,5 @@
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. %% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.

14
apps/emqx/src/emqx_connection.erl
View File

@ -119,10 +119,7 @@
limiter_timer :: undefined | reference(), limiter_timer :: undefined | reference(),
%% QUIC conn pid if is a pid %% QUIC conn pid if is a pid
quic_conn_pid :: maybe(pid()), quic_conn_pid :: maybe(pid())
%% QUIC control stream callback state
quic_ctrl_state :: map()
}). }).
-record(retry, { -record(retry, {
@ -378,8 +375,7 @@ init_state(
limiter_buffer = queue:new(), limiter_buffer = queue:new(),
limiter_timer = undefined, limiter_timer = undefined,
%% for quic streams to inherit %% for quic streams to inherit
quic_conn_pid = maps:get(conn_pid, Opts, undefined), quic_conn_pid = maps:get(conn_pid, Opts, undefined)
quic_ctrl_state = #{}
}. }.
run_loop( run_loop(
@ -928,12 +924,6 @@ handle_info({sock_error, Reason}, State) ->
handle_info({sock_closed, Reason}, close_socket(State)); handle_info({sock_closed, Reason}, close_socket(State));
handle_info({quic, Event, Handle, Prop}, State) -> handle_info({quic, Event, Handle, Prop}, State) ->
emqx_quic_stream:Event(Handle, Prop, State); emqx_quic_stream:Event(Handle, Prop, State);
%% handle_info({quic, peer_send_shutdown, _Stream}, State) ->
%% handle_info({sock_closed, force}, close_socket(State));
%% handle_info({quic, closed, _Channel, ReasonFlag}, State) ->
%% handle_info({sock_closed, ReasonFlag}, State);
%% handle_info({quic, closed, _Stream}, State) ->
%% handle_info({sock_closed, force}, State);
handle_info(Info, State) -> handle_info(Info, State) ->
with_channel(handle_info, [Info], State). with_channel(handle_info, [Info], State).

43
apps/emqx/src/emqx_quic_connection.erl
View File

@ -17,13 +17,11 @@
%% @doc impl. the quic connection owner process. %% @doc impl. the quic connection owner process.
-module(emqx_quic_connection). -module(emqx_quic_connection).
-include("logger.hrl").
-ifndef(BUILD_WITHOUT_QUIC). -ifndef(BUILD_WITHOUT_QUIC).
-include("logger.hrl").
-include_lib("quicer/include/quicer.hrl"). -include_lib("quicer/include/quicer.hrl").
-include_lib("emqx/include/emqx_quic.hrl"). -include_lib("emqx/include/emqx_quic.hrl").
-else.
-define(QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0).
-endif.
-behavior(quicer_connection). -behavior(quicer_connection).
@ -55,10 +53,9 @@
%% Pid of ctrl stream %% Pid of ctrl stream
ctrl_pid := undefined | pid(), ctrl_pid := undefined | pid(),
%% quic connecion handle %% quic connecion handle
conn := undefined | quicer:conneciton_hanlder(), conn := undefined | quicer:conneciton_handle(),
%% streams that handoff from this process, excluding control stream %% Data streams that handoff from this process
%% these streams could die/closed without effecting the connecion/session. %% these streams could die/close without effecting the connecion/session.
%@TODO type? %@TODO type?
streams := [{pid(), quicer:stream_handle()}], streams := [{pid(), quicer:stream_handle()}],
%% New stream opts %% New stream opts
@ -82,22 +79,20 @@ activate_data_streams(ConnOwner, {PS, Serialize, Channel}) ->
gen_server:call(ConnOwner, {activate_data_streams, {PS, Serialize, Channel}}, infinity). gen_server:call(ConnOwner, {activate_data_streams, {PS, Serialize, Channel}}, infinity).
%% @doc conneciton owner init callback %% @doc conneciton owner init callback
-spec init(map() | list()) -> {ok, cb_state()}. -spec init(map()) -> {ok, cb_state()}.
init(ConnOpts) when is_list(ConnOpts) ->
init(maps:from_list(ConnOpts));
init(#{stream_opts := SOpts} = S) when is_list(SOpts) -> init(#{stream_opts := SOpts} = S) when is_list(SOpts) ->
init(S#{stream_opts := maps:from_list(SOpts)}); init(S#{stream_opts := maps:from_list(SOpts)});
init(ConnOpts) when is_map(ConnOpts) -> init(ConnOpts) when is_map(ConnOpts) ->
{ok, init_cb_state(ConnOpts)}. {ok, init_cb_state(ConnOpts)}.
-spec closed(quicer:conneciton_hanlder(), quicer:conn_closed_props(), cb_state()) -> -spec closed(quicer:conneciton_handle(), quicer:conn_closed_props(), cb_state()) ->
{stop, normal, cb_state()}. {stop, normal, cb_state()}.
closed(_Conn, #{is_peer_acked := _} = Prop, S) -> closed(_Conn, #{is_peer_acked := _} = Prop, S) ->
?SLOG(debug, Prop), ?SLOG(debug, Prop),
{stop, normal, S}. {stop, normal, S}.
%% @doc handle the new incoming connecion as the connecion acceptor. %% @doc handle the new incoming connecion as the connecion acceptor.
-spec new_conn(quicer:connection_handler(), quicer:new_conn_props(), cb_state()) -> -spec new_conn(quicer:connection_handle(), quicer:new_conn_props(), cb_state()) ->
{ok, cb_state()} | {error, any()}. {ok, cb_state()} | {error, any()}.
new_conn( new_conn(
Conn, Conn,
@ -133,7 +128,7 @@ new_conn(
end. end.
%% @doc callback when connection is connected. %% @doc callback when connection is connected.
-spec connected(quicer:connection_handler(), quicer:connected_props(), cb_state()) -> -spec connected(quicer:connection_handle(), quicer:connected_props(), cb_state()) ->
{ok, cb_state()} | {error, any()}. {ok, cb_state()} | {error, any()}.
connected(_Conn, Props, S) -> connected(_Conn, Props, S) ->
?SLOG(debug, Props), ?SLOG(debug, Props),
@ -185,21 +180,21 @@ new_stream(
Props Props
), ),
quicer:handoff_stream(Stream, NewStreamOwner, {PS, Serialize, Channel}), quicer:handoff_stream(Stream, NewStreamOwner, {PS, Serialize, Channel}),
%% @TODO keep them in ``inactive_streams' %% @TODO maybe keep them in `inactive_streams'
{ok, S#{streams := [{NewStreamOwner, Stream} | Streams]}}. {ok, S#{streams := [{NewStreamOwner, Stream} | Streams]}}.
%% @doc callback for handling for remote connecion shutdown. %% @doc callback for handling remote connecion shutdown.
-spec shutdown(quicer:connection_handle(), quicer:error_code(), cb_state()) -> cb_ret(). -spec shutdown(quicer:connection_handle(), quicer:error_code(), cb_state()) -> cb_ret().
shutdown(Conn, _ErrorCode, S) -> shutdown(Conn, ErrorCode, S) ->
%% @TODO check spec what to set for the ErrorCode? ErrorCode =/= 0 andalso ?SLOG(debug, #{error_code => ErrorCode, state => S}),
quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
{ok, S}. {ok, S}.
%% @doc callback for handling for transport error, such as idle timeout %% @doc callback for handling transport error, such as idle timeout
-spec transport_shutdown(quicer:connection_handle(), quicer:transport_shutdown_props(), cb_state()) -> -spec transport_shutdown(quicer:connection_handle(), quicer:transport_shutdown_props(), cb_state()) ->
cb_ret(). cb_ret().
transport_shutdown(_C, _DownInfo, S) -> transport_shutdown(_C, DownInfo, S) when is_map(DownInfo) ->
%% @TODO some counter ?SLOG(debug, DownInfo),
{ok, S}. {ok, S}.
%% @doc callback for handling for peer addr changed. %% @doc callback for handling for peer addr changed.
@ -238,6 +233,7 @@ peer_needs_streams(_C, undefined, S) ->
{ok, S}. {ok, S}.
%% @doc handle API calls %% @doc handle API calls
-spec handle_call(Req :: term(), gen_server:from(), cb_state()) -> cb_ret().
handle_call( handle_call(
{activate_data_streams, {PS, Serialize, Channel} = ActivateData}, {activate_data_streams, {PS, Serialize, Channel} = ActivateData},
_From, _From,
@ -256,7 +252,6 @@ handle_call(_Req, _From, S) ->
{reply, {error, unimpl}, S}. {reply, {error, unimpl}, S}.
%% @doc handle DOWN messages from streams. %% @doc handle DOWN messages from streams.
%% @TODO handle DOWN from supervisor?
handle_info({'EXIT', Pid, Reason}, #{ctrl_pid := Pid, conn := Conn} = S) -> handle_info({'EXIT', Pid, Reason}, #{ctrl_pid := Pid, conn := Conn} = S) ->
case Reason of case Reason of
normal -> normal ->
@ -302,3 +297,7 @@ init_cb_state(#{zone := _Zone} = Map) ->
serialize => undefined, serialize => undefined,
is_resumed => false is_resumed => false
}. }.
%% BUILD_WITHOUT_QUIC
-else.
-endif.

119
apps/emqx/src/emqx_quic_data_stream.erl
View File

@ -1,5 +1,5 @@
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. %% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%% %%
%% Licensed under the Apache License, Version 2.0 (the "License"); %% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License. %% you may not use this file except in compliance with the License.
@ -21,11 +21,14 @@
%% %%
-module(emqx_quic_data_stream). -module(emqx_quic_data_stream).
-ifndef(BUILD_WITHOUT_QUIC).
-behaviour(quicer_remote_stream).
-include_lib("snabbkaffe/include/snabbkaffe.hrl"). -include_lib("snabbkaffe/include/snabbkaffe.hrl").
-include_lib("quicer/include/quicer.hrl"). -include_lib("quicer/include/quicer.hrl").
-include("emqx_mqtt.hrl"). -include("emqx_mqtt.hrl").
-include("logger.hrl"). -include("logger.hrl").
-behaviour(quicer_remote_stream).
%% Connection Callbacks %% Connection Callbacks
-export([ -export([
@ -37,12 +40,12 @@
peer_receive_aborted/3, peer_receive_aborted/3,
send_shutdown_complete/3, send_shutdown_complete/3,
stream_closed/3, stream_closed/3,
peer_accepted/3,
passive/3 passive/3
]). ]).
-export([handle_stream_data/4]). -export([handle_stream_data/4]).
%% gen_server API
-export([activate_data/2]). -export([activate_data/2]).
-export([ -export([
@ -51,9 +54,19 @@
handle_continue/2 handle_continue/2
]). ]).
-type cb_ret() :: quicer_stream:cb_ret().
-type cb_state() :: quicer_stream:cb_state().
-type error_code() :: quicer:error_code().
-type connection_handle() :: quicer:connection_handle().
-type stream_handle() :: quicer:stream_handle().
-type handoff_data() :: {
emqx_frame:parse_state() | undefined,
emqx_frame:serialize_opts() | undefined,
emqx_channel:channel() | undefined
}.
%% %%
%% @doc Activate the data handling. %% @doc Activate the data handling.
%% Data handling is disabled before control stream allows the data processing. %% Note, data handling is disabled before finishing the validation over control stream.
-spec activate_data(pid(), { -spec activate_data(pid(), {
emqx_frame:parse_state(), emqx_frame:serialize_opts(), emqx_channel:channel() emqx_frame:parse_state(), emqx_frame:serialize_opts(), emqx_channel:channel()
}) -> ok. }) -> ok.
@ -61,9 +74,12 @@ activate_data(StreamPid, {PS, Serialize, Channel}) ->
gen_server:call(StreamPid, {activate, {PS, Serialize, Channel}}, infinity). gen_server:call(StreamPid, {activate, {PS, Serialize, Channel}}, infinity).
%% %%
%% @doc Handoff from previous owner, mostly from the connection owner. %% @doc Handoff from previous owner, from the connection owner.
%% @TODO parse_state doesn't look necessary since we have it in post_handoff %% Note, unlike control stream, there is no acceptor for data streams.
%% @TODO -spec %% The connection owner get new stream, spawn new proc and then handover to it.
%%
-spec init_handoff(stream_handle(), map(), connection_handle(), quicer:new_stream_props()) ->
{ok, cb_state()}.
init_handoff( init_handoff(
Stream, Stream,
_StreamOpts, _StreamOpts,
@ -75,10 +91,9 @@ init_handoff(
%% %%
%% @doc Post handoff data stream %% @doc Post handoff data stream
%% %%
%% @TODO -spec -spec post_handoff(stream_handle(), handoff_data(), cb_state()) -> cb_ret().
%%
post_handoff(_Stream, {undefined = _PS, undefined = _Serialize, undefined = _Channel}, S) -> post_handoff(_Stream, {undefined = _PS, undefined = _Serialize, undefined = _Channel}, S) ->
%% Channel isn't ready yet. %% When the channel isn't ready yet.
%% Data stream should wait for activate call with ?MODULE:activate_data/2 %% Data stream should wait for activate call with ?MODULE:activate_data/2
{ok, S}; {ok, S};
post_handoff(Stream, {PS, Serialize, Channel}, S) -> post_handoff(Stream, {PS, Serialize, Channel}, S) ->
@ -86,53 +101,35 @@ post_handoff(Stream, {PS, Serialize, Channel}, S) ->
quicer:setopt(Stream, active, 10), quicer:setopt(Stream, active, 10),
{ok, S#{channel := Channel, serialize := Serialize, parse_state := PS}}. {ok, S#{channel := Channel, serialize := Serialize, parse_state := PS}}.
%% -spec peer_receive_aborted(stream_handle(), error_code(), cb_state()) -> cb_ret().
%% @doc for local initiated stream peer_receive_aborted(Stream, ErrorCode, #{is_unidir := _} = S) ->
%%
peer_accepted(_Stream, _Flags, S) ->
%% we just ignore it
{ok, S}.
peer_receive_aborted(Stream, ErrorCode, #{is_unidir := false} = S) ->
%% we abort send with same reason %% we abort send with same reason
quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, ErrorCode),
{ok, S};
peer_receive_aborted(Stream, ErrorCode, #{is_unidir := true, is_local := true} = S) ->
quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, ErrorCode), quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT, ErrorCode),
{ok, S}. {ok, S}.
peer_send_aborted(Stream, ErrorCode, #{is_unidir := false} = S) -> -spec peer_send_aborted(stream_handle(), error_code(), cb_state()) -> cb_ret().
peer_send_aborted(Stream, ErrorCode, #{is_unidir := _} = S) ->
%% we abort receive with same reason %% we abort receive with same reason
quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, ErrorCode),
{ok, S};
peer_send_aborted(Stream, ErrorCode, #{is_unidir := true, is_local := false} = S) ->
quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, ErrorCode), quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, ErrorCode),
{ok, S}. {ok, S}.
peer_send_shutdown(Stream, _Flags, S) -> -spec peer_send_shutdown(stream_handle(), undefined, cb_state()) -> cb_ret().
peer_send_shutdown(Stream, undefined, S) ->
ok = quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0), ok = quicer:async_shutdown_stream(Stream, ?QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 0),
{ok, S}. {ok, S}.
-spec send_complete(stream_handle(), IsCanceled :: boolean(), cb_state()) -> cb_ret().
send_complete(_Stream, false, S) -> send_complete(_Stream, false, S) ->
{ok, S}; {ok, S};
send_complete(_Stream, true = _IsCanceled, S) -> send_complete(_Stream, true = _IsCanceled, S) ->
{ok, S}. {ok, S}.
-spec send_shutdown_complete(stream_handle(), error_code(), cb_state()) -> cb_ret().
send_shutdown_complete(_Stream, _Flags, S) -> send_shutdown_complete(_Stream, _Flags, S) ->
{ok, S}. {ok, S}.
handle_stream_data( -spec handle_stream_data(stream_handle(), binary(), quicer:recv_data_props(), cb_state()) ->
Stream, cb_ret().
Bin,
_Flags,
#{
is_unidir := false,
channel := undefined,
data_queue := Queue,
stream := Stream
} = State
) when is_binary(Bin) ->
{ok, State#{data_queue := [Bin | Queue]}};
handle_stream_data( handle_stream_data(
_Stream, _Stream,
Bin, Bin,
@ -145,6 +142,7 @@ handle_stream_data(
task_queue := TQ task_queue := TQ
} = State } = State
) when ) when
%% assert get stream data only after channel is created
Channel =/= undefined Channel =/= undefined
-> ->
{MQTTPackets, NewPS} = parse_incoming(list_to_binary(lists:reverse([Bin | QueuedData])), PS), {MQTTPackets, NewPS} = parse_incoming(list_to_binary(lists:reverse([Bin | QueuedData])), PS),
@ -157,25 +155,12 @@ handle_stream_data(
), ),
{{continue, handle_appl_msg}, State#{parse_state := NewPS, task_queue := NewTQ}}. {{continue, handle_appl_msg}, State#{parse_state := NewPS, task_queue := NewTQ}}.
%% Reserved for unidi streams -spec passive(stream_handle(), undefined, cb_state()) -> cb_ret().
%% handle_stream_data(Stream, Bin, _Flags, #{is_unidir := true, peer_stream := PeerStream, conn := Conn} = State) ->
%% case PeerStream of
%% undefined ->
%% {ok, StreamProc} = quicer_stream:start_link(?MODULE, Conn,
%% [ {open_flag, ?QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL}
%% , {is_local, true}
%% ]),
%% {ok, _} = quicer_stream:send(StreamProc, Bin),
%% {ok, State#{peer_stream := StreamProc}};
%% StreamProc when is_pid(StreamProc) ->
%% {ok, _} = quicer_stream:send(StreamProc, Bin),
%% {ok, State}
%% end.
passive(Stream, undefined, S) -> passive(Stream, undefined, S) ->
quicer:setopt(Stream, active, 10), quicer:setopt(Stream, active, 10),
{ok, S}. {ok, S}.
-spec stream_closed(stream_handle(), quicer:stream_closed_props(), cb_state()) -> cb_ret().
stream_closed( stream_closed(
_Stream, _Stream,
#{ #{
@ -197,28 +182,20 @@ stream_closed(
-> ->
{stop, normal, S}. {stop, normal, S}.
-spec handle_call(Request :: term(), From :: {pid(), term()}, cb_state()) -> cb_ret().
handle_call(Call, _From, S) -> handle_call(Call, _From, S) ->
case do_handle_call(Call, S) of do_handle_call(Call, S).
{ok, NewS} ->
{reply, ok, NewS};
{error, Reason, NewS} ->
{reply, {error, Reason}, NewS};
{{continue, _} = Cont, NewS} ->
{reply, ok, NewS, Cont};
{hibernate, NewS} ->
{reply, ok, NewS, hibernate};
{stop, Reason, NewS} ->
{stop, Reason, {stopped, Reason}, NewS}
end.
-spec handle_continue(Continue :: term(), cb_state()) -> cb_ret().
handle_continue(handle_appl_msg, #{task_queue := Q} = S) -> handle_continue(handle_appl_msg, #{task_queue := Q} = S) ->
case queue:out(Q) of case queue:out(Q) of
{{value, Item}, Q2} -> {{value, Item}, Q2} ->
do_handle_appl_msg(Item, S#{task_queue := Q2}); do_handle_appl_msg(Item, S#{task_queue := Q2});
{empty, Q} -> {empty, _Q} ->
{ok, S} {ok, S}
end. end.
%%% Internals
do_handle_appl_msg( do_handle_appl_msg(
{outgoing, Packets}, {outgoing, Packets},
#{ #{
@ -248,7 +225,7 @@ do_handle_appl_msg({incoming, {frame_error, _} = FE}, #{channel := Channel} = S)
-> ->
with_channel(handle_in, [FE], S); with_channel(handle_in, [FE], S);
do_handle_appl_msg({close, Reason}, S) -> do_handle_appl_msg({close, Reason}, S) ->
%% @TODO shall we abort shutdown or graceful shutdown? %% @TODO shall we abort shutdown or graceful shutdown here?
with_channel(handle_info, [{sock_closed, Reason}], S); with_channel(handle_info, [{sock_closed, Reason}], S);
do_handle_appl_msg({event, updated}, S) -> do_handle_appl_msg({event, updated}, S) ->
%% Data stream don't care about connection state changes. %% Data stream don't care about connection state changes.
@ -294,7 +271,6 @@ with_channel(Fun, Args, #{channel := Channel, task_queue := Q} = S) when
}} }}
end. end.
%%% Internals
handle_outgoing(#mqtt_packet{} = P, S) -> handle_outgoing(#mqtt_packet{} = P, S) ->
handle_outgoing([P], S); handle_outgoing([P], S);
handle_outgoing(Packets, #{serialize := Serialize, stream := Stream, is_unidir := false}) when handle_outgoing(Packets, #{serialize := Serialize, stream := Stream, is_unidir := false}) when
@ -373,7 +349,7 @@ init_state(Stream, Connection, OpenFlags, PS) ->
task_queue => queue:new() task_queue => queue:new()
}. }.
-spec do_handle_call(term(), quicer_stream:cb_state()) -> quicer_stream:cb_ret(). -spec do_handle_call(term(), cb_state()) -> cb_ret().
do_handle_call( do_handle_call(
{activate, {PS, Serialize, Channel}}, {activate, {PS, Serialize, Channel}},
#{ #{
@ -386,7 +362,7 @@ do_handle_call(
%% We use quic protocol for flow control, and we don't check return val %% We use quic protocol for flow control, and we don't check return val
case quicer:setopt(Stream, active, true) of case quicer:setopt(Stream, active, true) of
ok -> ok ->
{ok, NewS}; {reply, ok, NewS};
{error, E} -> {error, E} ->
?SLOG(error, #{msg => "set stream active failed", error => E}), ?SLOG(error, #{msg => "set stream active failed", error => E}),
{stop, E, NewS} {stop, E, NewS}
@ -484,3 +460,6 @@ is_datastream_out_pkt(#mqtt_packet{header = #mqtt_packet_header{type = Type}}) w
true; true;
is_datastream_out_pkt(_) -> is_datastream_out_pkt(_) ->
false. false.
%% BUILD_WITHOUT_QUIC
-else.
-endif.

77
apps/emqx/src/emqx_quic_stream.erl
View File

@ -17,8 +17,12 @@
%% MQTT/QUIC Stream %% MQTT/QUIC Stream
-module(emqx_quic_stream). -module(emqx_quic_stream).
-ifndef(BUILD_WITHOUT_QUIC).
-behaviour(quicer_remote_stream). -behaviour(quicer_remote_stream).
-include("logger.hrl").
%% emqx transport Callbacks %% emqx transport Callbacks
-export([ -export([
type/1, type/1,
@ -33,31 +37,14 @@
sockname/1, sockname/1,
peercert/1 peercert/1
]). ]).
-include("logger.hrl").
-ifndef(BUILD_WITHOUT_QUIC).
-include_lib("quicer/include/quicer.hrl"). -include_lib("quicer/include/quicer.hrl").
-else.
%% STREAM SHUTDOWN FLAGS
-define(QUIC_STREAM_SHUTDOWN_FLAG_NONE, 0).
% Cleanly closes the send path.
-define(QUIC_STREAM_SHUTDOWN_FLAG_GRACEFUL, 1).
% Abruptly closes the send path.
-define(QUIC_STREAM_SHUTDOWN_FLAG_ABORT_SEND, 2).
% Abruptly closes the receive path.
-define(QUIC_STREAM_SHUTDOWN_FLAG_ABORT_RECEIVE, 4).
% Abruptly closes both send and receive paths.
-define(QUIC_STREAM_SHUTDOWN_FLAG_ABORT, 6).
-define(QUIC_STREAM_SHUTDOWN_FLAG_IMMEDIATE, 8).
-endif.
-type cb_ret() :: gen_statem:event_handler_result(). -type cb_ret() :: quicer_stream:cb_ret().
-type cb_data() :: emqtt_quic:cb_data(). -type cb_data() :: quicer_stream:cb_state().
-type connection_handle() :: quicer:connection_handle(). -type connection_handle() :: quicer:connection_handle().
-type stream_handle() :: quicer:stream_handle(). -type stream_handle() :: quicer:stream_handle().
-export([ -export([
new_stream/3,
send_complete/3, send_complete/3,
peer_send_shutdown/3, peer_send_shutdown/3,
peer_send_aborted/3, peer_send_aborted/3,
@ -79,13 +66,8 @@
}. }.
%% for accepting %% for accepting
-spec wait -spec wait({pid(), connection_handle(), socket_info()}) ->
({pid(), connection_handle(), socket_info()}) -> {ok, socket()} | {error, enotconn}.
{ok, socket()} | {error, enotconn};
%% For handover
({pid(), connection_handle(), stream_handle(), socket_info()}) ->
{ok, socket()} | {error, any()}.
%%% For Accepting New Remote Stream %%% For Accepting New Remote Stream
wait({ConnOwner, Conn, ConnInfo}) -> wait({ConnOwner, Conn, ConnInfo}) ->
{ok, Conn} = quicer:async_accept_stream(Conn, []), {ok, Conn} = quicer:async_accept_stream(Conn, []),
@ -105,15 +87,8 @@ wait({ConnOwner, Conn, ConnInfo}) ->
{'EXIT', ConnOwner, _Reason} -> {'EXIT', ConnOwner, _Reason} ->
{error, enotconn} {error, enotconn}
end. end.
%% UNUSED, for ownership handover,
%% wait({PrevOwner, Conn, Stream, SocketInfo}) ->
%% case quicer:wait_for_handoff(PrevOwner, Stream) of
%% ok ->
%% {ok, socket(Conn, Stream, SocketInfo)};
%% owner_down ->
%% {error, owner_down}
%% end.
-spec type(_) -> quic.
type(_) -> type(_) ->
quic. quic.
@ -155,7 +130,7 @@ getopts(_Socket, _Opts) ->
{buffer, 80000} {buffer, 80000}
]}. ]}.
%% @TODO supply some App Error Code %% @TODO supply some App Error Code from caller
fast_close({ConnOwner, Conn, _ConnInfo}) when is_pid(ConnOwner) -> fast_close({ConnOwner, Conn, _ConnInfo}) when is_pid(ConnOwner) ->
%% handshake aborted. %% handshake aborted.
quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0), quicer:async_shutdown_connection(Conn, ?QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0),
@ -185,15 +160,13 @@ ensure_ok_or_exit(Fun, Args = [Sock | _]) when is_atom(Fun), is_list(Args) ->
async_send({quic, _Conn, Stream, _Info}, Data, _Options) -> async_send({quic, _Conn, Stream, _Info}, Data, _Options) ->
case quicer:async_send(Stream, Data, ?QUICER_SEND_FLAG_SYNC) of case quicer:async_send(Stream, Data, ?QUICER_SEND_FLAG_SYNC) of
{ok, _Len} -> ok; {ok, _Len} -> ok;
{error, X, Y} -> {error, {X, Y}};
Other -> Other Other -> Other
end. end.
%%% %%%
%%% quicer stream callbacks %%% quicer stream callbacks
%%% %%%
-spec new_stream(stream_handle(), quicer:new_stream_props(), cb_data()) -> cb_ret().
new_stream(_Stream, #{flags := _Flags, is_orphan := _IsOrphan}, _Conn) ->
{stop, unimpl}.
-spec peer_receive_aborted(stream_handle(), non_neg_integer(), cb_data()) -> cb_ret(). -spec peer_receive_aborted(stream_handle(), non_neg_integer(), cb_data()) -> cb_ret().
peer_receive_aborted(Stream, ErrorCode, S) -> peer_receive_aborted(Stream, ErrorCode, S) ->
@ -222,28 +195,12 @@ send_complete(_Stream, true = _IsCancelled, S) ->
send_shutdown_complete(_Stream, _IsGraceful, S) -> send_shutdown_complete(_Stream, _IsGraceful, S) ->
{ok, S}. {ok, S}.
%% Local stream, Unidir
%% -spec handle_stream_data(stream_handle(), binary(), quicer:recv_data_props(), cb_data())
%% -> cb_ret().
%% handle_stream_data(Stream, Bin, Flags, #{ is_local := true
%% , parse_state := PS} = S) ->
%% ?SLOG(debug, #{data => Bin}, Flags),
%% case parse(Bin, PS, []) of
%% {keep_state, NewPS, Packets} ->
%% quicer:setopt(Stream, active, once),
%% {keep_state, S#{parse_state := NewPS},
%% [{next_event, cast, P } || P <- lists:reverse(Packets)]};
%% {stop, _} = Stop ->
%% Stop
%% end;
%% %% Remote stream
%% handle_stream_data(_Stream, _Bin, _Flags,
%% #{is_local := false, is_unidir := true, conn := _Conn} = _S) ->
%% {stop, unimpl}.
-spec passive(stream_handle(), undefined, cb_data()) -> cb_ret(). -spec passive(stream_handle(), undefined, cb_data()) -> cb_ret().
passive(Stream, undefined, S) -> passive(Stream, undefined, S) ->
quicer:setopt(Stream, active, 10), case quicer:setopt(Stream, active, 10) of
ok -> ok;
Error -> ?SLOG(error, #{message => "set active error", error => Error})
end,
{ok, S}. {ok, S}.
-spec stream_closed(stream_handle(), quicer:stream_closed_props(), cb_data()) -> cb_ret(). -spec stream_closed(stream_handle(), quicer:stream_closed_props(), cb_data()) -> cb_ret().
@ -277,3 +234,7 @@ stream_closed(
-spec socket(connection_handle(), stream_handle(), socket_info()) -> socket(). -spec socket(connection_handle(), stream_handle(), socket_info()) -> socket().
socket(Conn, CtrlStream, Info) when is_map(Info) -> socket(Conn, CtrlStream, Info) when is_map(Info) ->
{quic, Conn, CtrlStream, Info}. {quic, Conn, CtrlStream, Info}.
%% BUILD_WITHOUT_QUIC
-else.
-endif.

1706
apps/emqx/test/emqtt_quic_SUITE.erl
View File

File diff suppressed because it is too large Load Diff

13
apps/emqx/test/emqx_mqtt_protocol_v5_SUITE.erl
View File

@ -79,19 +79,6 @@ end_per_group(_Group, _Config) ->
init_per_suite(Config) -> init_per_suite(Config) ->
%% Start Apps %% Start Apps
dbg:tracer(process, {fun dbg:dhandler/2, group_leader()}),
dbg:p(all, c),
dbg:tp(emqx_quic_connection, cx),
dbg:tp(quicer_connection, cx),
%% dbg:tp(emqx_quic_stream, cx),
%% dbg:tp(emqtt_quic, cx),
%% dbg:tp(emqtt, cx),
%% dbg:tp(emqtt_quic_stream, cx),
%% dbg:tp(emqtt_quic_connection, cx),
%% dbg:tp(emqx_cm, open_session, cx),
%% dbg:tpl(emqx_cm, lookup_channels, cx),
%% dbg:tpl(emqx_cm, register_channel, cx),
%% dbg:tpl(emqx_cm, unregister_channel, cx),
emqx_common_test_helpers:boot_modules(all), emqx_common_test_helpers:boot_modules(all),
emqx_common_test_helpers:start_apps([]), emqx_common_test_helpers:start_apps([]),
Config. Config.

1880
apps/emqx/test/emqx_quic_multistreams_SUITE.erl
View File

File diff suppressed because it is too large Load Diff