Streaming support

[robertkowalski]

Right now the http and ws adapters for grenache are both buffering data.

That is becoming a problem, when we send larger data. One contract all transports have with each other is that the message format it always in JSON. The message contains a key and a request / msg id. Our current use case for streaming is mainly binary data, like big file uploads.

Recently I thought about streaming integration into the Transports, but I ran into multiple problems.

Right now our system catches the request event emitted by the http module, then starts buffering and then emits itself a request event with the buffered data.

A typical handler looks like this:

service.on('request', (rid, key, payload, handler) => {
  // rid, key also comes from the buffered data, which is always JSON
  // payload is from the buffered data
  handler.reply(null, 'world')
})

The payload is buffered here:

grenache-nodejs-http/lib/TransportRPCServer.js

Lines 38 to 46 in ddc25cc

	socket.on('request', (req, rep) => {
	let body = []
	req.on('data', (chunk) => {
	body.push(chunk)
	}).on('end', () => {
	body = Buffer.concat(body).toString()
	const cert = secure ? req.socket.getPeerCertificate() : undefined
	this.handleRequest(

How could a streaming API look like?

service.on('stream', (req, res) => {
  // rid, key would either have to be parsed from the stream or they are not needed... (?)
  req.pipe(consumer).on('end', () => { res.end('') })
})

internally we would take a look if we are about to receive a stream:

    socket.on('request', (req, rep) => {
      if (req.headers['transfer-encoding'] === 'chunked') {
        this.handleStream(req, rep)
        return
      }

      // old buffer code follows here...

handleStream would then emit a stream event:

handleStream (req, res) {
  this.emit('stream', req, res)
}

Thoughts and considerations:

1. overwriting the request event

right now we catch the request event from the socket, then do the buffering and emit our own event named request with the buffered data. internally we could rename our request event to grenache-request and additionally just proxy the original request event. that allows users to build all kind of different, specialized request handlers. or maybe, to not introduce a breaking change, we additionally emit the original request event under a different name, e.g. original-request. this way users of the lib can build very specific, custom request handlers by accessing the raw request object.

2. svc-js integration of streaming

right now we are searching for the corresponding method on the object and just call the method with the data that was buffered before: https://github.com/bitfinexcom/bfx-wrk-api/blob/c695ff36a801103bad6bb3c4fa059f38c7081f9a/api/base.js#L41-L66

With stream integration we have to pass the stream or chunks into the handler. the big question is how we can integrate that in a good way.

we could take the current api handler methods and let them emit data. still unclear is how to stream data back:

foo (space, args, cbOrStream) {
  this.on('data') // ...

  cbOrStream.end('') // uh?!
}

another way is to have two handlers:

foo (space, args, cb) {}
fooStream (space, req, res) {}

bfx-wrk-api would then check if the method exists and in an error case return something like ERR_API_NO_STREAM_METHOD. So it would depend on the service developer if they want to support streaming.

what do you think about the svc-js integration? do you have a more elegant solution?

[robertkowalski]

update: @mafintosh mentioned that steam detection on http content type is not a good idea. the request should always be a stream, and payload parsing etc would then be handled later.

[robertkowalski]

current main question: how to integrate the streams into the current svc-js and its request handlers?