Flow & Generics

Source: tutorial/02_flow_and_generics See in Playground

Generates the integers 1..upper, splits them around threshold, nudges each half by a different offset, and merges the two halves back into a single stream, using two small generic treatments defined right in the file and reused twice each.

Running

cd tutorial/02_flow_and_generics
melodium run Compo.toml --upper 10 --threshold 5 --offset_above 100 --offset_below=-100

Running with --upper 10 --threshold 5, the at-or-below-threshold stream logs 1, 2, 3, 4, 5 and above-threshold logs 6, 7, 8, 9, 10: the symmetric split that 1..10 around 5 implies.

Optional: add --api-report and an API token (MELODIUM_API_TOKEN) to see this run’s full trace on Cadence.CI .

How it works

This example uses no models either: everything is stateless treatments, functions, and two custom generic treatments.

Data flow

Building the sequence

1..upper is built with generate, which produces upper placeholder values, and count, which numbers them starting at 1:

howMany:   emit<u128>(value=upper)
seed:      generate<u128>(data=0)
numbering: count<u128>()
asI64:     saturatingToI64<u128>()
 
startup.trigger -> howMany.trigger,emit -> seed.length,stream -> numbering.stream
numbering.count -> asI64.value,into -> split.value

A generic treatment for splitting: aboveThreshold<N: PartialOrder>

aboveThreshold is instantiated once (as split) but is written generically: it never mentions i64 in its body, and works for any type that can be ordered (PartialOrder). Inside, it builds a same-length stream of the threshold value with toVoid and fill, so it can compare element-by-element with greaterThan, then uses that boolean stream to filter the input into above/below:

treatment aboveThreshold<N: PartialOrder>(const threshold: N)
  input  value: Stream<N>
  output above: Stream<N>
  output below: Stream<N>
{
    asVoid:      toVoid<N>()
    thresholdSt: fill<N>(value=threshold)
    isAbove:     greaterThan<N>()
    partition:   filter<N>()
 
    Self.value  -> asVoid.value
    asVoid.iter -> thresholdSt.pattern
 
    Self.value          -> isAbove.a
    thresholdSt.filled  -> isAbove.b
 
    Self.value -> partition.value
    isAbove.is -> partition.select
 
    partition.accepted -> Self.above
    partition.rejected -> Self.below
}

Self.value is read three times here (into asVoid, isAbove, and partition): reading the same input more than once is allowed, only writing to the same input twice is not.

A second generic treatment: shift<N: Add>

shift is instantiated twice (bumpAbove, bumpBelow) with two different offset values: same treatment, same toVoid + fill trick, this time feeding add:

treatment shift<N: Add>(const offset: N)
  input  value:   Stream<N>
  output shifted: Stream<N>
{
    asVoid:   toVoid<N>()
    offsetSt: fill<N>(value=offset)
    adder:    add<N>()
 
    Self.value  -> asVoid.value
    asVoid.iter -> offsetSt.pattern
 
    Self.value      -> adder.a
    offsetSt.filled -> adder.b
    adder.sum -> Self.shifted
}

bumpAbove: shift<i64>(offset=offset_above)
bumpBelow: shift<i64>(offset=offset_below)
split.above -> bumpAbove.value
split.below -> bumpBelow.value

Merging back into one stream

merge interleaves the two shifted streams unpredictably: running the program twice, the log order may differ, which is expected. Mélodium streams have no implicit ordering guarantee across branches:

combined: merge<i64>()
bumpAbove.shifted -> combined.a
bumpBelow.shifted -> combined.b

A final count numbers the merged stream just to show it is a stream like any other.

Dependencies

[dependencies]
std = "0.10.3"  # core flows, logging, data structures