Rxjs 双工流
Rxjs Duplex Streams
RxJs 有没有什么方法可以创建一个 Observable/Observer 对象,该对象具有接收事件和发出事件的逻辑,类似于 Rx.Subject
除了没有结合 Observable 和 Observer,所以当事件发生时从 Observable 的主体内部发出,它们不由 Observer 主体函数处理。
像这样:
var someSystem = /* some system that receives and emits events */;
function someFn() {
return ObservableObserver((observer, observable) => {
observable.subscribe((e) => someSystem.emit(e));
someSystem.on("data", (d) => observer.onNext(d));
});
}
const mySystem = someFn();
mySystem.subscribe(
(e) => {
console.log("Received from 'someSystem':", e);
mySystem.onNext("sending this to 'someSystem'");
}
);
编辑:
我相信我正在寻找的与 Node.js 双工流相似:
https://nodejs.org/api/stream.html#stream_class_stream_duplex
如果我按照您的要求,您可以使用 Subject#create
方法:
function someFn() {
return Subject.create(
Observer.create((e) => someSystem.emit(e)),
Observable.fromEvent(someSystem, 'data')
);
}
const mySystem = someFn()
source.subscribe(mySystem);
mySystem.subscribe((e) => {
console.log("Received from 'someSystem':", e);
});
RxJs 有没有什么方法可以创建一个 Observable/Observer 对象,该对象具有接收事件和发出事件的逻辑,类似于 Rx.Subject
除了没有结合 Observable 和 Observer,所以当事件发生时从 Observable 的主体内部发出,它们不由 Observer 主体函数处理。
像这样:
var someSystem = /* some system that receives and emits events */;
function someFn() {
return ObservableObserver((observer, observable) => {
observable.subscribe((e) => someSystem.emit(e));
someSystem.on("data", (d) => observer.onNext(d));
});
}
const mySystem = someFn();
mySystem.subscribe(
(e) => {
console.log("Received from 'someSystem':", e);
mySystem.onNext("sending this to 'someSystem'");
}
);
编辑:
我相信我正在寻找的与 Node.js 双工流相似:
https://nodejs.org/api/stream.html#stream_class_stream_duplex
如果我按照您的要求,您可以使用 Subject#create
方法:
function someFn() {
return Subject.create(
Observer.create((e) => someSystem.emit(e)),
Observable.fromEvent(someSystem, 'data')
);
}
const mySystem = someFn()
source.subscribe(mySystem);
mySystem.subscribe((e) => {
console.log("Received from 'someSystem':", e);
});