I'm trying to read some file by opening read stream and send chunks of the file through ZMQ to another process to consume them. The stream is working like it should, however when I start the worker, it doesn't see the data that's been sent.
I tried sending data through socket every 500ms, not in a callback, and when I start the worker it collects all previous chunks of data:
sender = zmq.socket('push')
setInterval(() ->
console.log('sending work');
sender.send('some work')
, 500)
receiver = zmq.socket("pull")
receiver.on "message", (msg) ->
console.log('work is here: %s', msg.toString())
Outputs:
sending work
sending work
sending work
sending work
sending work
// here I start the worker
sending work
work is here: some work
work is here: some work
work is here: some work
work is here: some work
work is here: some work
work is here: some work
sending work
work is here: some work
sending work
work is here: some work
sending work
work is here: some work
So, when the worker starts, it begins with pulling all the previous data and then it pulls it every time sth new comes in. This does not apply when I do this:
readStream = fs.createReadStream("./data/pg2701.txt", {'bufferSize': 100 * 1024})
readStream.on "data", (data) ->
console.log('sending work');
sender.send('some work'); // I'd send 'data' if it worked..
In this scenario, the worker doesn't pull any data at all. Are those kind of sockets supposed to create a queue or not? What am I missing here?
Yes, push socket is blocking until HWM is reached, and there's nobody to send to. Maybe the sender hasn't bound yet, try something like this:
sender.bind('address', function(err) {
if (err) throw err;
console.log('sender bound!');
// the readStream code.
}
also a connect is missing from your code example, I bet it's there, but maybe you forgot it.