# `GRPC.Stream`
[🔗](https://github.com/elixir-grpc/grpc_server/blob/v1.0.1/lib/grpc/stream.ex#L1)

Provides a `Flow`-based abstraction layer for building gRPC streaming pipelines in Elixir.

This module allows you to consume gRPC request streams as `Flow` pipelines with support for
backpressure via GenStage. You can also produce gRPC responses by materializing a `Flow`
back into the gRPC stream.

## Capabilities

- Transforms an incoming gRPC request stream into a `Flow` with backpressure.
- Emits messages back into the gRPC response stream using `run_with/3`.
- Supports joining with external producers (e.g., RabbitMQ, Kafka) for unbounded or fan-in stream sources.
- Offers composable functional operators (`map/2`, `filter/2`, `flat_map/2`, etc.) on the stream.

## Example: Bidirectional Streaming

    defmodule MyGRPCService do
      use GRPC.Server, service: MyService.Service

      def route_chat(input, materializer) do
        GRPC.Stream.from(input, max_demand: 10)
        |> GRPC.Stream.map(fn note -> process_note(note) end)
        |> GRPC.Stream.run_with(materializer)
      end

      defp process_note(note), do: %Response{message: "Received"}
    end

## Example: Joining with an External Producer

When integrating with external unbounded sources (e.g., message queues),
you can pass a running `GenStage` producer using the `:join_with` option:

    defmodule MyGRPCService do
      use GRPC.Server, service: MyService.Service

      def stream_events(input, materializer) do
        {:ok, pid} = MyApp.RabbitMQ.Producer.start_link([])

        GRPC.Stream.from(input, join_with: pid, max_demand: 10)
        |> GRPC.Stream.map(&handle_event/1)
        |> GRPC.Stream.run_with(materializer)
      end

      defp handle_event({_, msg}), do: msg
      defp handle_event(event), do: %MyGRPC.Event{data: inspect(event)}
    end

# `from`

```elixir
@spec from(any(), Keyword.t()) :: t()
```

Converts a gRPC input into a `Flow` pipeline with backpressure support.

## Parameters

  - `input`: A gRPC request stream (struct, enumerable, or Elixir `Stream`).

## Options

  - `:flow` — An optional pre-built `Flow` to use instead of building from input.
  - `:join_with` — An optional external `GenStage` producer to merge with the gRPC input.
  - `: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

  A `GRPC.Stream` struct that represents the transformed stream.

## Example

    flow = GRPC.Stream.from(request, max_demand: 50)

# `unary`

```elixir
@spec unary(any(), Keyword.t()) :: t()
```

Converts a single gRPC request into a `Flow` pipeline with support for backpressure.
This is useful for unary gRPC requests where you want to use the Flow API.

## Parameters

  - `input`: The single gRPC message to convert into a Flow.

## Options

  - `:join_with` - An optional additional producer stage PID to include in the Flow.
  - `:dispatcher` - An optional `GenStage` dispatcher to use in the underlying `Flow`. 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
  - A `GRPC.Stream` that emits the single gRPC message under demand.

## Example

    flow = GRPC.Stream.unary(request, max_demand: 5)

# `run`

```elixir
@spec run(stream :: t()) :: :noreply
```

Executes the underlying `Flow` for a unary stream.

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.

The `stream` argument must be initialized as a `:unary` stream with
a `:materializer` set.

## Example

    def say_unary_hello(request, mat) do
      GRPC.Stream.unary(request, materializer: mat)
      |> GRPC.Stream.map(fn
        %HelloReply{} = reply ->
          %HelloReply{message: "[Reply] message"}

        {:error, _reason} ->
          GRPC.RPCError.exception(message: "[Error] Something bad happened")
      end)
      |> GRPC.Stream.run()
    end

# `run_with`

```elixir
@spec run_with(t(), Stream.t(), Keyword.t()) :: :ok
```

Executes the flow 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.

## Options

  - `:dry_run` — If `true`, responses are not sent (used for testing or inspection).

## Returns

  - `:ok` if the stream was processed successfully.

## Example

    def say_bid_stream_hello(request, materializer) do
      output_stream =
        Stream.repeatedly(fn ->
          %HelloReply{message: "I'm the Server ;)"}
        end)

      GRPC.Stream.from(request, join_with: output_stream)
      |> GRPC.Stream.map(fn
        %HelloRequest{} = _hello ->
          %HelloReply{message: "Welcome Sr!"}

        {:error, _reason} ->
          GRPC.RPCError.exception(message: "[Error] Something bad happened")
      end)
      |> GRPC.Stream.run_with(materializer)
    end

# `filter`

```elixir
@spec filter(t(), (term() -> term())) :: t()
```

Filters the stream using the given predicate function.

The filter function is applied concurrently to the stream entries, so it shouldn't rely on execution order.

# `flat_map`

```elixir
@spec flat_map(t(), (term() -> Enumerable.t())) :: t()
```

Applies a function to each entry and concatenates the resulting lists.

Useful for emitting multiple messages for each input.

# `map`

```elixir
@spec map(t(), (term() -> term())) :: t()
```

Applies a function to each stream item.

# `map_error`

Intercepts and transforms error tuples or unexpected exceptions that occur
within a gRPC stream pipeline.

`map_error/3` allows graceful handling or recovery from errors produced by previous
operators (e.g. `map/2`, `flat_map/2`) or from validation logic applied to incoming data.

The provided `handler/1` function receives the error reason (or the exception struct) like:

    {:error, reason} -> failure
    {:error, {:exception, exception}} -> failure due to exception
    {:error, {kind, reason}} -> failure due to throw or exit

And can either:

  * Return a new error tuple — e.g. `{:error, new_reason}` — to re-emit a modified error.
  * Return any other value to recover from the failure and continue the pipeline.

This makes it suitable for both input validation and capturing unexpected runtime errors
in stream transformations.

## Parameters

  - `stream` — The input stream or `Flow` pipeline.
  - `func` — A function that takes an error reason or exception and returns either a new value or an error tuple.

## Returns

  - A new stream where all error tuples and raised exceptions are processed by `func/1`.

## Examples

    iex> GRPC.Stream.from([1, 2])
    ...> |> GRPC.Stream.map(fn
    ...>   2 -> raise "boom"
    ...>   x -> x
    ...> end)
    ...> |> GRPC.Stream.map_error(fn
    ...>   {:error, {:exception, _reason}} ->
    ...>     {:error, GRPC.RPCError.exception(message: "Validation or runtime error")}
    ...> end)

In this example:

* The call to `GRPC.Stream.map/2` raises an exception for value `2`.
* `map_error/3` catches the error and wraps it in a `GRPC.RPCError` struct with a custom message.
* The pipeline continues execution, transforming errors into structured responses.

## Notes

- `map_error/3` is lazy and only executes when the stream is materialized
  (via `GRPC.Stream.run/1` or `GRPC.Stream.run_with/3`).

- Use this operator to implement robust error recovery, input validation, or
  to normalize exceptions from downstream Flow stages into well-defined gRPC errors.

# `map_with_context`

```elixir
@spec map_with_context(t(), (map(), term() -> term())) :: t()
```

Applies a transformation function to each stream item, passing the context as an additional argument.
This is useful for operations that require access to the stream's headers.

# `partition`

```elixir
@spec partition(
  t(),
  keyword()
) :: t()
```

Partitions the stream to allow grouping of items by key or condition.

Use this before stateful operations such as `reduce/3`.

## Note

Excessive use of partitioning can impact performance and memory usage.
Only partition when required for correctness or performance.
See https://hexdocs.pm/flow/Flow.html#module-partitioning for more details.

# `reduce`

```elixir
@spec reduce(t(), (-> acc), (term(), acc -> acc)) :: t() when acc: term()
```

Reduces items in the stream using an accumulator.

## Parameters

  - `acc_fun` initializes the accumulator,
  - `reducer_fun` updates it for each input.

## Note
See https://hexdocs.pm/flow/Flow.html#reduce/3 for more details.

# `to_flow`

```elixir
@spec to_flow(t()) :: Flow.t()
```

Extracts the underlying `Flow` pipeline from a `GRPC.Stream`.

Raises an `ArgumentError` if the `Flow` has not been initialized.

## Returns

  A `Flow` pipeline.

# `uniq`

```elixir
@spec uniq(t()) :: t()
```

Emits only distinct items from the stream. See `uniq_by/2` for more information.

# `uniq_by`

```elixir
@spec uniq_by(t(), (term() -> term())) :: t()
```

Emits only unique items as determined by the result of the given function.

## Note
This function requires care when used for unbounded flows. For more information see https://hexdocs.pm/flow/Flow.html#uniq_by/2

# `ask`

```elixir
@spec ask(t(), pid() | atom(), non_neg_integer()) ::
  t() | {:error, :timeout | :process_not_alive}
```

Sends a request to an external process and awaits a response.

If `target` is a PID, a message in the format `{:request, item, from}` is sent, and a reply
in the format `{:response, msg}` is expected.

If `target` is an `atom` we will try to locate the process through `Process.whereis/1`.

## Parameters

  - `stream`: The `GRPC.Stream` pipeline.
  - `target`: Target process PID or atom name.
  - `timeout`: Timeout in milliseconds (defaults to `5000`).

# `ask!`

```elixir
@spec ask!(t(), pid() | atom(), non_neg_integer()) :: t()
```

Same as `ask/3`, but raises an exception on failure.

## Caution

This version propagates errors via raised exceptions, which can crash the Flow worker and halt the pipeline.
Prefer `ask/3` for production usage unless failure should abort the stream.

# `effect`

```elixir
@spec effect(t(), (term() -> any())) :: t()
```

Applies a side-effect function to each element of the stream without altering its values.

The `effect/2` function is useful for performing imperative or external actions
(such as logging, sending messages, collecting metrics, or debugging)
while preserving the original stream data.

It behaves like `Enum.each/2`, but returns the stream itself so it can continue in the pipeline.

## Examples

```elixir
iex> parent = self()
iex> stream =
...>   GRPC.Stream.from([1, 2, 3])
...>   |> GRPC.Stream.effect(fn x -> send(parent, {:seen, x*2}) end)
...>   |> GRPC.Stream.to_flow()
...>   |> Enum.to_list()
iex> assert_receive {:seen, 2}
iex> assert_receive {:seen, 4}
iex> assert_receive {:seen, 6}
iex> stream
[1, 2, 3]
```
In this example, the effect/2 function sends a message to the current process
for each element in the stream, but the resulting stream values remain unchanged.

## Parameters

- `stream` — The input `GRPC.Stream`.
- `effect_fun` — A function that receives each item and performs a side effect
(e.g. IO.inspect/1, Logger.info/1, send/2, etc.).

### Notes

- This function is lazy — the `effect_fun` will only run once the stream is materialized
(e.g. via `GRPC.Stream.run/1` or `GRPC.Stream.run_with/3`).
- The use of `effect/2` ensures that the original item is returned unchanged,
enabling seamless continuation of the pipeline.

# `item`

```elixir
@type item() :: any()
```

# `reason`

```elixir
@type reason() :: any()
```

# `t`

```elixir
@type t() :: %GRPC.Stream{flow: Flow.t(), metadata: map(), options: Keyword.t()}
```

# `get_headers`

```elixir
@spec get_headers(GRPC.Server.Stream.t()) :: map()
```

Retrieves HTTP/2 headers from a `GRPC.Server.Stream`.

## Client Note

To receive headers on the client side, use the `:return_headers` option. See `GRPC.Stub`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*