Streaming

The streaming project shows a stream of packets flowing through streaming nodes and being collected back into one array. It is the example that goes with the streams language page.

The complete project ships in this repository and can be copied and run as-is:

• streaming.fbp — the main flow
• split.fbp, filter.fbp — variations
• stream.echo.yml, stream.split.yml, stream.filter.yml, console.log.yml — the node definitions
• package.json

The flow

title: Streaming echo

provider ./{ns}.{name}.yml

["a", "b", "c"] => in Echo(stream/echo)
Echo out >= msg Log(console/log)

Two arrows do the work:

• => splits the array into a stream of packets — one delivery per element.
• >= collects the stream back into a single array when the batch closes.

The nodes in between are the interesting part.

Echo(stream/echo)

A body-form streaming node. streaming: true tells the runtime to own the loop; the fn runs once per packet:

title: Stream Echo
ns: stream
name: echo
streaming: true
ports:
  input:
    in:
      type: any
  output:
    out:
      type: any
fn: |
  emit(packet.read())

packet is the current packet, emit sends an output packet. Echo runs three times and echoes "a", then "b", then "c". On the way out, the >= connection buffers those packets and hands Log the whole array at once.

Split(stream/split)

State lives in state, not a closure. The fn toggles state.inCode on every code fence, and routes each chunk to a named port with an object-map emit:

fn: |
  let rest = String(packet.read());
  let i = rest.indexOf('```');
  while (i !== -1) {
    const head = rest.slice(0, i);
    if (head) emit(state.inCode ? {code: head} : {text: head});
    state.inCode = !state.inCode;
    rest = rest.slice(i + 3);
    i = rest.indexOf('```');
  }
  if (rest) emit(state.inCode ? {code: rest} : {text: rest});

emit({text: head}) sends to the text port; emit({code: head}) to code. Each named port can be collected separately:

title: Streaming split

provider ./{ns}.{name}.yml

["text> ```js", "const x = 1;", "``` <text"] => in Split(stream/split)
Split text >= msg TextLog(console/log)
Split code >= msg CodeLog(console/log)

Filter(stream/filter)

Conditional emit — a packet is either emitted or passes silently:

fn: |
  const m = String(packet.read()).match(/@tool\((\w+)\)/);
  if (m) emit(m[1]);

title: Streaming filter

provider ./{ns}.{name}.yml

["const x = 1;", "@tool(readFile)(\"a.txt\");", "y"] => in Filter(stream/filter)
Filter out >= msg Log(console/log)

Run it

From inside the project directory:

$ fbpx run streaming.fbp
[ 'a', 'b', 'c' ]

$ fbpx run split.fbp
[ 'text> ', ' <text' ]
[ 'js', 'const x = 1;' ]

$ fbpx run filter.fbp
[ 'readFile' ]

There are no npm dependencies to install — the nodes are plain body-form definitions. fbpx install is only needed when a node declares dependencies.npm.