node.js - Reading CSV file and sending data in intervals with websockets (Node, Socket.io) -

i'm relatively new node , express.js. i'm trying create websocket server push csv data in irregular intervals stored in file itself, line after line. csv structure this: [timeout [ms], data1, data2, data3 ...]

i've created websocket server communicates client.

i'm looking best solution this: 1. read line of csv file 2. send line websockets 3. pause reading period of time stored in first value of row 4. resume reading after interval has passed, , step 1.

so far, got far (please feel free trash code might wrong - said, i'm new it. seems pause() doesn't anything.

var $    = require('jquery') ,csv = require('csv');  exports.index = function(server){   var io   = require('socket.io').listen(server);    io.sockets.on('connection', function (socket) {    socket.on('starttransmission', function(msg) {     csv()     .from.path('c:/dev/node_express/csv/test.csv', { delimiter: ',', escape: '"' })     .on('record', function(row,index){       var rowarray = $.parsejson(json.stringify(row));       var json = {},           = this;         $.each(rowarray, function(i,value){           json[keys[i]] = value;         });         socket.emit('transmitdatadata', json);         //this.pause(); //i guess around here i'd pause          // settimeout(function(){         //   that.resume();  //and resume here after timeout, stored in first value (rowarray[0])             // }, rowarray[0]);      }); }); }); }; 

the commented out code unfortunately not work - data sent immediately, row after row, function doesn't pause

i ran same sort of thing use case. issue calling pause() on stream pauses underlying stream reading not csv record parsing, record event can called remainder of records made last read stream chunk. synchronized them, in case, this:

var rows=0, actions=0;  stream.on('record', function(row, index){                                                                       rows++;                                      // pause here, expect more record events until raw read stream exhausted     stream.pause();      runner.do(row, function(err, result) {                                                           // when actions have caught rows read, read more rows.         if (actions==rows) {             stream.resume();         }                         }); }); 

in case, i'd buffer rows , release them timer. here's untested re-factoring give idea of mean:

var $ = require('jquery'),     csv = require('csv');  exports.index = function(server){    var io = require('socket.io').listen(server);   io.sockets.on('connection', function (socket) {        socket.on('starttransmission', function(msg) {          var timer=null, buffered=[], stream=csv().from.path('c:/dev/node_express/csv/test.csv', { delimiter: ',', escape: '"' });          function transmit(row) {                     socket.emit('transmitdatadata', row);                                              }                 function drain(timeout) {                                                                 if (!timer) {                 timer = settimeout(function() {                                                         timer = null;                     if (buffered.length<=1) { // more rows ahead of time don't run out. otherwise, skip beat.                         stream.resume(); // more rows                     } else {                                                 var row = buffered.shift();                         transmit(row);                         drain(row[0]);                                             }                  }, timeout);                            }                         }          stream.on('record', function(row,index){                                     stream.pause();                                                                                                if (index == 0) {                                             transmit(row);                                                            } else {                                             buffered.push(row);                                                }                                                                    drain(row[0]); // assuming row[0] contains timeout value.                                                                           });          stream.on('end', function() {             // no more rows. wait buffer empty, cleanup.         });          stream.on('error', function() {             // handle error.         });      }); };