1

我对 Node 和 Express.js 比较陌生。我正在尝试创建一个 websocket 服务器以不规则的间隔推送 CSV 数据，这些数据存储在文件本身中，一行一行地。CSV 结构是这样的：[timeout [ms], data1, data2, data3 ...]

我已经成功创建了一个与客户端通信的 websocket 服务器。

我正在寻找有效执行以下操作的最佳解决方案： 1. 读取 CSV 文件的一行 2. 使用 WebSockets 发送一行 3. 将读取暂停一段时间，存储在第 4 行的第一个值中. 间隔过后继续读取，并返回步骤 1。

到目前为止，我已经走到了这一步（请随意完全丢弃我的代码，因为它可能非常错误 - 正如我所说，我是新手。看起来 pause() 没有做任何事情。

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]);

    });
});
});
};

不幸的是，注释掉的代码不起作用 - 所有数据都立即发送，逐行发送，函数不会暂停

4

1 回答 1

0

我在另一个用例中遇到了同样的事情。问题是在流上调用 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.
        });

    });
};
于 2013-08-23T03:28:26.273 回答