Node Collab Summit Berlin 2019

BOB (Future? Streams)

1. Why
2. Goals
3. Api
4. Current Status
6. Discussion

1. Why reinvent streams?

The Streams user experiance is bad

Streams performance is underwhelming

2. Goals

Some terms:

• Consumer / "Sink" - API end where data goes to.


• Producer / "Source" - API end where data comes from.


• "Protocol" - combination of a Consumer & Producer

Protocol Goals

• Pull-based
• Binary-only
• Stateless (in-protocol)
• One-to-one (no events)
• Timing agnostic
• No buffering (in-protocol)
• In-line errors and "end"

The Current Proposed API

Sink API

class Sink {
  constructor (opts) {}

  bindSource (source, bindCb) {
    this.source = source

  }

  next (status, error, buffer, bytes) {
    // do stuff with received data

  }
}
    

Sink API

class Sink {
  constructor (opts) {}

  bindSource (source, bindCb) {
    this.source = source
    this.source.bindSink(this)
  }

  next (status, error, buffer, bytes) {
    // do stuff with received data
    this.source.pull(null, buffer)
  }
}
    

Source API

class Source {
  constructor (opts) {}

  bindSink (sink) {
    this.sink = sink
  }

  pull (error, buffer) {
    // get data


  }
}
    

Source API

class Source {
  constructor (opts) {}

  bindSink (sink) {
    this.sink = sink
  }

  pull (error, buffer) {
    // get data
    this.sink.next(status_type.continue, null,
                   buffer, bytesWritten)
  }
}
    

class PassThrough {
  bindSource (source) {
    source.bindSink(this)
    this.source = source
    return this
  }

  bindSink (sink) {
    this.sink = sink
  }

  next (status, error, buffer, bytes) {
    this.sink.next(status, error, buffer, bytes)
  }

  pull (error, buffer) {
    this.source.pull(error, buffer)
  }
}
    

Composing a source & sink

const source = new Source(opts)
const sink = new Sink(opts)



sink.bindSource(source, error => {
  // The stream is finished when this is called.
})
    

Composition with a transform

const source = new Source(opts)
const sink = new Sink(opts)
const transform = new Transform(opts)

transform.bindSource(source)
sink.bindSource(transform, error => {
  // The stream is finished when this is called.
})
    

github.com/Fishrock123/bob

/diagrams

Current Status

Current Modules

• bob-status
• fs-source
• fs-sink
• zlib-transform
• github.com/Fishrock123/socket

Current API Performance
For any single stream

• ~30% decrease of CPU time in bad cases
• Up to 8x decrease in good cases
• Should correlate to overall throughput but may not be exact.

Discussion!

github.com/Fishrock123/bob

Open Questions

Is this API likeable?

Do we need a buffer pooling helper?

Is it too API-pattern focused?

How to enforce single active request?

Where to start implementing in core?

Libuv pull streams?