feat(quic): adapt to new quicer

apps/emqx/src/emqx_listeners.erl

@@ -178,9 +178,6 @@ do_start_listener(Type, ListenerName, #{bind := ListenOn} = Opts)
 do_start_listener(quic, ListenerName, #{bind := ListenOn} = Opts) ->
     case [ A || {quicer, _, _} = A<-application:which_applications() ] of
         [_] ->
-            %% @fixme unsure why we need reopen lib and reopen config.
-            quicer_nif:open_lib(),
-            quicer_nif:reg_open(),
             DefAcceptors = erlang:system_info(schedulers_online) * 8,
             ListenOpts = [ {cert, maps:get(certfile, Opts)}
                          , {key, maps:get(keyfile, Opts)}
@@ -195,7 +192,7 @@ do_start_listener(quic, ListenerName, #{bind := ListenOn} = Opts) ->
                               , zone => zone(Opts)
                               , listener => {quic, ListenerName}
                               },
-            StreamOpts = [],
+            StreamOpts = [{stream_callback, emqx_quic_stream}],
             quicer:start_listener(listener_id(quic, ListenerName),
                                   port(ListenOn), {ListenOpts, ConnectionOpts, StreamOpts});
         [] ->

apps/emqx/src/emqx_quic_connection.erl

@@ -17,8 +17,43 @@
 -module(emqx_quic_connection).
 
 %% Callbacks
--export([ new_conn/2
+-export([ init/1
+        , new_conn/2
+        , connected/2
+        , shutdown/2
         ]).
 
-new_conn(Conn, {_L, COpts, _S}) when is_map(COpts) ->
-    emqx_connection:start_link(emqx_quic_stream, Conn, COpts).
+-type cb_state() :: map() | proplists:proplist().
+
+
+-spec init(cb_state()) -> cb_state().
+init(ConnOpts) when is_list(ConnOpts) ->
+    init(maps:from_list(ConnOpts));
+init(ConnOpts) when is_map(ConnOpts) ->
+    ConnOpts.
+
+-spec new_conn(quicer:connection_handler(), cb_state()) -> {ok, cb_state()} | {error, any()}.
+new_conn(Conn, S) ->
+    case emqx_connection:start_link(emqx_quic_stream, Conn, S) of
+        {ok, _Pid} ->
+            ok = quicer:async_handshake(Conn),
+            {ok, S};
+        Other ->
+            {error, Other}
+    end.
+
+-spec connected(quicer:connection_handler(), cb_state()) -> {ok, cb_state()} | {error, any()}.
+connected(Conn, #{slow_start := false} = S) ->
+    case emqx_connection:start_link(emqx_quic_stream, Conn, S) of
+        {ok, _Pid} ->
+            {ok, S};
+        Other ->
+            {error, Other}
+    end;
+connected(_Conn, S) ->
+    {ok, S}.
+
+-spec shutdown(quicer:connection_handler(), cb_state()) -> {ok, cb_state()} | {error, any()}.
+shutdown(Conn, S) ->
+    quicer:async_close_connection(Conn),
+    {ok, S}.

apps/emqx/src/emqx_quic_stream.erl

@@ -88,5 +88,8 @@ ensure_ok_or_exit(Fun, Args = [Sock|_]) when is_atom(Fun), is_list(Args) ->
 async_send(Stream, Data, Options) when is_list(Data) ->
     async_send(Stream, iolist_to_binary(Data), Options);
 async_send(Stream, Data, _Options) when is_binary(Data) ->
-    {ok, _Len} = quicer:send(Stream, Data),
-    ok.
+    case quicer:send(Stream, Data) of
+        {ok, _Len} -> ok;
+        Other ->
+            Other
+    end.