我正在尝试解决 Nodejs 流挑战。我已多次阅读有关流的节点文档,并实施了不同的尝试来解决挑战。尝试双工、转换、可读和可写:)
我有多个 HTTP 可读流,目标是将数据发送到单个管道,并具有背压工作。我认为这张照片有助于解释挑战:
更新(2017 年 9 月 13 日)。再次阅读文档后,我正在实现一个自定义的书面双工流。
我正在尝试解决 Nodejs 流挑战。我已多次阅读有关流的节点文档,并实施了不同的尝试来解决挑战。尝试双工、转换、可读和可写:)
我有多个 HTTP 可读流,目标是将数据发送到单个管道,并具有背压工作。我认为这张照片有助于解释挑战:
更新(2017 年 9 月 13 日)。再次阅读文档后,我正在实现一个自定义的书面双工流。
This represents a great usecase for a duplex stream, combined with manuel flow control of the HTTP stream.
I have written a custom duplex stream, where the readable and writable part, is structured like this:
If you are interested in the specific code for the duplex stream, please send me a PM.
The code could look something like this (but it's pretty old, and could probably be simplified even more):
import 'rxjs/add/operator/skip';
import 'rxjs/add/operator/take';
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import * as stream from 'stream';
import { logger, streamInspector } from '../shared';
export class DuplexStreamLinker extends stream.Duplex {
public readCount: number = 0;
public acceptDataCount: number = 0;
public acceptData$: BehaviorSubject<boolean>;
public streamName: string;
constructor(options) {
super(options);
this.streamName = this.constructor.name;
this.acceptData$ = new BehaviorSubject(false);
streamInspector(this, this.constructor.name);
}
public _read(size) {
this.readCount++;
this.acceptData$.next(true);
}
public _write(chunk, encoding, cb) {
const acceptData = this.acceptData$.getValue();
if (acceptData) {
cb(this.pushData(chunk));
} else {
this.acceptData$.skip(1).take(1).subscribe(() => {
logger.silly('I dont fire...');
this.acceptDataCount++;
cb(this.pushData(chunk));
});
}
}
public endReadableStream() {
logger.debug('DuplexStreamLinker@endReadableStream was called!');
this.end();
this.push(null);
}
public _final(cb) {
logger.debug('DuplexStreamLinker@_final was called!');
cb(null);
}
private pushData(chunk): null | Error {
const ok = this.push(chunk);
if (ok === false) { this.acceptData$.next(false); }
return null;
}
}