讀取CSV文件並使用websockets（Node，Socket.io）間隔發送數據

Question

我對Node和Express.js比較陌生。我正在嘗試創建一個websocket服務器，以一行一行的方式將CSV數據以不規則的間隔存儲在文件本身中。 的CSV結構是這樣的： [超時[毫秒]，數據1，數據2，數據3 ...]

我已經成功地創建了一個websocket服務器與客戶端通信。

我在尋找有效的做這樣的事情一個最好的解決辦法：1。 讀取一行的CSV文件 2.發線使用WebSockets 3.暫停讀取存儲在一段時間該行的第一個值 4.在間隔過去後恢復讀數，並返回步驟1.

到目前爲止，我得到了這麼多（請隨時將我的代碼完全垃圾，因爲它可能是非常錯誤的 - 正如我所說，我是新來的，看起來像暫停（）沒有做任何事情，

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 = {}, 
      that = this; 
     $.each(rowArray, function(i,value){ 
      json[keys[i]] = value; 
     }); 
     socket.emit('transmitDataData', json); 
     //this.pause(); //I guess around here is where I'd like to pause 
     // setTimeout(function(){ 
     // that.resume(); //and resume here after the timeout, stored in the first value (rowArray[0])  
     // }, rowArray[0]); 

    }); 
}); 
}); 
}; 

註釋掉的代碼不幸不起作用 - 所有數據立即發送，一排一排，功能不會暫停

Answer 1

我遇到了與其他用例相同類型的事情。問題在於調用流上的pause（）會暫停基礎流讀取而不是csv記錄解析，因此可以使用構成最後一個讀取流塊的記錄的其餘部分調用record事件。我同步它們，在我的情況下，像這樣：

var rows=0, actions=0; 

stream.on('record', function(row, index){                 

    rows++;         

    // pause here, but expect more record events until the raw read stream is exhausted 
    stream.pause(); 

    runner.do(row, function(err, result) {             

     // when actions have caught up to rows read, read more rows. 
     if (actions==rows) { 
      stream.resume(); 
     }      
    }); 
}); 

在你的情況，我會緩衝行並釋放他們的計時器。這裏有一個未經測試的重新分解只是爲了讓你知道我的意思：

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) { // get more rows ahead of time so we don't run out. otherwise, we could skip a beat. 
         stream.resume(); // get 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 a timeout value.                 
     }); 

     stream.on('end', function() { 
      // no more rows. wait for buffer to empty, then cleanup. 
     }); 

     stream.on('error', function() { 
      // handle error. 
     }); 

    }); 
};