Node.js: can you use asynchronous functions from within streams?

Question

Consider the following:

var asyncFunction = function(data, callback) {
  doAsyncyThing(function(data){
    // do some stuff
    return callback(err)
  })
}
fs.createReadStream('eupmc_lite_metadata_2016_04_15.json')
  .pipe(JSONstream.parse())
  .on('data', asyncFunction)   // <- how to let asyncFunction complete before continuing

How does the stream know when asyncFunction has completed? Is there any way to use asynchronous functions from within streams?

I'm not sure how the stream will handle the callback since the documentation doesn't show two parameters in on('data', function). If you do want to do something fancy though, you can pause the stream, do your stuff, then resume the stream. — Dave Briand
– Dave Briand, Commented Apr 16, 2016 at 11:01
@DaveBriand are you saying categorically that you cannot do this, or are you restating the question? :) — Fergie
– Fergie, Commented Apr 16, 2016 at 13:47
Categorically you can't pass a two argument function to the data stream event. However, you can pause the stream on the data event, do some asynchronous processing, then resume the stream when your processing is complete. — Dave Briand
– Dave Briand, Commented Apr 18, 2016 at 14:42
Great! Is there a cleanish way to code this? Could you give an example? — Fergie
– Fergie, Commented Apr 19, 2016 at 20:44
Just for clarity: yes, on('data', asyncFunction) cannot deal with callbacks, since asyncFunction must be in the form function(data). My point is: "how then do you deal with callbacks?" — Fergie
– Fergie, Commented Aug 10, 2016 at 16:51

t7tran · Accepted Answer · 2022-04-19 08:23:11Z

24

Check out transform streams. They give you the ability to run async code on a chunk, and then call a callback when you are finished. Here are the docs: https://nodejs.org/api/stream.html#transform_transformchunk-encoding-callback

As a simple example, you can do something like:

const Transform = require('stream').Transform
class WorkerThing extends Transform {
    _transform(chunk, encoding, cb) {
        asyncFunction(chunk, cb)
    }
}

const workerThing = new WorkerThing()

fs.createReadStream('eupmc_lite_metadata_2016_04_15.json')
.pipe(JSONstream.parse())
.pipe(workerThing)

edited Apr 19, 2022 at 8:23

t7tran

2,1941 gold badge19 silver badges17 bronze badges

answered May 5, 2016 at 3:43

Ryan Quinn

1,2051 gold badge16 silver badges29 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Fergie Over a year ago

I didnt really understand what you were saying at first, but yes, I see now that transform streams are probably the way forward. Will try this out.

Pasalino · Accepted Answer · 2023-06-11 21:35:08Z

2

I think this is enough:

const Transform = require('node:stream').Transform

const deferTransform = new Transform({
  transform: (chunk, encoding, next) => {
    Promise.resolve(`${chunk.toString().toUpperCase()} `).then((data) =>
      next(null, data)
    );
  },
});


fs.createReadStream('eupmc_lite_metadata_2016_04_15.json')
.pipe(JSONstream.parse())
.pipe(deferTransform)

answered Jun 11, 2023 at 21:35

Pasalino

1,2021 gold badge11 silver badges15 bronze badges

Collectives™ on Stack Overflow

Node.js: can you use asynchronous functions from within streams?

2 Answers 2

1 Comment

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related