feat(api): get owners

Get - listener owner - connection owner - stream owner
emqx · Nov 24, 2023 · 323d504 · 323d504
1 parent bcc799d
commit 323d504
Show file tree

Hide file tree

Showing 12 changed files with 225 additions and 1 deletion.
diff --git a/c_src/quicer_connection.c b/c_src/quicer_connection.c
@@ -1731,6 +1731,30 @@ get_connectionsX(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
   return res;
 }
 
+ERL_NIF_TERM
+get_conn_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerConnCTX *c_ctx = NULL;
+  ERL_NIF_TERM res = ATOM_UNDEFINED;
+  CXPLAT_FRE_ASSERT(argc == 1);
+  if (!enif_get_resource(env, argv[0], ctx_connection_t, (void **)&c_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+
+  enif_mutex_lock(c_ctx->lock);
+  if (c_ctx->owner == NULL)
+    {
+      res = ERROR_TUPLE_2(ATOM_UNDEFINED);
+      goto exit;
+    }
+  res = SUCCESS(enif_make_pid(env, &(c_ctx->owner->Pid)));
+  ;
+exit:
+  enif_mutex_unlock(c_ctx->lock);
+  return res;
+}
+
 ///_* Emacs
 ///====================================================================
 /// Local Variables:

diff --git a/c_src/quicer_connection.h b/c_src/quicer_connection.h
@@ -57,4 +57,7 @@ peercert1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM
 get_connectionsX(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 
+ERL_NIF_TERM
+get_conn_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
+
 #endif // __QUICER_CONNECTION_H_
diff --git a/c_src/quicer_listener.c b/c_src/quicer_listener.c
@@ -648,3 +648,19 @@ get_listenersX(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
     }
   return res;
 }
+
+ERL_NIF_TERM
+get_listener_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerListenerCTX *l_ctx;
+  ERL_NIF_TERM res = ATOM_UNDEFINED;
+  CXPLAT_FRE_ASSERT(argc == 1);
+  if (!enif_get_resource(env, argv[0], ctx_listener_t, (void **)&l_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+  enif_mutex_lock(l_ctx->lock);
+  res = SUCCESS(enif_make_pid(env, &(l_ctx->listenerPid)));
+  enif_mutex_unlock(l_ctx->lock);
+  return res;
+}
diff --git a/c_src/quicer_listener.h b/c_src/quicer_listener.h
@@ -38,4 +38,7 @@ close_listener1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 ERL_NIF_TERM
 get_listenersX(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
 
+ERL_NIF_TERM
+get_listener_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
+
 #endif // __QUICER_LISTENER_H_
diff --git a/c_src/quicer_nif.c b/c_src/quicer_nif.c
@@ -1571,6 +1571,9 @@ static ErlNifFunc nif_funcs[] = {
   { "get_listeners", 1, get_listenersX, 0},
   { "get_connections", 0, get_connectionsX, 0},
   { "get_connections", 1, get_connectionsX, 0},
+  { "get_conn_owner", 1, get_conn_owner1, 0},
+  { "get_stream_owner", 1, get_stream_owner1, 0},
+  { "get_listener_owner", 1, get_listener_owner1, 0}
   // clang-format on
 };
 

diff --git a/c_src/quicer_stream.c b/c_src/quicer_stream.c
@@ -1247,6 +1247,29 @@ reset_stream_recv(QuicerStreamCTX *s_ctx)
   s_ctx->TotalBufferLength = 0;
 }
 
+ERL_NIF_TERM
+get_stream_owner1(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[])
+{
+  QuicerStreamCTX *s_ctx;
+  ERL_NIF_TERM res = ATOM_UNDEFINED;
+  CXPLAT_FRE_ASSERT(argc == 1);
+  if (!enif_get_resource(env, argv[0], ctx_stream_t, (void **)&s_ctx))
+    {
+      return ERROR_TUPLE_2(ATOM_BADARG);
+    }
+
+  enif_mutex_lock(s_ctx->lock);
+  if (!s_ctx->owner)
+    {
+      res = ERROR_TUPLE_2(ATOM_BADARG);
+      goto exit;
+    }
+  res = SUCCESS(enif_make_pid(env, &(s_ctx->owner->Pid)));
+exit:
+  enif_mutex_unlock(s_ctx->lock);
+  return res;
+}
+
 ///_* Emacs
 ///====================================================================
 /// Local Variables:

diff --git a/c_src/quicer_stream.h b/c_src/quicer_stream.h
@@ -63,3 +63,6 @@ _IRQL_requires_max_(DISPATCH_LEVEL)
 ERL_NIF_TERM
 get_stream_rid1(ErlNifEnv *env, int args, const ERL_NIF_TERM argv[]);
 #endif // __QUICER_STREAM_H_
+
+ERL_NIF_TERM
+get_stream_owner1(ErlNifEnv *env, int args, const ERL_NIF_TERM argv[]);
diff --git a/src/quicer.erl b/src/quicer.erl
@@ -117,6 +117,9 @@
         , get_connections/0
         , get_connections/1
         , close_registration/1
+        , get_conn_owner/1
+        , get_stream_owner/1
+        , get_listener_owner/1
         ]).
 
 -export([ spawn_listener/3 %% start application over quic
@@ -1005,6 +1008,18 @@ get_connections(global) ->
 get_connections(Reg) ->
   quicer_nif:get_connections(Reg).
 
+-spec get_conn_owner(C) -> quicer_nif:get_conn_owner(C).
+get_conn_owner(Conn) ->
+  quicer_nif:get_conn_owner(Conn).
+
+-spec get_stream_owner(S) -> quicer_nif:get_stream_owner(S).
+get_stream_owner(Stream) ->
+  quicer_nif:get_stream_owner(Stream).
+
+-spec get_listener_owner(L) -> quicer_nif:get_listener_owner(L).
+get_listener_owner(Listener) ->
+  quicer_nif:get_listener_owner(Listener).
+
 %% @doc set controlling process for Connection/Stream.
 %% mimic {@link ssl:controlling_process/2}
 %% @see wait_for_handoff/2

diff --git a/src/quicer_nif.erl b/src/quicer_nif.erl
@@ -57,6 +57,9 @@
         , get_listeners/1
         , get_connections/0
         , get_connections/1
+        , get_conn_owner/1
+        , get_stream_owner/1
+        , get_listener_owner/1
         ]).
 
 -export([abi_version/0]).
@@ -307,6 +310,18 @@ controlling_process(_H, _P) ->
 peercert(_Handle) ->
   erlang:nif_error(nif_library_not_loaded).
 
+-spec get_conn_owner(connection_handle()) -> {ok, pid()} | {error, undefined | badarg}.
+get_conn_owner(_) ->
+  erlang:nif_error(nif_library_not_loaded).
+
+-spec get_stream_owner(connection_handle()) -> {ok, pid()} | {error, undefined | badarg}.
+get_stream_owner(_) ->
+  erlang:nif_error(nif_library_not_loaded).
+
+-spec get_listener_owner(listener_handle()) -> {ok, pid()} | {error, undefined | badarg}.
+get_listener_owner(_) ->
+  erlang:nif_error(nif_library_not_loaded).
+
 -spec get_listeners() -> [listener_handle()].
 get_listeners() ->
   erlang:nif_error(nif_library_not_loaded).

diff --git a/test/quicer_SUITE.erl b/test/quicer_SUITE.erl
@@ -66,6 +66,8 @@
         , tc_stream_active_switch_to_passive/1
         , tc_stream_controlling_process/1
         , tc_stream_controlling_process_demon/1
+        , tc_stream_get_owner_local/1
+        , tc_stream_get_owner_remote/1
 
         , tc_dgram_client_send/1
 
@@ -82,6 +84,7 @@
         , tc_getopt_global_lib_git_hash/1
         , tc_getopt_stream_active/1
         , tc_setopt/1
+        , tc_setopt_remote_addr/1
         , tc_getopt_settings/1
 
         %% @TODO following two tcs are failing due to:
@@ -1346,6 +1349,12 @@ tc_setopt(Config) ->
     ct:fail("listener_timeout")
   end.
 
+tc_setopt_remote_addr(_Config) ->
+  {ok, Conn} = quicer:open_connection(),
+  ok = quicer:setopt(Conn, param_conn_remote_address, "8.8.8.8:443"),
+  ?assertEqual({ok, {{8,8,8,8}, 443}}, quicer:getopt(Conn, param_conn_remote_address)),
+  quicer:shutdown_connection(Conn).
+
 tc_setopt_bad_opt(_Config)->
   Port = select_port(),
   {error, badarg} = quicer:connect("localhost", Port,
@@ -2579,9 +2588,84 @@ tc_peercert_server_nocert(Config) ->
   ensure_server_exit_normal(Ref),
   ok.
 
-tc_abi_version(Config) ->
+tc_abi_version(_Config) ->
   ?assertEqual(1, quicer:abi_version()).
 
+tc_stream_get_owner_local(Config) ->
+  Port = select_port(),
+  Owner = self(),
+  {SPid, Ref} = spawn_monitor(
+                   fun() ->
+                       simple_conn_server(Owner, Config, Port)
+                   end),
+  receive listener_ready -> ok end,
+  {ok, Conn} = quicer:connect("localhost", Port,
+                              default_conn_opts(),
+                              5000),
+  {ok, {_, _}} = quicer:sockname(Conn),
+  {ok, Conn} = quicer:async_accept_stream(Conn, []),
+  {ok, Stm} = quicer:async_csend(Conn, <<"hello">>, [{active, true}], ?QUIC_SEND_FLAG_START),
+  ?assertEqual({ok, self()}, quicer:get_stream_owner(Stm)),
+  ok = quicer:close_stream(Stm),
+  _ = quicer:close_connection(Conn),
+  SPid ! done,
+  ensure_server_exit_normal(Ref),
+  ok.
+
+tc_stream_get_owner_remote(Config) ->
+  Port = select_port(),
+  Owner = self(),
+  {SPid, Ref} = spawn_monitor(
+                  fun() ->
+                      echo_server(Owner, Config, Port)
+                  end),
+  receive
+    listener_ready ->
+      ok
+  after 5000 ->
+    ct:fail("listener_timeout")
+  end,
+  {ok, Conn} = quicer:connect("127.0.0.1", Port,
+                              default_conn_opts() ++ [{peer_unidi_stream_count, 1}], 5000),
+  {ok, Stm0} = quicer:start_stream(Conn, [{active, true},
+                                          {start_flag, ?QUIC_STREAM_START_FLAG_INDICATE_PEER_ACCEPT},
+                                          {open_flag, ?QUIC_STREAM_OPEN_FLAG_UNIDIRECTIONAL}
+                                         ]),
+  {ok, 5} = quicer:send(Stm0, <<"ping1">>),
+  receive
+    {quic, <<"ping1">>, Stm0, _} ->
+      ct:fail("We should not recv ping1 due to flow control: bidir stream 0")
+  after 1000 ->
+      ct:pal("recv ping1 timeout"),
+      SPid ! {flow_ctl, 10, 1}
+  end,
+  quicer:async_accept_stream(Conn, []),
+  %% check with server if peer addr is correct.
+  receive
+    {quic, peer_accepted, Stm0, undefined} ->
+      ct:pal("peer_accepted received")
+  after 1000 ->
+      ct:fail("peer_accepted timeout")
+  end,
+
+  %% Now we expect server initiat an Server -> Stream unidirectional stream
+  receive
+    {quic, new_stream, Stm1, #{ flags := Flags }} ->
+      ?assert(quicer:is_unidirectional(Flags)),
+      ?assertEqual({ok, self()}, quicer:get_stream_owner(Stm1)),
+      %% We also expect server send reply over new stream
+      receive
+        {quic, <<"ping1">>, Stm0, _} ->
+          ct:fail("Data recvd from client -> server unidirectional stream");
+        {quic, <<"ping1">>, Stm1, _} ->
+          ct:pal("Data recvd from server -> client unidirectional stream")
+      end
+  after 2000 ->
+      ct:fail("No new_stream for stream initiated from Server")
+  end,
+  SPid ! done,
+  ensure_server_exit_normal(Ref).
+
 %%% ====================
 %%% Internal helpers
 %%% ====================

diff --git a/test/quicer_connection_SUITE.erl b/test/quicer_connection_SUITE.erl
@@ -684,6 +684,35 @@ tc_conn_list(Config) ->
                lists:map(fun quicer:peername/1, Conns)),
   SPid ! done.
 
+tc_get_conn_owner_client(_Config) ->
+  {ok, Conn} = quicer:open_connection(),
+  {ok, Pid} = quicer:get_conn_owner(Conn),
+  quicer:close_connection(Conn),
+  ?assertEqual(self(), Pid).
+
+tc_get_conn_owner_server(Config) ->
+  Port = select_port(),
+  {ok, L} = quicer:listen(Port, default_listen_opts(Config)),
+  {ok, L} = quicer:async_accept(L, #{}),
+  {ClientPid, CMref} = erlang:spawn_monitor(fun()->
+                                                {ok, Conn} = quicer:connect("127.0.0.1", Port, default_conn_opts(), 1000),
+                                                {ok, _} = quicer:csend(Conn, <<"ping">>),
+                                                receive
+                                                  done ->
+                                                    quicer:connection_close(Conn),
+                                                    ok
+                                                end
+                                            end),
+  receive
+    {quic, new_conn, SConn, _} ->
+      {ok, Pid} = quicer:get_conn_owner(SConn),
+      ?assertEqual(self(), Pid),
+      quicer:close_connection(SConn),
+      quicer:close_listener(L),
+      ClientPid ! done;
+    {'DOWN', CMref, process, ClientPid, Reason} -> ct:fail({client_fail, Reason})
+  end.
+
 %%%
 %%% Helpers
 %%%

diff --git a/test/quicer_listener_SUITE.erl b/test/quicer_listener_SUITE.erl
@@ -503,6 +503,12 @@ tc_get_listeners_from_reg(Config) ->
   {ok, L2} = quicer:listen(Port2, default_listen_opts(Config)),
   ?assertEqual([L2, L1], quicer:get_listeners(RegH)).
 
+tc_get_listener_owner(Config) ->
+  Port = select_port(),
+  {ok, L} = quicer:listen(Port, default_listen_opts(Config)),
+  ?assertEqual({ok, self()}, quicer:get_listener_owner(L)),
+  quicer:close_listener(L).
+
 select_port() ->
   Port = select_free_port(quic),
   timer:sleep(100),