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.
Summary
Types
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.
Functions
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.
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).
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.
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.
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.
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.
Route into a tasklet node, or close a merge on one. Unnamed mentions of one module unify (fan-in).
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 an edge, or key a merged node on the mapped value.
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.
Open a merge: one tasklet fed by several source streams, each under its own
mapping, closed with into/2.
Name the channel the source tasklet node produces on. The tasklet keeps its
own name, or the identity into_named/3 gave it.
The same rule off a node record: named/2's name, the first name its outputs
wrote, else the tasklet's own name.
The name a node's Cells are identified by: into_named/3's, else the tasklet's.
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/2 with a custom serializer. Returns the handle, so a channel carrying
an output AND a consumer (the hybrid shape) is one pipeline.
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/4 into a router node of the given name.
Types
@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.
@type t() :: %Flezha.SDK.ChannelBuilder{ref: reference()}
Functions
@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.
@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)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.
@spec from(Flezha.SDK.Merge.t(), Flezha.SDK.Stream.t(), (term() -> 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.
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"})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.
@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).
@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.
@spec key_by(Flezha.SDK.Stream.t() | Flezha.SDK.Merge.t(), (term() -> 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.
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.
@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.
Name the channel the source tasklet node produces on. The tasklet keeps its
own name, or the identity into_named/3 gave it.
The same rule off a node record: named/2's name, the first name its outputs
wrote, else the tasklet's own name.
The name a node's Cells are identified by: into_named/3's, else the tasklet's.
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/2 with a custom serializer. Returns the handle, so a channel carrying
an output AND a consumer (the hybrid shape) is one pipeline.
@spec route_latest(t(), Flezha.SDK.Stream.t(), Flezha.SDK.Stream.t(), (term() -> 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.
@spec route_latest_named( t(), Flezha.SDK.Stream.t(), Flezha.SDK.Stream.t(), (term() -> String.t()), String.t() ) :: Flezha.SDK.Stream.t()
route_latest/4 into a router node of the given name.