使用管道将数据从msssql流传输到节点

Question

我正在使用节点和node-mssql 6.0.1，并且我使用流从DB中获取大量数据并将其发送到前端。

我正在尝试按照docs的建议使用管道和流，但是我无法使其正常工作。

我的代码是

const pool = new sql.ConnectionPool(config);
const poolConnect = pool.connect();

const testStream = (ws , id) => {     
    poolConnect.then((pool) => {   
      const request = new sql.Request(pool);
      request.pipe(stream);

         request
        .input('id_param', sql.Int, parseInt(id)) 
        .query('SELECT * FROM dataTable WHERE id = @id_param ')  
 }) 

我得到stream is not defined。

我的目标是像示例here一样组合流和管道，然后使用websockets将数据发送到客户端。我什至不知道结合流和websocket是否有意义。我想不出任何组合或测试的方法

我当前仅使用流的工作代码，看起来像这样

const sql = require('mssql');
const pool = new sql.ConnectionPool(config);
const poolConnect = pool.connect();

const testStream = (ws , id) => {       
    poolConnect.then((pool) => {   
      const request = new sql.Request(pool);
      request.stream = true;   
      request
      .input('id_param', sql.Int, parseInt(id))
      .query('SELECT * FROM table WHERE id = @id_param ')  

      let rowsToProcess = [];
      let data = [];  

      request.on('row', row => {   
        rowsToProcess.push(row); 
        if (rowsToProcess.length >= 20) {  
          request.pause();
          processRows(false);
        } 
      });

      request.on('done', () => {      
        processRows(true); 
        sql.close(); 
      });

      const processRows = () => { 
          rowsToProcess.forEach((item)=>{ 
                data.push(item.name);   
                data.push(item.surname);   
                data.push(item.age);   
            });    
          ws.send(JSON.stringify({ success:true, message: data }));
          rowsToProcess = [];
          data = []; 
      }//processRows 

    }) //poolConnect.then  

我在这里想念什么？请帮助我开始使用管道和流，在Google中找不到管道/流的示例

谢谢

Answer 1

简单可写流

const { Writable } = require('stream');

const stream = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();    
  }
})

Answer 2

在您的代码中stream未定义，这就是为什么您会收到该错误的原因。您希望将其通过管道传递到WriteableStream（文件，标准输出...）。因此，尝试：

const testStream = (ws , id) => {     
    poolConnect.then((pool) => {   
      const request = new sql.Request(pool);
      request.pipe(process.stdout);

         request
        .input('id_param', sql.Int, parseInt(id)) 
        .query('SELECT * FROM dataTable WHERE id = @id_param ')  
 })

也可以通过管道传输到文件或任何其他流

const fs = require('fs');

const testStream = (ws , id) => {     
    poolConnect.then((pool) => {
      const stream = fs.createWriteStream('out.txt');  
      const request = new sql.Request(pool);
      request.pipe(stream);

      /* ... */
 })