대리 스트림이 생성되었을 때 사용할 수없는 다른 스트림을 래핑 할 수있는 Node.js 대리 스트림을 어떻게 만들 수 있습니까?

Question

간단한 HTTP 요청을 만들고 스트림을 다시 가져올 수 있습니다.

하지만 HTTP 요청을 한 다음 폴링하여 데이터 준비가되었는지 확인한 다음 데이터를 다시 요청하십시오.

내가 스트림을 반환하는 하나의 방법에있는 모든 그래서 내가 할 수있는 그렇게하고 싶습니다 : 나는 multiStepMethod이 대리 읽을 수를 반환해야

multiStepMethod(options, (err, stream) => { 
    stream.pipe(wherever); 
}) 

:

multiStepMethod(options).pipe(wherever); 

대신에 일부 이벤트를 기다린 다음 (현재 사용 가능한) 스트림을 감싸고 파이프로 데이터를 보내기 시작합니다.

Answer 1

#!/usr/bin/env node 

'use strict'; 

const stream = require('stream'); 

// This is an example of a 'readable' stream that has to go through a multi- 
// step process to finally get the actual readable stream. So we are 
// asynchronously wrapping another readable stream. 

// The key to simplicity here was to use a transform stream instead of a 
// readable stream because it allows us to pipe the stream to ourselves. 
class ReadableWrappingTransform extends stream.Transform { 
    constructor() { 
    super({ 
     objectMode: true, 
     // Our _transform method doesn't really do anything and we don't want to 
     // hog up any more additional memory than necessary. 
     highWaterMark: 1 
    }); 
    process.nextTick(() => { 
     if (new Date().getTime() % 5 === 1) { 

     // Here we simulate an error that happened somewhere in the multi-step 
     // process to get the final stream. So we just emit 'error' and we're 
     // done. 
     this.emit('error', new Error('Could not get the stream.')); 

     //Assuming based on the node docs that we should not emit 
     // 'close' or 'end' on error. If we do emit 'end', it will trigger the 
     // writable's 'finish' event, which is probably not desired. You either 
     // want an 'error' OR a 'finish'. 

     // NODE END EVENT DOCS 
     // The 'end' event is emitted when there is no more data to be consumed 
     // from the stream. 

     // Note: The 'end' event will not be emitted unless the data is 
     // completely consumed. This can be accomplished by switching the stream 
     // into flowing mode, or by calling stream.read() repeatedly until all 
     // data has been consumed. 

     // this.emit('end'); 

     // NODE CLOSE EVENT DOCS 
     // The 'close' event is emitted when the stream and any of its 
     // underlying resources (a file descriptor, for example) have been 
     // closed. The event indicates that no more events will be emitted, and 
     // no further computation will occur. 

     // Not all Readable streams will emit the 'close' event. 

     // this.emit('close'); 
     } else { 
     // We successfully got the stream we wanted after a long, hard, multi- 
     // step process, so first we need to copy all our listeners over to it 
     // -- NOT. 

     // ['close', 'data', 'end', 'error'].forEach((eventName) => { 
     // this.listeners(eventName).forEach((l) => { 
     //  readable.on(eventName, l); 
     // }); 
     // }); 

     // Turns out that .pipe propagates ALL listeners EXCEPT the 'error' 
     // listener. What's up with that !?! If we copy any of the others we 
     // get double the events -- including double the data. So here we just 
     // copy over the 'error' listener to make sure we get 'error' events. 
     ['error'].forEach((eventName) => { 
      this.listeners(eventName).forEach((l) => { 
      readable.on(eventName, l); 
      }); 
     }); 

     // Then just pipe the final readable to ourselves, and we are good. 
     readable 
      .pipe(this); 
     } 
    }); 
    } 
    _transform(data, encoding, callback) { 
    // Nothing special to do here just pass along the data. 
    this.push(data); 
    callback(); 
    } 
} 

// This is just a very unreliable test readable stream. 
const readable = new stream.Readable({ 
    objectMode: true, 
    read() { 
    for (let i = 0; i < 10; i++) { 
     if (new Date().getTime() % 13 === 1) { 
     this.__err = new Error('Sorry, error reading data.'); 
     this.emit('error', this.__err); 
     return; 
     } 
     this.push({ 
     Name: `Mikey ${i}` 
     }); 
    } 
    this.push(null); 
    } 
}); 

// Any old writable that we can pipe to. 
const writable = new stream.Writable({ 
    objectMode: true, 
    write(chunk, encoding, callback) { 
    console.log(chunk, encoding); 
    callback(); 
    } 
}); 

new ReadableWrappingTransform() 
    // if your stream emits close you get close. 
    .on('close',() => { 
    console.error('CLOSE'); 
    }) 
    // if you push null you get end from read. 
    .on('end',() => { 
    console.error('END'); 
    }) 
    // error needs to be both places !?! seriously node? 
    .on('error', (error) => { 
    console.error('ERROR', error); 
    }) 
    // Finish does no good here. It's a writable event. 
    // .on('finish',() => { 
    // console.error('FINISH'); 
    // }) 
    .pipe(writable) 
    // Close and End do no good here, they are readable events. 
    // They are not propagated to the writable. 
    // 
    // // if your stream emits close you get close. 
    // .on('close',() => { 
    // console.error('CLOSE'); 
    // }) 
    // // if you push null you get end from read. 
    // .on('end',() => { 
    // console.error('END'); 
    // }) 
    // error needs to be both places !?! seriously node? 
    .on('error', (error) => { 
    console.error('ERROR', error); 
    }) 
    // you should always get either finish or error or something was done 
    // incorrectly. 
    .on('finish',() => { 
    console.error('FINISH'); 
    });