merge with master

Author: sleipnir

README.md

@@ -144,8 +144,28 @@ def say_bid_stream_hello(request, materializer) do
   |> GRPC.Stream.run_with(materializer)
 end
 ```
-
-The Stream API supports composable stream transformations via `ask`, `map`, `run` and other functions, enabling clean and declarative stream pipelines. For a complete list of available operators and detailed documentation, see [`GRPC.Stream`](lib/grpc/stream.ex).
+The Stream API supports composable stream transformations via `ask`, `map`, `run` and others functions, enabling clean and declarative stream pipelines. See the table below:
+
+| Function                         | Description  | Parameters / Options  |
+|:---------------------------------|:-------------|:----------------------|
+| **`from(input, opts \\\\ [])`**  | Converts a gRPC stream (or list) into a `Flow` with backpressure support. Allows joining with external `GenStage` producers. | **Parameters:**<br>• `input` — stream, list, or gRPC struct.<br>**Options:**<br>• `:join_with` — PID or name of an external `GenStage` producer.<br>• `:dispatcher` — dispatcher module (default: `GenStage.DemandDispatcher`).<br>• `:propagate_context` — if `true`, propagates the materializer context.<br>• `:materializer` — the current `%GRPC.Server.Stream{}`.<br>• Other options supported by `Flow`. |
+| **`unary(input, opts \\\\ [])`** | Creates a `Flow` from a single gRPC request (unary). Useful for non-streaming calls that still leverage the Flow API. | **Parameters:**<br>• `input` — single gRPC message.<br>**Options:** same as `from/2`. |
+| **`to_flow(stream)`**            | Returns the underlying `Flow` from a `GRPC.Stream`. If uninitialized, returns `Flow.from_enumerable([])`. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}` struct. |
+| **`run(stream)`**                | Executes the `Flow` for a unary stream and returns the first materialized result. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}` with `unary: true` option. |
+| **`run_with(stream, materializer, opts \\\\ [])`** | Executes the `Flow` and sends responses into the gRPC server stream. Supports `:dry_run` for test mode without sending messages. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `materializer` — `%GRPC.Server.Stream{}`.<br>**Options:**<br>• `:dry_run` — if `true`, responses are not sent. |
+| **`ask(stream, target, timeout \\\\ 5000)`** | Sends a request to an external process (`PID` or named process) and waits for a response (`{:response, msg}`). Returns an updated stream or an error. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `target` — PID or atom.<br>• `timeout` — in milliseconds. |
+| **`ask!(stream, target, timeout \\\\ 5000)`** | Same as `ask/3`, but raises an exception on failure (aborts the Flow). | Same parameters as `ask/3`. |
+| **`filter(stream, fun)`** | Filters items in the stream by applying a concurrent predicate function. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — function `(item -> boolean)`. |
+| **`flat_map(stream, fun)`** | Applies a function returning a list or enumerable, flattening the results. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — `(item -> Enumerable.t())`. |
+| **`map(stream, fun)`** | Applies a transformation function to each item in the stream. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — `(item -> term)`. |
+| **`map_with_context(stream, fun)`** | Applies a function to each item, passing the stream context (e.g., headers) as an additional argument. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — `(context, item -> term)`. |
+| **`partition(stream, opts \\\\ [])`** | Partitions the stream to group items by key or condition before stateful operations like `reduce/3`. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `opts` — partitioning options (`Flow.partition/2`). |
+| **`reduce(stream, acc_fun, reducer_fun)`** | Reduces the stream using an accumulator, useful for aggregations. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `acc_fun` — initializer function `() -> acc`.<br>• `reducer_fun` — `(item, acc -> acc)`. |
+| **`uniq(stream)`** | Emits only distinct items from the stream (no custom uniqueness criteria). | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`. |
+| **`uniq_by(stream, fun)`** | Emits only unique items based on the return value of the provided function. | **Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — `(item -> term)` for uniqueness determination. |
+| **`get_headers(stream)`** | Retrieves HTTP/2 headers from a `%GRPC.Server.Stream{}`. | **Parameters:**<br>• `stream` — `%GRPC.Server.Stream{}`.<br>**Returns:** `map` containing decoded headers. |
+
+For a complete list of available operators see [here](lib/grpc/stream.ex).
 
 ---
 
@@ -186,6 +206,25 @@ This section demonstrates how to establish client connections and perform RPC ca
 
 ## Basic Connection and RPC
 
+
+Typically, you start this client supervisor as part of your application's supervision tree:
+
+```elixir
+children = [
+  {GRPC.Client.Supervisor, []}
+]
+
+opts = [strategy: :one_for_one, name: MyApp.Supervisor]
+Supervisor.start_link(children, opts)
+``` 
+
+You can also start it manually in scripts or test environments:
+```elixir
+{:ok, _pid} = DynamicSupervisor.start_link(strategy: :one_for_one, name: GRPC.Client.Supervisor)
+``` 
+
+Then connect with gRPC server:
+
 ```elixir
 iex> {:ok, channel} = GRPC.Stub.connect("localhost:50051")
 iex> request = Helloworld.HelloRequest.new(name: "grpc-elixir")
@@ -211,7 +250,7 @@ iex> {:ok, reply} = channel |> Helloworld.GreetingServer.Stub.say_unary_hello(re
 
 ## Target Schemes and Resolvers
 
-The `connect/2` function supports URI-like targets that are resolved via the internal **gRPC** [Resolver](lib/grpc/client/resolver.ex).
+The `connect/2` function supports URI-like targets that are resolved via the internal **gRPC** [Resolver](lib/grpc/client/resolver.ex).  
 You can connect using `DNS`, `Unix Domain sockets`, `IPv4/IPv6`, or even `xDS-based endpoints`.
 
 ### Supported formats:
@@ -257,7 +296,7 @@ iex> {:ok, channel} =
 
 ## Client Adapters
 
-By default, `GRPC.Stub.connect/2` uses the **Gun** adapter.
+By default, `GRPC.Stub.connect/2` uses the **Gun** adapter.  
 You can switch to **Mint** (pure Elixir HTTP/2) or other adapters as needed.
 
 ### Using Mint Adapter
@@ -346,7 +385,7 @@ defmodule Helloworld.Greeter.Server do
 end
 ```
 
-See full application code in [helloworld_transcoding](https://github.com/elixir-grpc/tree/master/examples/helloworld_transcoding) example.
+See full application code in [helloworld_transcoding](examples/helloworld_transcoding) example.
 
 ### **CORS**
 
@@ -382,9 +421,9 @@ end
 
 ## Benchmark
 
-1. [Simple benchmark](https://github.com/elixir-grpc/tree/master/examples/helloworld/README.md#Benchmark) by using [ghz](https://ghz.sh/)
+1. [Simple benchmark](examples/helloworld/README.md#Benchmark) by using [ghz](https://ghz.sh/)
 
-2. [Benchmark](https://github.com/elixir-grpc/tree/master/benchmark) followed by official spec
+2. [Benchmark](benchmark) followed by official spec
 
 ## Contributing
 

examples/route_guide/lib/server.ex

@@ -1,20 +1,5 @@
-defmodule CustomErrorInterceptor do
-  @behaviour GRPC.Server.Interceptor
-
-  def init(opts), do: opts
-
-  def call(req, stream, next, opts) do
-    try do
-      next.(req, stream)
-    rescue
-      CustomError ->
-        Server.send_reply(stream, ...)
-    end
-  end
-end
-
 defmodule Routeguide.RouteGuide.Server do
-  use GRPC.Server, service: Routeguide.RouteGuide.Service, interceptors: [CustomErrorInterceptor]
+  use GRPC.Server, service: Routeguide.RouteGuide.Service
   alias GRPC.Server
   alias RouteGuide.Data
 

interop/mix.lock

@@ -4,6 +4,7 @@
   "extrace": {:hex, :extrace, "0.5.0", "4ee5419fbc3820c4592daebe0f8527001aa623578d9a725d8ae521315fce0277", [:mix], [{:recon, "~> 2.5", [hex: :recon, repo: "hexpm", optional: false]}], "hexpm", "2a3ab7fa0701949efee1034293fa0b0e65926ffe256ccd6d0e10dd8a9406cd02"},
   "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"},
+  "googleapis": {:hex, :googleapis, "0.1.0", "13770f3f75f5b863fb9acf41633c7bc71bad788f3f553b66481a096d083ee20e", [:mix], [{:protobuf, "~> 0.12", [hex: :protobuf, repo: "hexpm", optional: false]}], "hexpm", "1989a7244fd17d3eb5f3de311a022b656c3736b39740db46506157c4604bd212"},
   "grpc": {:git, "https://github.com/elixir-grpc/grpc.git", "21422839798e49bf6d29327fab0a7add51becedd", []},
   "grpc_statsd": {:hex, :grpc_statsd, "0.1.0", "a95ae388188486043f92a3c5091c143f5a646d6af80c9da5ee616546c4d8f5ff", [:mix], [{:grpc, ">= 0.0.0", [hex: :grpc, repo: "hexpm", optional: true]}, {:statix, ">= 0.0.0", [hex: :statix, repo: "hexpm", optional: true]}], "hexpm", "de0c05db313c7b3ffeff345855d173fd82fec3de16591a126b673f7f698d9e74"},
   "gun": {:hex, :gun, "2.0.1", "160a9a5394800fcba41bc7e6d421295cf9a7894c2252c0678244948e3336ad73", [:make, :rebar3], [{:cowlib, "2.12.1", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "a10bc8d6096b9502205022334f719cc9a08d9adcfbfc0dbee9ef31b56274a20b"},