Any plans for real streaming?

[pocesar]

subscribe is good but it offer no backpressure mechanism that a real stream implementation does. are there any plans for having a .pipe() method for usage with streams? I don't know if neo4j has native streaming (cursors) implemented though, it's a wild guess that it could implemented

[lutovich]

Hi @pocesar,

We do have plans for back pressure. Neo4j database has cursors that can be lazily consumed. However, bolt protocol does not expose any back pressure messages so clients can't tell server to stop sending.

Could you please describe what kind of problems do you experience with #subscribe() and how back pressure would help? It would be nice to know more about real world use-cases.

[pocesar]

I'm having to stream a lot of event objects to node.js and filter them using mongodb operators in-memory (ie using an array), and after reaching the memory limit from the docker container, it crashes with OOM, and having problems with the order in the arrays having a racing condition (because of the async comparator function plus latency) and the order of the original sort is lost because it depends on the order by clause in neo4j cypher. I could use an external queue using rabbitmq but that would add so much complexity for a simple task.

the reason I'm using an in-memory comparator is because I don't want to mirror the neo4j content to a mongodb database, because that would be a waste of resources, and I need to define, per user, some internal filters (defined as json objects inside neo4j as well, such as [{ operator: '$lte', field: 'birth', value: 1500303646 }, { operator: '$eq', field: 'gender', value: 'm' }]) and add them from outside without having to touch server code. If I would be using a stream interface, I could compare and drop them as they come, and stream already filtered objects directly to the express response without having to keep anything in memory at all

[lutovich]

@pocesar thanks for explanation! Do you use #subscribe() right now and process records as they arrive in onNext() callback?

[pocesar]

yeah, that's what I'm doing right now. since the objects are quite large, after some hundreds kept in memory, it quickly exhausts the available memory without any real traffic (that's the main concern)

if I had a way to do native streaming (tell the server to 'pause' sending new records), JSON streaming with express is a breeze using res.write and res.end and would not require creating an intermediate array, and I could happily .pipe to the res after initiating the streaming. oboe.js on the client side would manage to parse mid-way through as the JSON object come through the pipe

I assume onNext would receive a second parameter called next, that if the callback function had 2 parameters, assume "streaming" / "async" and would wait for the next function to be called (and if next called with an error, abort the stream). this change would be backwards compatible and would not require a major version bump, everyone is happy. but I guess would require some changes to the protocol itself

res.setHeader('content-type', 'application/json')
res.write('[')
tx.subscribe({
  onNext(record, next) {
    const asyncStream = asyncComparator(record, currentUser)
    asyncStream.pipe(res, { end: false })
    asyncStream.on('end', next)
  },
  onError() {
    res.end(']') // empty response, ie: []
  },
  onCompleted()  {
    res.end(']')
  }
})

that is a proposal in an ideal world, but if neo4j driver could expose an stream() method to generate a readable stream without any additional boilerplate (and handle errors internally) would be the best, but wouldn't be browser ready without some serious effort.

EDIT: after reading the stream-observer code I now realize that it ALSO keeps a copy of the received records during the subscribe lifecycle, so that's the reason why it's OOM'ing so fast with just some thousands objects... so my best guess would, indeed to implement a stream() method, so it would not refurbish the "concat" nature of stream-observer

[technige]

Implement flow control and back pressure throughout Bolt stack

[technige]

Flow Control, Back Pressure & Reactive Drivers