# `Flezha.SDK.ChannelBuilder`
[🔗](https://github.com/flezha/flezha_elixir_sdk/blob/v0.1.0/lib/flezha/sdk/channel_builder.ex#L1)

The graph-builder token handed to `channels/1`. Openers (`input`/`input_with`),
edge transforms (`map`/`filter`/`flat_map`/`key_by`), consumers
(`into`/`into_named`/`output`/`output_with`/`call`,
`merge` with `from`, and `route_latest`/`route_latest_named`), per-edge knobs
(`named`/`on_full`/`retry_max`/`retry_backoff_ms`) and `keyed` (output
identity on the plugin wire the handle reaches). Fan-out is handle reuse, or
`branch` over a list of arms. The effectively-once tier is a class setting
(`Flezha.SDK.Relay`), not a builder call.

This builder RECORDS; it does not judge. Every fluent call appends one link
to the chain its handle carries — a transform, a key, a plugin output, an edge
into a tasklet, a merge source closing on one — and the record is kept in
the process dictionary for the duration of `channels/1` (the twin of the Rust
builder's `pending`). The declaration step hands the finished record to
`flz_core`, which judges it as it judges every SDK's.

The refusals that stay here are the ones the surface itself makes: `keyed`,
`named`, an output and a call on an entry handle name a fact an entry channel
has no end for, and none reaches the record.

# `link`

```elixir
@type link() ::
  {:map | :filter | :flat_map | :key, (term() -> term())}
  | :keyed
  | {:sink, String.t(), (term() -> binary())}
  | {:call, map(), (term() -> binary())}
  | {:into, reference()}
  | {:merge, reference(), non_neg_integer()}
  | {:branch, non_neg_integer()}
```

One link of a chain, in the order the author called it. A transform and a
key carry the closure the derivation fuses; a terminator carries what the
edge or the plugin wire it closes on needs.

# `t`

```elixir
@type t() :: %Flezha.SDK.ChannelBuilder{ref: reference()}
```

# `branch`

```elixir
@spec branch(Flezha.SDK.Stream.t(), [(Flezha.SDK.Stream.t() -> term())]) ::
  Flezha.SDK.Stream.t()
```

Fan out: call each arm with this same stream, which builds exactly the
topology the arms' calls off one held handle would, and answer the stream.

    placed |> branch([&into(&1, ConfirmFill), &into(&1, RecordOrder)])

The arity is recorded on a chain of its own, so the check can refuse a
branch of fewer than two arms as `BranchTooFewArms`.

# `call`

```elixir
@spec call(Flezha.SDK.Stream.t(), String.t() | nil, keyword()) ::
  Flezha.SDK.Stream.t()
```

End the chain in a call to another relay: every value reaching it, after
the chain's transforms and key, is one request, written as `output/2` writes
(`serialize:`). `target` names an instance, or a partition group whose
member the chain's key chooses; `factory:` instead names a factory whose
instance the key names, started when absent unless `create: false`.
`deadline_ms:` is required and positive; `channel:` names the callee's
entry channel the request enters on, and `answer_channel:` the callee's
channel whose output answers (its first output for the request when
unset).

Answers a handle on the Response channel, the entry channel
`<channel>.response`, whose values are `%Flezha.SDK.Response{}`: the answer
read by `decode:` as `input_with/3`'s parser reads (the raw bytes by
default). A call is not an edge, so the Response channel may feed the
tasklet that made the call. A channel ends in one output or one call
(`ChannelEndsTwice`).

    turns |> call("pricing", deadline_ms: 300) |> into(Price)
    turns |> key_by(& &1.session) |> call(factory: "agent", deadline_ms: 60_000, create: false)

# `channel_name`

```elixir
@spec channel_name(map(), {:entry, String.t()} | {:node, reference()}) :: String.t()
```

The channel a source is named by on the wire: an entry's own name, a node's
`named/2` name, the name its `output` declared, else its tasklet's.

# `filter`

# `flat_map`

# `from`

```elixir
@spec from(Flezha.SDK.Merge.t(), Flezha.SDK.Stream.t(), (term() -&gt; term())) ::
  Flezha.SDK.Merge.t()
```

Add a source stream to a merge and the mapping that lifts its value into the
merged input. The mapping is recorded on that source's own chain, after any
transform the handle already carries.

# `input`

Entry channel: its key and its tasklet receive the payload decoded by
`codec:`, which the relay's API also announces. `:json`, the default, hands
the decoded JSON value (`nil` for an empty payload), `:text` the payload as
a UTF-8 string, and `:bytes` the payload as it arrives; `:msgpack` hands the
payload as it arrives too, since this SDK carries no MessagePack decoder. A
payload its codec cannot read is unreadable. `description:` says what the
channel is for, and `schema:` is a `:json` channel's JSON Schema, a map with
a `type`; any other schema refuses the class as `SchemaInvalid`.

    c |> input("place", description: "Places an order", schema: %{"type" => "object"})

# `input_with`

Entry channel with a custom parser, `{:ok, value}` or `{:error, reason}`; a
parse failure rides the entry ladder. `description:` is announced as
`input/3` announces it; the API names the codec `custom`, so a `schema:`
refuses the class as `SchemaInvalid`.

# `into`

```elixir
@spec into(Flezha.SDK.Stream.t() | Flezha.SDK.Merge.t(), module()) ::
  Flezha.SDK.Stream.t()
```

Route into a tasklet node, or close a merge on one. Unnamed mentions of one
module unify (fan-in).

# `into_named`

```elixir
@spec into_named(Flezha.SDK.Stream.t() | Flezha.SDK.Merge.t(), module(), String.t()) ::
  Flezha.SDK.Stream.t()
```

Route into a tasklet node named `name`, or close a merge on one: `name` is
the node's identity, which its Cells are named by in place of the tasklet's
`name/0`, so one module can back two nodes. Identity collisions refuse.

# `key_by`

```elixir
@spec key_by(Flezha.SDK.Stream.t() | Flezha.SDK.Merge.t(), (term() -&gt; term())) ::
  Flezha.SDK.Stream.t() | Flezha.SDK.Merge.t()
```

Key an edge, or key a merged node on the mapped value.

A merge's key is declared once, on the merge, and runs on every source's
mapped value. A second declaration is recorded as a second key on every
source, which the check refuses as `DuplicateKeyBy`.

# `keyed`

Declare output identity on the plugin wire this handle reaches — the
reference's `keyed()`: its frames carry `flz.cell`, `flz.seq` and
`flz.key`. Refused on an entry handle, which has no output of its own to
stamp; a chain declared keyed that reaches no `output` is refused by the check
as `KeyedChannelWithoutPluginWire`.

# `map`

# `merge`

```elixir
@spec merge(t()) :: Flezha.SDK.Merge.t()
```

Open a merge: one tasklet fed by several source streams, each under its own
mapping, closed with `into/2`.

    c
    |> merge()
    |> from(ticks, &{:tick, &1})
    |> from(quotes, &{:quote, &1})
    |> key_by(&symbol_of/1)
    |> into(Detector)

One import writes the whole pipe: `from/3`, `key_by/2`, `into/2` and
`into_named/3` all dispatch on the handle they receive, so a merge step needs
no module prefix of its own.

# `named`

Name the channel the source tasklet node produces on. The tasklet keeps its
own name, or the identity `into_named/3` gave it.

# `node_channel_name`

The same rule off a node record: `named/2`'s name, the first name its outputs
wrote, else the tasklet's own name.

# `node_tasklet_name`

The name a node's Cells are identified by: `into_named/3`'s, else the tasklet's.

# `on_full`

# `output`

Terminate the channel on the plugin layer with default serialization
(bytes/UTF-8/JSON), naming it. The string written here IS the channel's name
everywhere one appears; `output/1` leaves it derived as it would be without an
output at all. Returns the handle.

# `output_with`

`output/2` with a custom serializer. Returns the handle, so a channel carrying
an output AND a consumer (the hybrid shape) is one pipeline.

# `retry_backoff_ms`

# `retry_max`

# `route_latest`

```elixir
@spec route_latest(t(), Flezha.SDK.Stream.t(), Flezha.SDK.Stream.t(), (term() -&gt;
                                                                   String.t())) ::
  Flezha.SDK.Stream.t()
```

Hand every key `key_fn` reads off `traffic` each distinct document on
`documents` once, on that key's own next traffic item, as `{key, document}`.
Expands to a merge into the packaged `Flezha.SDK.RouteLatest` node, named
`route_latest`, and answers its output stream.

# `route_latest_named`

```elixir
@spec route_latest_named(
  t(),
  Flezha.SDK.Stream.t(),
  Flezha.SDK.Stream.t(),
  (term() -&gt; String.t()),
  String.t()
) :: Flezha.SDK.Stream.t()
```

`route_latest/4` into a router node of the given name.

---

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