grpcbox-stream: Fix to support multiple simultaneous unary/strem calls

src/grpcbox_channel.erl

@@ -5,13 +5,16 @@
 -export([start_link/3,
          is_ready/1,
          pick/2,
-         stop/1]).
+         stop/1,
+         add_channel/3,
+         delete_channel/1]).
 -export([init/1,
          callback_mode/0,
          terminate/3,
          connected/3,
          idle/3]).
 
+-include_lib("stdlib/include/ms_transform.hrl").
 -include("grpcbox.hrl").
 
 -define(CHANNEL(Name), {via, gproc, {n, l, {?MODULE, Name}}}).
@@ -45,6 +48,16 @@
                stats_handler :: module() | undefined,
                refresh_interval :: timer:time()}).
 
+%% @doc add a new channel
+-spec add_channel(name(), [endpoint()], options()) ->  {ok, pid()}.
+add_channel(Name, Endpoints, Options) ->
+    grpcbox_channel_sup:start_child(Name, Endpoints, Options).
+
+%% @doc Delete a channel
+-spec delete_channel(pid()) -> any().
+delete_channel(Pid) when is_pid(Pid) ->
+    ok = supervisor:terminate_child(grpcbox_channel_sup, Pid).
+
 -spec start_link(name(), [endpoint()], options()) -> {ok, pid()}.
 start_link(Name, Endpoints, Options) ->
     gen_statem:start_link(?CHANNEL(Name), ?MODULE, [Name, Endpoints, Options], []).
@@ -129,6 +142,12 @@ handle_event(_, _, Data) ->
     {keep_state, Data}.
 
 terminate(_Reason, _State, #data{pool=Name}) ->
+    [ets:delete(?CHANNELS_TAB, Key) || Key <-
+        ets:select(?CHANNELS_TAB, ets:fun2ms(fun
+            ({{N, V}, _}) when N == Name  ->
+                {N, V}
+        end))
+    ],
     gproc_pool:force_delete(Name),
     ok.
 

src/grpcbox_stream.erl

@@ -8,6 +8,7 @@
 
 -export([send/2,
          send/3,
+         is_alive/1,
          send_headers/2,
          add_headers/2,
          add_trailers/2,
@@ -423,6 +424,12 @@ update_headers(Headers, State=#state{resp_headers=RespHeaders}) ->
 update_trailers(Trailers, State=#state{resp_trailers=RespTrailers}) ->
     State#state{resp_trailers=RespTrailers ++ Trailers}.
 
+is_alive(#state{socket = Socket}) ->
+    case sock:peername(Socket) of
+        {ok,_} -> true;
+        _ -> false
+    end.
+
 send(Message, #state{handler=Pid}) ->
     Pid ! {send_proto, Message}.
 

test/grpcbox_SUITE.erl

@@ -28,6 +28,7 @@ all() ->
      chain_interceptor,
      stream_interceptor,
      bidirectional,
+     stress_test,
      client_stream,
      compression,
      stats_handler,
@@ -189,6 +190,15 @@ init_per_testcase(bidirectional, Config) ->
                                           services => #{'routeguide.RouteGuide' => routeguide_route_guide}}}]),
     application:ensure_all_started(grpcbox),
     Config;
+init_per_testcase(stress_test, Config) ->
+    application:load(grpcbox),
+    application:set_env(grpcbox, client, #{channels => [{default_channel,
+                                                         [{http, "localhost", 8080, []}], #{}}]}),
+    application:set_env(grpcbox, servers,
+                        [#{grpc_opts => #{service_protos => [route_guide_pb],
+                                          services => #{'routeguide.RouteGuide' => routeguide_route_guide}}}]),
+    application:ensure_all_started(grpcbox),
+    Config;
 init_per_testcase(client_stream, Config) ->
     application:set_env(grpcbox, client, #{channels => [{default_channel,
                                                          [{http, "localhost", 8080, []}], #{}}]}),
@@ -467,8 +477,8 @@ multiple_servers(_Config) ->
     unary(_Config),
     unary(_Config).
 
-bidirectional(_Config) ->
-    {ok, S} = routeguide_route_guide_client:route_chat(ctx:new()),
+bidirectional(Config) ->
+    {ok, S} = routeguide_route_guide_client:route_chat(ctx:new(), proplists:get_value(options, Config, #{})),
     %% send 2 before receiving since the server only sends what it already had in its list of messages for the
     %% location of your last send.
     ok = grpcbox_client:send(S, #{location => #{latitude => 1, longitude => 1}, message => <<"hello there">>}),
@@ -540,6 +550,56 @@ stream_interceptor(_Config) ->
     ?assertMatch({ok, #{name := <<"Louvre">>}}, grpcbox_client:recv_data(Stream)),
     ?assertMatch({ok, {_, _, #{<<"x-grpc-stream-interceptor">> := <<"true">>}}}, grpcbox_client:recv_trailers(Stream)).
 
+stress_test_function(Fun, Config, Ref, Parent) ->
+    Parent ! {stress_test, Ref, Fun(Config)}.
+
+stress_test(Config) ->
+    stress_test(Config,
+        erlang:list_to_integer(
+            os:getenv("GRPCBOX_STRESS_TEST", "10")
+        )).
+
+stress_test(Config, Count) ->
+    lists:foreach(fun
+        (ProcId) ->
+            Parent = self(),
+            spawn(fun() ->
+                Channel = erlang:list_to_atom("proc_" ++ erlang:integer_to_list(ProcId)),
+                erlang:register(Channel, self()),
+                {ok, _Pid} = grpcbox_channel:add_channel(
+                    Channel,
+                    [{http, "localhost", 8080, []}],
+                    #{}
+                ),
+                lists:foldl(fun
+                    (_, not_ready) ->
+                        timer:sleep(10),
+                        grpcbox_channel:is_ready(Channel);
+                    (_,Acc) ->
+                        Acc
+                end, not_ready, lists:seq(1, 100)),
+
+                stress_test_function(fun bidirectional/1,
+                    [{options,#{channel => Channel}} | Config],
+                    ProcId, Parent),
+                ok
+                %% grpcbox_channel:delete_channel(Pid)
+            end)
+    end, lists:seq(1, Count)),
+
+    Loop = fun Loop(LoopCount) ->
+        receive
+            {stress_test, _Ref, _Reply} when LoopCount < Count ->
+                Loop(LoopCount + 1);
+            {stress_test, _Ref, _Reply} when LoopCount < Count ->
+                LoopCount + 1
+        after
+            2000 ->
+                LoopCount
+        end
+    end,
+    ?assertEqual(Count, Loop(0)).
+
 %%
 
 cert_dir(Config) ->

test/routeguide_route_guide.erl

@@ -23,10 +23,6 @@
     #{lo => point(),
       hi => point()}.
 
--type route_note() ::
-    #{location => point(),
-      message => string()}.
-
 -type feature() ::
     #{name => string(),
       location => point()}.