Ref/rename stream mat

README.md

@@ -76,8 +76,8 @@ In the following sections you will see how to implement gRPC server logic.
 defmodule Helloworld.Greeter.Server do
   use GRPC.Server, service: Helloworld.Greeter.Service
 
-  @spec say_hello(Helloworld.HelloRequest.t, GRPC.Server.Stream.t) :: Helloworld.HelloReply.t
-  def say_hello(request, _stream) do
+  @spec say_hello(Helloworld.HelloRequest.t, GRPC.Server.Materializer.t) :: Helloworld.HelloReply.t
+  def say_hello(request, _mat) do
     Helloworld.HelloReply.new(message: "Hello #{request.name}")
   end
 end
@@ -160,8 +160,8 @@ defmodule Helloworld.Greeter.Server do
     service: Helloworld.Greeter.Service,
     http_transcode: true
 
-  @spec say_hello(Helloworld.HelloRequest.t, GRPC.Server.Stream.t) :: Helloworld.HelloReply.t
-  def say_hello(request, _stream) do
+  @spec say_hello(Helloworld.HelloRequest.t, GRPC.Server.Materializer.t) :: Helloworld.HelloReply.t
+  def say_hello(request, _mat) do
     %Helloworld.HelloReply{message: "Hello #{request.name}"}
   end
 end

examples/helloworld/lib/server.ex

@@ -3,7 +3,7 @@ defmodule Helloworld.Greeter.Server do
     service: Helloworld.Greeter.Service,
     http_transcode: true
 
-  @spec say_hello(Helloworld.HelloRequest.t(), GRPC.Server.Stream.t()) ::
+  @spec say_hello(Helloworld.HelloRequest.t(), GRPC.Server.Materializer.t()) ::
           Helloworld.HelloReply.t()
   def say_hello(request, _stream) do
     %Helloworld.HelloReply{
@@ -12,7 +12,7 @@ defmodule Helloworld.Greeter.Server do
     }
   end
 
-  @spec say_hello_from(Helloworld.HelloFromRequest.t(), GRPC.Server.Stream.t()) ::
+  @spec say_hello_from(Helloworld.HelloFromRequest.t(), GRPC.Server.Materializer.t()) ::
           Helloworld.HelloReply.t()
   def say_hello_from(request, _stream) do
     %Helloworld.HelloReply{

examples/helloworld_streams/.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

examples/helloworld_streams/.gitignore

@@ -0,0 +1,26 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+helloworld_streams-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

examples/helloworld_streams/README.md

@@ -0,0 +1,21 @@
+# HelloworldStreams
+
+**TODO: Add description**
+
+## Installation
+
+If [available in Hex](https://hex.pm/docs/publish), the package can be installed
+by adding `helloworld_streams` to your list of dependencies in `mix.exs`:
+
+```elixir
+def deps do
+  [
+    {:helloworld_streams, "~> 0.1.0"}
+  ]
+end
+```
+
+Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
+and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
+be found at <https://hexdocs.pm/helloworld_streams>.
+

examples/helloworld_streams/lib/helloworld_streams.ex

@@ -0,0 +1,5 @@
+defmodule HelloworldStreams do
+  @moduledoc """
+  Documentation for `HelloworldStreams`.
+  """
+end

examples/helloworld_streams/lib/helloworld_streams/application.ex

@@ -0,0 +1,19 @@
+defmodule HelloworldStreams.Application do
+  @moduledoc false
+  use Application
+
+  @impl true
+  def start(_type, _args) do
+    children = [
+      HelloworldStreams.Utils.Transformer,
+      GrpcReflection,
+      {
+        GRPC.Server.Supervisor,
+        endpoint: HelloworldStreams.Endpoint, port: 50053, start_server: true
+      }
+    ]
+
+    opts = [strategy: :one_for_one, name: HelloworldStreams.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
+end

examples/helloworld_streams/lib/helloworld_streams/endpoint.ex

@@ -0,0 +1,8 @@
+defmodule HelloworldStreams.Endpoint do
+  @moduledoc false
+  use GRPC.Endpoint
+
+  intercept(GRPC.Server.Interceptors.Logger)
+  run(HelloworldStreams.Utils.Reflection)
+  run(HelloworldStreams.Server)
+end

examples/helloworld_streams/lib/helloworld_streams/server.ex

@@ -0,0 +1,61 @@
+defmodule HelloworldStreams.Server do
+  @moduledoc """
+  gRPC service for streaming data.
+  """
+  use GRPC.Server, service: Stream.EchoServer.Service
+
+  alias HelloworldStreams.Utils.Transformer
+  alias GRPC.Stream, as: GRPCStream
+
+  alias Stream.HelloRequest
+  alias Stream.HelloReply
+
+  @spec say_unary_hello(HelloRequest.t(), GRPC.Server.Materializer.t()) :: any()
+  def say_unary_hello(request, _materializer) do
+    GRPCStream.unary(request)
+    |> GRPCStream.ask(Transformer)
+    |> GRPCStream.map(fn %HelloReply{} = reply ->
+      %HelloReply{message: "[Reply] #{reply.message}"}
+    end)
+    |> GRPCStream.run()
+  end
+
+  @spec say_server_hello(HelloRequest.t(), GRPC.Server.Materializer.t()) :: any()
+  def say_server_hello(request, materializer) do
+    create_output_stream(request)
+    |> GRPCStream.from()
+    |> GRPCStream.run_with(materializer)
+  end
+
+  defp create_output_stream(msg) do
+    Stream.repeatedly(fn ->
+      index = :rand.uniform(10)
+      %HelloReply{message: "[#{index}] I'm the Server for #{msg.name}"}
+    end)
+    |> Stream.take(10)
+    |> Enum.to_list()
+  end
+
+  @spec say_bid_stream_hello(Enumerable.t(), GRPC.Server.Materializer.t()) :: any()
+  def say_bid_stream_hello(request, materializer) do
+    # simulate a infinite stream of data
+    # this is a simple example, in a real world application
+    # you would probably use a GenStage or similar
+    # to handle the stream of data
+    output_stream =
+      Stream.repeatedly(fn ->
+        index = :rand.uniform(10)
+        %HelloReply{message: "[#{index}] I'm the Server ;)"}
+      end)
+
+    GRPCStream.from(request, join_with: output_stream)
+    |> GRPCStream.map(fn
+      %HelloRequest{} = hello ->
+        %HelloReply{message: "Welcome #{hello.name}"}
+
+      output_item ->
+        output_item
+    end)
+    |> GRPCStream.run_with(materializer)
+  end
+end

examples/helloworld_streams/lib/helloworld_streams/stream.pb.ex

@@ -0,0 +1,156 @@
+defmodule Stream.HelloRequest do
+  @moduledoc false
+
+  use Protobuf, protoc_gen_elixir_version: "0.14.0", syntax: :proto3
+
+  def descriptor do
+    # credo:disable-for-next-line
+    %Google.Protobuf.DescriptorProto{
+      name: "HelloRequest",
+      field: [
+        %Google.Protobuf.FieldDescriptorProto{
+          name: "name",
+          extendee: nil,
+          number: 1,
+          label: :LABEL_OPTIONAL,
+          type: :TYPE_STRING,
+          type_name: nil,
+          default_value: nil,
+          options: nil,
+          oneof_index: nil,
+          json_name: "name",
+          proto3_optional: nil,
+          __unknown_fields__: []
+        }
+      ],
+      nested_type: [],
+      enum_type: [],
+      extension_range: [],
+      extension: [],
+      options: nil,
+      oneof_decl: [],
+      reserved_range: [],
+      reserved_name: [],
+      __unknown_fields__: []
+    }
+  end
+
+  field(:name, 1, type: :string)
+end
+
+defmodule Stream.HelloReply do
+  @moduledoc false
+
+  use Protobuf, protoc_gen_elixir_version: "0.14.0", syntax: :proto3
+
+  def descriptor do
+    # credo:disable-for-next-line
+    %Google.Protobuf.DescriptorProto{
+      name: "HelloReply",
+      field: [
+        %Google.Protobuf.FieldDescriptorProto{
+          name: "message",
+          extendee: nil,
+          number: 1,
+          label: :LABEL_OPTIONAL,
+          type: :TYPE_STRING,
+          type_name: nil,
+          default_value: nil,
+          options: nil,
+          oneof_index: nil,
+          json_name: "message",
+          proto3_optional: nil,
+          __unknown_fields__: []
+        }
+      ],
+      nested_type: [],
+      enum_type: [],
+      extension_range: [],
+      extension: [],
+      options: nil,
+      oneof_decl: [],
+      reserved_range: [],
+      reserved_name: [],
+      __unknown_fields__: []
+    }
+  end
+
+  field(:message, 1, type: :string)
+end
+
+defmodule Stream.EchoServer.Service do
+  @moduledoc false
+
+  use GRPC.Service, name: "stream.EchoServer", protoc_gen_elixir_version: "0.14.0"
+
+  def descriptor do
+    # credo:disable-for-next-line
+    %Google.Protobuf.ServiceDescriptorProto{
+      name: "EchoServer",
+      method: [
+        %Google.Protobuf.MethodDescriptorProto{
+          name: "SayUnaryHello",
+          input_type: ".stream.HelloRequest",
+          output_type: ".stream.HelloReply",
+          options: %Google.Protobuf.MethodOptions{
+            deprecated: false,
+            idempotency_level: :IDEMPOTENCY_UNKNOWN,
+            features: nil,
+            uninterpreted_option: [],
+            __pb_extensions__: %{},
+            __unknown_fields__: []
+          },
+          client_streaming: false,
+          server_streaming: false,
+          __unknown_fields__: []
+        },
+        %Google.Protobuf.MethodDescriptorProto{
+          name: "SayServerHello",
+          input_type: ".stream.HelloRequest",
+          output_type: ".stream.HelloReply",
+          options: %Google.Protobuf.MethodOptions{
+            deprecated: false,
+            idempotency_level: :IDEMPOTENCY_UNKNOWN,
+            features: nil,
+            uninterpreted_option: [],
+            __pb_extensions__: %{},
+            __unknown_fields__: []
+          },
+          client_streaming: false,
+          server_streaming: true,
+          __unknown_fields__: []
+        },
+        %Google.Protobuf.MethodDescriptorProto{
+          name: "SayBidStreamHello",
+          input_type: ".stream.HelloRequest",
+          output_type: ".stream.HelloReply",
+          options: %Google.Protobuf.MethodOptions{
+            deprecated: false,
+            idempotency_level: :IDEMPOTENCY_UNKNOWN,
+            features: nil,
+            uninterpreted_option: [],
+            __pb_extensions__: %{},
+            __unknown_fields__: []
+          },
+          client_streaming: true,
+          server_streaming: true,
+          __unknown_fields__: []
+        }
+      ],
+      options: nil,
+      __unknown_fields__: []
+    }
+  end
+
+  rpc(:SayUnaryHello, Stream.HelloRequest, Stream.HelloReply)
+
+  rpc(:SayServerHello, Stream.HelloRequest, stream(Stream.HelloReply))
+
+  rpc(:SayBidStreamHello, stream(Stream.HelloRequest), stream(Stream.HelloReply))
+end
+
+defmodule Stream.EchoServer.Stub do
+  @moduledoc false
+
+  use GRPC.Stub, service: Stream.EchoServer.Service
+end

examples/helloworld_streams/lib/helloworld_streams/utils/reflection.ex

@@ -0,0 +1,8 @@
+defmodule HelloworldStreams.Utils.Reflection do
+  @moduledoc """
+  gRPC reflection server.
+  """
+  use GrpcReflection.Server,
+    version: :v1,
+    services: [Stream.EchoServer.Service]
+end

examples/helloworld_streams/lib/helloworld_streams/utils/transformer.ex

@@ -0,0 +1,20 @@
+defmodule HelloworldStreams.Utils.Transformer do
+  @moduledoc """
+  `Transformer` GenServer for example purposes.
+  """
+  use GenServer
+
+  alias Stream.HelloRequest
+  alias Stream.HelloReply
+
+  def start_link(_) do
+    GenServer.start_link(__MODULE__, nil, name: __MODULE__)
+  end
+
+  def init(_), do: {:ok, %{}}
+
+  def handle_info({:request, %HelloRequest{} = value, from}, state) do
+    Process.send(from, {:response, %HelloReply{message: "Hello #{value.name}"}}, [])
+    {:noreply, state}
+  end
+end