Merge pull request #31 from emqx/chore/EMQX-13447/otp-27-gun-2.1

chore: update to use `gun` 2.1.0 + Erlang/OTP 27 readiness

.github/workflows/run_test_case.yaml

@@ -8,11 +8,11 @@ jobs:
     runs-on: windows-latest
 
     steps:
-    - uses: actions/checkout@v1
+    - uses: actions/checkout@v4
     - uses: erlef/setup-beam@v1
       with:
-        otp-version: '24'
-        rebar3-version: '3.16.1'
+        otp-version: '27'
+        rebar3-version: '3.23.0'
     - run: rebar3 xref
     - run: rebar3 dialyzer
     - run: rebar3 eunit
@@ -23,46 +23,49 @@ jobs:
 
     strategy:
       matrix:
+        builder:
+          - ghcr.io/emqx/emqx-builder/5.4-3:1.17.3-27.2-1
         os:
+          - ubuntu24.04
+          - ubuntu22.04
           - ubuntu20.04
-          - ubuntu18.04
-          - ubuntu16.04
+          - debian12
+          - debian11
           - debian10
-          - debian9
-          - centos8
-          - centos7
-          - raspbian10
+          - amzn2023
+          - el9
 
     steps:
-    - uses: actions/checkout@v1
+    - uses: actions/checkout@v4
     - name: Code analyze
-      env:
-        ERL_OTP: erl24.3.4.2-1
-        SYSTEM: ${{ matrix.os }}
       run: |
         version=$(echo ${{ github.ref }} | sed -r  "s .*/.*/(.*) \1 g")
-        sudo docker run --rm --privileged multiarch/qemu-user-static:register --reset
-        sudo docker run --rm -i --name $SYSTEM -v $(pwd):/repos emqx/build-env:$ERL_OTP-$SYSTEM sh -c "cd /repos && make xref && make dialyzer"
+        docker run --rm --privileged multiarch/qemu-user-static:register --reset
+        docker run --rm -i --name ${{ matrix.os }} -v $(pwd):/repos ${{ matrix.builder }}-${{ matrix.os }} \
+          sh -c "cd /repos && make xref && make dialyzer"
     - name: Run tests
-      env:
-        ERL_OTP: erl24.3.4.2-1
-        SYSTEM: ${{ matrix.os }}
       run: |
         version=$(echo ${{ github.ref }} | sed -r  "s .*/.*/(.*) \1 g")
-        sudo docker run --rm --privileged multiarch/qemu-user-static:register --reset
-        sudo docker run --rm -i --name $SYSTEM -v $(pwd):/repos emqx/build-env:$ERL_OTP-$SYSTEM sh -c "cd /repos && make eunit && make ct"
+        docker run --rm --privileged multiarch/qemu-user-static:register --reset
+        docker run --rm -i --name ${{ matrix.os }} -v $(pwd):/repos ${{ matrix.builder }}-${{ matrix.os }} \
+          sh -c "cd /repos && make eunit && make ct"
 
   docker:
     runs-on: ubuntu-latest
 
+    strategy:
+      matrix:
+        builder:
+          - ghcr.io/emqx/emqx-builder/5.4-3:1.17.3-27.2-1
+
     steps:
-    - uses: actions/checkout@v1
+    - uses: actions/checkout@v4
     - name: Code analyze
-      env:
-        ERL_OTP: erl24.3.4.2-1
-      run: docker run --rm -i --name alpine -v $(pwd):/repos emqx/build-env:$ERL_OTP-alpine sh -c "cd /repos && make xref && make dialyzer"
+      run: |
+        docker run --rm -i --name alpine -v $(pwd):/repos ${{ matrix.builder }}-alpine3.15.1 \
+          sh -c "cd /repos && make xref && make dialyzer"
     - name: Run tests
-      env:
-        ERL_OTP: erl24.3.4.2-1
-      run: docker run --rm -i --name alpine -v $(pwd):/repos emqx/build-env:$ERL_OTP-alpine sh -c "cd /repos && make eunit && make ct"
+      run: |
+        docker run --rm -i --name alpine -v $(pwd):/repos ${{ matrix.builder }}-alpine3.15.1 \
+          sh -c "cd /repos && make eunit && make ct"
 

rebar.config

@@ -3,7 +3,7 @@
 {erl_opts, [debug_info]}.
 
 {plugins,
- [ {grpc_plugin, {git, "https://github.com/HJianBo/grpc_plugin", {tag, "v0.10.2"}}}
+ [ {grpc_plugin, {git, "https://github.com/HJianBo/grpc_plugin", {tag, "v0.10.3"}}}
 ]}.
 
 {grpc,
@@ -20,10 +20,10 @@
  ]}.
 
 {deps,
- [ {gpb, "~> 4.11"}
- , {gproc, "0.8.0"}
+ [ {gpb, "~> 4.21"}
+ , {gproc, "1.0.0"}
  , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.0"}}}
- , {gun, {git, "https://github.com/emqx/gun", {tag, "1.3.7"}}}
+ , {gun, "2.1.0"}
  ]}.
 
 {xref_checks,
@@ -34,6 +34,8 @@
 {xref_ignores, [grpc_health_pb,grpc_reflection_pb,
                 grpc_greeter_pb,grpc_route_guide_pb]}.
 
+{dialyzer, [{plt_apps, all_deps}]}.
+
 {profiles,
  [{test,
    [{cover_enabled, true},

src/client/grpc_client.erl

@@ -42,6 +42,9 @@
         , code_change/3
         ]).
 
+-export_type([ options/0
+             , grpcstream/0]).
+
 -record(state, {
           %% Pool name
           pool,
@@ -59,7 +62,7 @@
           %% XXX: Bad impl.
           encoding :: grpc_frame:encoding(),
           %% Streams
-          streams :: #{reference() := stream()},
+          streams :: #{gun:stream_ref() => stream()},
           %% Client options
           client_opts :: client_opts(),
           %% Flush timer reference
@@ -118,7 +121,7 @@
         #{protocols => [http2],
           connect_timeout => 5000,
           http2_opts => #{keepalive => 60000},
-          transport_opts => [{nodelay, true}]
+          tcp_opts => [{nodelay, true}]
          }).
 
 -define(STREAM_RESERVED_TIMEOUT, 15000).
@@ -387,7 +390,7 @@ handle_info({timeout, TRef, flush_streams_sendbuff},
 handle_info({gun_up, GunPid, http2}, State = #state{gun_pid = GunPid}) ->
     {noreply, State};
 
-handle_info({gun_down, GunPid, http2, Reason, KilledStreamRefs, _},
+handle_info({gun_down, GunPid, http2, Reason, KilledStreamRefs},
             State = #state{gun_pid = GunPid, streams = Streams}) ->
     Nowts = erlang:system_time(millisecond),
     %% Reply killed streams error
@@ -633,45 +636,24 @@ do_connect(State = #state{server = {_, Host, Port}, client_opts = ClientOpts}) -
     GunOpts = maps:get(gun_opts, ClientOpts, #{}),
     case gun:open(Host, Port, GunOpts) of
         {ok, Pid} ->
-            case gun_await_up_helper(Pid) of
+            %% NOTE
+            %% By default, `gun` retries failed connection attempts 5 times, with
+            %% 5 seconds delay in-between. Give it a bit more spare time, just in
+            %% case.
+            MRef = monitor(process, Pid),
+            case gun:await_up(Pid, 60_000, MRef) of
                 {ok, _Protocol} ->
-                    MRef = monitor(process, Pid),
                     State#state{mref = MRef, gun_pid = Pid};
-                {error, Reason} ->
-                    gun:close(Pid),
-                    {error, Reason}
+                {error, {down, Reason}} ->
+                    {error, Reason};
+                {error, timeout} ->
+                    _ = gun:close(Pid),
+                    {error, timeout}
             end;
         {error, Reason} ->
             {error, Reason}
     end.
 
-gun_await_up_helper(Pid) ->
-    gun_await_up_helper(Pid, 50, undefined).
-gun_await_up_helper(_Pid, 0, LastRet) ->
-   LastRet ;
-gun_await_up_helper(Pid, Retry, LastRet) ->
-    case gun:await_up(Pid, 100) of
-        {ok, _} = Ret ->
-            Ret;
-        {error, timeout} ->
-            case gun_last_reason(Pid) of
-                undefined ->
-                    gun_await_up_helper(Pid, Retry-1, LastRet);
-                Reason ->
-                    {error, Reason}
-            end;
-        {error, _} = Ret ->
-            Ret
-    end.
-
-gun_last_reason(Pid) ->
-    %% XXX: Hard-coded to get detailed reason, because gun
-    %% does not expose it with a function
-    lists:last(
-      tuple_to_list(
-        element(2, sys:get_state(Pid)))
-     ).
-
 %%--------------------------------------------------------------------
 %% Helpers
 

src/client/grpc_client_sup.erl

@@ -32,7 +32,7 @@
 
 -type name() :: term().
 
--type options() :: grpc_client:client_opts()
+-type options() :: grpc_client:options()
                  | #{pool_size => non_neg_integer()}.
 
 %%--------------------------------------------------------------------

src/grpc.app.src

@@ -4,7 +4,7 @@
   {vsn, "git"},
   {registered, []},
   {mod, {grpc_app, []}},
-  {applications, [kernel,stdlib,gproc,cowboy,gun]},
+  {applications, [kernel,stdlib,ssl,public_key,gproc,cowboy,gun]},
   {env,[]},
   {modules, []},
   {licenses, ["Apache 2.0"]},

src/grpc.erl

@@ -29,16 +29,25 @@
 
 -type listen_on() :: {inet:ip_address(), inet:port_number()} | inet:port_number().
 -type services() :: #{protos := [module()],
-                      services := #{ServiceName :: atom() := HandlerModule :: module()}
+                      services := #{ServiceName :: atom() => HandlerModule :: module()}
                      }.
 
--type option() :: {ssl_options, ssl:server_option()}
+-type option() :: {ssl_options, ssl:tls_server_option()}
                 | {ranch_opts, ranch:opts()}
                 | {cowboy_opts, cowboy_http2:opts()}.
 
--type metadata() :: map().
+%% TODO: Figure out types.
+-type metadata_key() :: term().
+-type metadata_value() :: term().
+-type metadata() :: #{metadata_key() => metadata_value()}.
 
--export_type([metadata/0]).
+%% NOTE: Expected by generated code.
+-type options() :: grpc_client:options().
+
+-export_type([ metadata/0
+             , metadata_key/0
+             , metadata_value/0
+             , options/0]).
 
 %%--------------------------------------------------------------------
 %% APIs

src/grpc_frame.erl

@@ -22,6 +22,8 @@
         , split/2
         ]).
 
+-export_type([encoding/0]).
+
 -type encoding() :: identity | gzip.
 
 -define(GRPC_ERROR(Status, Message), {grpc_error, {Status, Message}}).

src/grpc_lib.erl

@@ -98,7 +98,7 @@ decode(Base64) when byte_size(Base64) rem 4 == 2 ->
 decode(Base64) ->
     base64:decode(Base64).
 
--spec maybe_encode_headers(grpc:meta_data()) -> grpc:meta_data().
+-spec maybe_encode_headers(grpc:metadata()) -> grpc:metadata().
 %% @doc Encode the header values to Base64 for those headers that have the name 
 %% ending with "-bin".
 maybe_encode_headers(Headers) ->

src/grpc_stream.erl

@@ -35,11 +35,13 @@
         , handle_out/3
         ]).
 
+-export_type([stream/0, error_response/0]).
+
 -type stream() :: #{ req           := cowboy_req:req()
                    , rest          := binary()
                    , metadata      := map()
                    , encoding      := grpc_frame:encoding()
-                   , compression   := grpc_frame:compression()
+                   , compression   := grpc_frame:encoding() %% TODO: Figure out types
                    , decoder       := function()
                    , encoder       := function()
                    , handler       := {atom(), atom()}
@@ -49,6 +51,9 @@
                    , client_info   := map()
                    }.
 
+%% TODO: Figure out types.
+-type error_response() :: term().
+
 %%--------------------------------------------------------------------
 %% APIs
 %%--------------------------------------------------------------------

test/grpc_SUITE.erl

@@ -61,7 +61,7 @@ init_per_group(GrpName, Cfg) ->
                     https ->
                         #{gun_opts =>
                           #{transport => ssl,
-                            transport_opts => [{cacertfile, CA}]}};
+                            tls_opts => [{cacertfile, CA}, {verify, verify_none}]}};
                     _ -> #{}
                 end,
     SvrAddr = case GrpName of