You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
@@ -144,28 +149,104 @@ def say_bid_stream_hello(request, materializer) do
144
149
|>GRPC.Stream.run_with(materializer)
145
150
end
146
151
```
147
-
The Stream API supports composable stream transformations via `ask`, `map`, `run` and others functions, enabling clean and declarative stream pipelines. See the table below:
|**`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`. |
152
-
|**`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`. |
153
-
|**`to_flow(stream)`**| Returns the underlying `Flow` from a `GRPC.Stream`. If uninitialized, returns `Flow.from_enumerable([])`. |**Parameters:**<br>• `stream` — `%GRPC.Stream{}` struct. |
154
-
|**`run(stream)`**| Executes the `Flow` for a unary stream and returns the first materialized result. |**Parameters:**<br>• `stream` — `%GRPC.Stream{}` with `unary: true` option. |
155
-
|**`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. |
156
-
|**`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. |
157
-
|**`ask!(stream, target, timeout \\\\ 5000)`**| Same as `ask/3`, but raises an exception on failure (aborts the Flow). | Same parameters as `ask/3`. |
158
-
|**`filter(stream, fun)`**| Filters items in the stream by applying a concurrent predicate function. |**Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — function `(item -> boolean)`. |
159
-
|**`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())`. |
160
-
|**`map(stream, fun)`**| Applies a transformation function to each item in the stream. |**Parameters:**<br>• `stream` — `%GRPC.Stream{}`.<br>• `fun` — `(item -> term)`. |
161
-
|**`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)`. |
162
-
|**`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`). |
163
-
|**`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)`. |
164
-
|**`uniq(stream)`**| Emits only distinct items from the stream (no custom uniqueness criteria). |**Parameters:**<br>• `stream` — `%GRPC.Stream{}`. |
165
-
|**`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. |
166
-
|**`get_headers(stream)`**| Retrieves HTTP/2 headers from a `%GRPC.Server.Stream{}`. |**Parameters:**<br>• `stream` — `%GRPC.Server.Stream{}`.<br>**Returns:**`map` containing decoded headers. |
167
-
168
-
For a complete list of available operators see [here](lib/grpc/stream.ex).
152
+
The Stream API supports composable stream transformations via `ask`, `map`, `run` and others functions, enabling clean and declarative stream pipelines. For a complete list of available operators see [here](lib/grpc/stream.ex).
153
+
154
+
---
155
+
156
+
### Effects and Error Handling
157
+
158
+
#### Side Effects
159
+
160
+
The `effect/2` operator executes user-defined functions for each element in the stream, allowing the integration of non-transformative actions such as logging, metrics, or external notifications.
161
+
162
+
Unlike transformation operators (e.g., `map/2`), `effect/2` does not modify or filter values — it preserves the original stream while executing the provided callback safely for each emitted element.
163
+
164
+
```elixir
165
+
iex> parent =self()
166
+
iex> stream =
167
+
...>GRPC.Stream.from([1, 2, 3])
168
+
...>|>GRPC.Stream.effect(fn x ->send(parent, {:seen, x *2}) end)
169
+
...>|>GRPC.Stream.to_flow()
170
+
...>|>Enum.to_list()
171
+
iex> assert_receive {:seen, 2}
172
+
iex> assert_receive {:seen, 4}
173
+
iex> assert_receive {:seen, 6}
174
+
iex> stream
175
+
[1, 2, 3]
176
+
```
177
+
178
+
Key characteristics:
179
+
180
+
* The callback function (`effect_fun`) is invoked for each item emitted downstream.
181
+
* The result of the callback is ignored, ensuring that the stream’s structure and values remain unchanged.
182
+
* Execution is lazy and occurs only when the stream is materialized using run/1, run_with/3, or to_flow/1.
183
+
* Exceptions raised inside the callback are captured internally, preventing interruption of the dataflow.
184
+
185
+
This operator is designed for observability, telemetry, auditing, and integration with external systems that must react to events flowing through the gRPC stream.
186
+
187
+
---
188
+
189
+
#### Recovery from errors
190
+
191
+
The `map_error/2` operator intercepts and transforms errors or exceptions emitted by previous stages in a stream pipeline.
192
+
193
+
It provides a unified mechanism for handling:
194
+
195
+
* Expected errors, such as validation or domain failures (`{:error, reason}`)
196
+
* Unexpected runtime errors, including raised or thrown exceptions inside other operators.
* The function inside `map/2` raises an exception for the value `2`.
213
+
*`map_error/2` captures and transforms that error into a structured `GRPC.RPCError` response.
214
+
* The stream continues processing without being interrupted.
215
+
216
+
This makes map_error/2 suitable for input validation, runtime fault recovery, and user-facing error translation within gRPC pipelines.
217
+
218
+
---
219
+
220
+
#### Unified Error Matching and Propagation
221
+
222
+
All stream operators share a unified error propagation model that guarantees consistent handling of exceptions and failures across the pipeline.
223
+
224
+
This ensures that user-defined functions within the stream — whether pure transformations, side effects, or external calls — always produce a predictable and recoverable result, maintaining the integrity of the dataflow even in the presence of unexpected errors.
By normalizing all possible outcomes, `GRPC.Stream` ensures fault-tolerant, exception-safe pipelines where operators can freely raise, throw, or return tuples without breaking the flow execution.
246
+
247
+
This unified model allows developers to build composable and reliable streaming pipelines that gracefully recover from both domain and runtime errors.
248
+
249
+
>_NOTE_: In the example above, we could use `map_error/2` instead of `map/2` to handle error cases explicitly. However, since the function also performs a transformation on successful values, `map/2` remains appropriate and useful in this context.
169
250
170
251
---
171
252
@@ -175,7 +256,7 @@ Add the server supervisor to your application's supervision tree:
0 commit comments