refactor: better interface and simpler implementation for GRPC.Stream.run (#453)

* fix: GRPC.Stream.run only consumes the first reply and returns :ok

* fix: simplify noreply workflow

Author: polvalente

examples/helloworld/mix.lock

@@ -1,7 +1,10 @@
 %{
   "cowboy": {:hex, :cowboy, "2.12.0", "f276d521a1ff88b2b9b4c54d0e753da6c66dd7be6c9fca3d9418b561828a3731", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "8a7abe6d183372ceb21caa2709bec928ab2b72e18a3911aa1771639bef82651e"},
   "cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"},
+  "flow": {:hex, :flow, "1.2.4", "1dd58918287eb286656008777cb32714b5123d3855956f29aa141ebae456922d", [:mix], [{:gen_stage, "~> 1.0", [hex: :gen_stage, repo: "hexpm", optional: false]}], "hexpm", "874adde96368e71870f3510b91e35bc31652291858c86c0e75359cbdd35eb211"},
+  "gen_stage": {:hex, :gen_stage, "1.3.2", "7c77e5d1e97de2c6c2f78f306f463bca64bf2f4c3cdd606affc0100b89743b7b", [:mix], [], "hexpm", "0ffae547fa777b3ed889a6b9e1e64566217413d018cabd825f786e843ffe63e7"},
   "google_protos": {:hex, :google_protos, "0.3.0", "15faf44dce678ac028c289668ff56548806e313e4959a3aaf4f6e1ebe8db83f4", [:mix], [{:protobuf, "~> 0.10", [hex: :protobuf, repo: "hexpm", optional: false]}], "hexpm", "1f6b7fb20371f72f418b98e5e48dae3e022a9a6de1858d4b254ac5a5d0b4035f"},
+  "googleapis": {:hex, :googleapis, "0.1.0", "13770f3f75f5b863fb9acf41633c7bc71bad788f3f553b66481a096d083ee20e", [:mix], [{:protobuf, "~> 0.12", [hex: :protobuf, repo: "hexpm", optional: false]}], "hexpm", "1989a7244fd17d3eb5f3de311a022b656c3736b39740db46506157c4604bd212"},
   "gun": {:hex, :gun, "2.1.0", "b4e4cbbf3026d21981c447e9e7ca856766046eff693720ba43114d7f5de36e87", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "52fc7fc246bfc3b00e01aea1c2854c70a366348574ab50c57dfe796d24a0101d"},
   "hpax": {:hex, :hpax, "1.0.2", "762df951b0c399ff67cc57c3995ec3cf46d696e41f0bba17da0518d94acd4aac", [:mix], [], "hexpm", "2f09b4c1074e0abd846747329eaa26d535be0eb3d189fa69d812bfb8bfefd32f"},
   "jason": {:hex, :jason, "1.3.0", "fa6b82a934feb176263ad2df0dbd91bf633d4a46ebfdffea0c8ae82953714946", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "53fc1f51255390e0ec7e50f9cb41e751c260d065dcba2bf0d08dc51a4002c2ac"},

lib/grpc/server/adapters/cowboy/handler.ex

@@ -561,9 +561,11 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do
   end
 
   defp do_call_rpc(server, path, %{http_method: http_method} = stream) do
-    result = server.__call_rpc__(path, http_method, stream)
+    case server.__call_rpc__(path, http_method, stream) do
+      {:ok, stream, :noreply} ->
+        GRPC.Server.send_trailers(stream, @default_trailers)
+        {:ok, stream}
 
-    case result do
       {:ok, stream, response} ->
         stream
         |> GRPC.Server.send_reply(response)

lib/grpc/stream.ex

@@ -71,7 +71,7 @@ defmodule GRPC.Stream do
     - `:dispatcher` — Specifies the `Flow` dispatcher (defaults to `GenStage.DemandDispatcher`).
     - `:propagate_context` - If `true`, the context from the `materializer` is propagated to the `Flow`.
     - `:materializer` - The `%GRPC.Server.Stream{}` struct representing the current gRPC stream context.
-    
+
   And any other options supported by `Flow`.
 
   ## Returns
@@ -134,30 +134,33 @@ defmodule GRPC.Stream do
   def to_flow(%__MODULE__{flow: flow}), do: flow
 
   @doc """
-  Executes the underlying `Flow` for unary streams and emits responses into the provided gRPC server stream.
-
-  ## Parameters
-
-    - `flow`: A `GRPC.Stream` struct containing the flow to be executed.
-    - `stream`: A `GRPC.Server.Stream` to which responses are sent.
-    - `:dry_run` — If `true`, responses are not sent (used for testing or inspection).
+  Executes the underlying `Flow` for a unary stream.
 
-  ## Example
+  The response will be emitted automatically to the provided
+  `:materializer` (set to a `GRPC.Server.Stream`) for the single resulting
+  item in the materialized enumerable.
 
-      GRPC.Stream.run(request)
+  The `stream` argument must be initialized as a `:unary` stream with
+  a `:materializer` set.
   """
-  @spec run(t()) :: any()
+  @spec run(stream :: t()) :: :noreply
   def run(%__MODULE__{flow: flow, options: opts}) do
-    if !Keyword.get(opts, :unary, false) do
-      raise ArgumentError, "run/2 is not supported for non-unary streams"
+    opts = Keyword.take(opts, [:unary, :materializer])
+
+    if opts[:unary] != true do
+      raise ArgumentError, "GRPC.Stream.run/1 only supports unary streams"
     end
 
-    # We have to call `Enum.to_list` because we want to actually run and materialize the full stream. 
-    # List.flatten and List.first are used so that we can obtain the first result of the materialized list.
-    flow
-    |> Enum.to_list()
-    |> List.flatten()
-    |> List.first()
+    materializer = opts[:materializer]
+
+    if is_nil(materializer) do
+      raise ArgumentError,
+            "GRPC.Stream.run/1 requires a materializer to be set in the GRPC.Stream"
+    end
+
+    send_response(materializer, Enum.at(flow, 0))
+
+    :noreply
   end
 
   @doc """
@@ -335,6 +338,20 @@ defmodule GRPC.Stream do
     opts = Keyword.merge(opts, metadata: metadata)
     dispatcher = Keyword.get(opts, :default_dispatcher, GenStage.DemandDispatcher)
 
+    if opts[:unary] do
+      case opts[:materializer] do
+        %GRPC.Server.Stream{grpc_type: :unary} ->
+          :ok
+
+        %GRPC.Server.Stream{} ->
+          raise ArgumentError,
+                "materializer must be set to a unary GRPC.Server.Stream when unary: true is passed"
+
+        _ ->
+          raise ArgumentError, "materializer is required when unary: true is passed"
+      end
+    end
+
     flow =
       case Keyword.get(opts, :join_with) do
         pid when is_pid(pid) ->

test/grpc/stream_test.exs

@@ -1,5 +1,5 @@
 defmodule GRPC.StreamTest do
-  use ExUnit.Case
+  use GRPC.Integration.TestCase
   doctest GRPC.Stream
 
   describe "simple test" do
@@ -9,22 +9,42 @@ defmodule GRPC.StreamTest do
 
     defmodule FakeAdapter do
       def get_headers(_), do: %{"content-type" => "application/grpc"}
-    end
 
-    test "unary/2 creates a flow from a unary input" do
-      input = %TestInput{message: 1}
+      def send_reply(%{test_pid: test_pid, ref: ref}, item, _opts) do
+        send(test_pid, {:send_reply, ref, item})
+      end
 
-      result =
-        GRPC.Stream.unary(input)
-        |> GRPC.Stream.map(& &1)
-        |> GRPC.Stream.run()
+      def send_trailers(%{test_pid: test_pid, ref: ref}, trailers) do
+        send(test_pid, {:send_trailers, ref, trailers})
+      end
+    end
 
-      assert result == input
+    test "unary/2 creates a flow from a unary input" do
+      test_pid = self()
+      ref = make_ref()
+
+      input = %Routeguide.Point{latitude: 1, longitude: 2}
+
+      materializer = %GRPC.Server.Stream{
+        adapter: FakeAdapter,
+        payload: %{test_pid: test_pid, ref: ref},
+        grpc_type: :unary
+      }
+
+      assert :noreply =
+               GRPC.Stream.unary(input, materializer: materializer)
+               |> GRPC.Stream.map(fn item ->
+                 item
+               end)
+               |> GRPC.Stream.run()
+
+      assert_receive {:send_reply, ^ref, response}
+      assert IO.iodata_to_binary(response) == Protobuf.encode(input)
     end
 
     test "unary/2 creates a flow with metadata" do
       input = %TestInput{message: 1}
-      materializer = %GRPC.Server.Stream{adapter: FakeAdapter}
+      materializer = %GRPC.Server.Stream{adapter: FakeAdapter, grpc_type: :unary}
 
       flow =
         GRPC.Stream.unary(input, materializer: materializer, propagate_context: true)
@@ -51,7 +71,7 @@ defmodule GRPC.StreamTest do
 
     test "from_as_ctx/3 creates a flow from enumerable input" do
       input = [%{message: "a"}, %{message: "b"}]
-      materializer = %GRPC.Server.Stream{adapter: FakeAdapter}
+      materializer = %GRPC.Server.Stream{adapter: FakeAdapter, grpc_type: :unary}
 
       flow =
         GRPC.Stream.from(input, propagate_context: true, materializer: materializer)
@@ -257,6 +277,38 @@ defmodule GRPC.StreamTest do
     end
   end
 
+  defmodule MyGRPCService do
+    use GRPC.Server, service: Routeguide.RouteGuide.Service
+
+    def get_feature(input, materializer) do
+      GRPC.Stream.unary(input, materializer: materializer)
+      |> GRPC.Stream.map(fn point ->
+        %Routeguide.Feature{location: point, name: "#{point.latitude},#{point.longitude}"}
+      end)
+      |> GRPC.Stream.run()
+    end
+  end
+
+  describe "run/1" do
+    test "runs a unary stream" do
+      run_server([MyGRPCService], fn port ->
+        point = %Routeguide.Point{latitude: 409_146_138, longitude: -746_188_906}
+        {:ok, channel} = GRPC.Stub.connect("localhost:#{port}", adapter_opts: [retry_timeout: 10])
+
+        expected_response = %Routeguide.Feature{
+          location: point,
+          name: "#{point.latitude},#{point.longitude}"
+        }
+
+        assert {:ok, response, %{trailers: trailers}} =
+                 Routeguide.RouteGuide.Stub.get_feature(channel, point, return_headers: true)
+
+        assert response == expected_response
+        assert trailers == GRPC.Transport.HTTP2.server_trailers()
+      end)
+    end
+  end
+
   defp receive_loop do
     receive do
       {:request, item, from} ->

test/support/integration_test_case.ex

@@ -10,7 +10,14 @@ defmodule GRPC.Integration.TestCase do
   end
 
   def run_server(servers, func, port \\ 0, opts \\ []) do
-    {:ok, _pid, port} = GRPC.Server.start(servers, port, opts)
+    {:ok, _pid, port} =
+      start_supervised(%{
+        id: {GRPC.Server, System.unique_integer([:positive])},
+        start: {GRPC.Server, :start, [servers, port, opts]},
+        type: :worker,
+        restart: :permanent,
+        shutdown: 500
+      })
 
     try do
       func.(port)
@@ -20,7 +27,14 @@ defmodule GRPC.Integration.TestCase do
   end
 
   def run_endpoint(endpoint, func, port \\ 0) do
-    {:ok, _pid, port} = GRPC.Server.start_endpoint(endpoint, port)
+    {:ok, _pid, port} =
+      start_supervised(%{
+        id: {GRPC.Server, System.unique_integer([:positive])},
+        start: {GRPC.Server, :start_endpoint, [endpoint, port]},
+        type: :worker,
+        restart: :permanent,
+        shutdown: 500
+      })
 
     try do
       func.(port)