如何解决可观察的主题事件侦听器
我一直在研究Observables及其与EventEmitter的区别，然后偶然发现Subjects（我可以看到Angulars EventEmitter基于）。
Observable似乎是单播，而多播的主题（然后EE只是将.next封装在emit中以提供正确接口的主题）。
可观察对象似乎很容易实现
class Observable {
constructor(subscribe) {
this._subscribe = subscribe;
}
subscribe(next,complete,error) {
const observer = new Observer(next,error);
// return way to unsubscribe
return this._subscribe(observer);
}
}
Observer只是一个包装，其中添加了一些try catch和monitor isComplete，以便可以清理并停止观察。
对于我想到的主题：
class Subject {
subscribers = new Set();
constructor() {
this.observable = new Observable(observer => {
this.observer = observer;
});
this.observable.subscribe((...args) => {
this.subscribers.forEach(sub => sub(...args))
});
}
subscribe(subscriber) {
this.subscribers.add(subscriber);
}
emit(...args) {
this.observer.next(...args);
}
}
哪种类型合并到一个EventEmitter中，并用.next封装在一起，用emit包装-但是捕获Observable的observe参数似乎是错误的-就像我刚刚破解了一个解决方案。从可观察（单播）产生主题（多播）的更好方法是什么？
我尝试查看RXJS，但看不到subscribers数组是如何填充的：/
解决方法
我认为您也可以通过使用调试器来更好地理解。打开一个StackBlitz RxJS项目，创建最简单的示例（取决于您要理解的内容），然后放置一些断点。 AFAIK，使用StackBlitz可以调试TypeScript文件，这看起来很棒。
首先，Subject类extends Observable：
export class Subject<T> extends Observable<T> implements SubscriptionLike { /* ... */ }
现在让我们检查一下Observable类。
它具有著名的pipe method：
pipe(...operations: OperatorFunction<any,any>[]): Observable<any> {
return operations.length ? pipeFromArray(operations)(this) : this;
}
其中pipeFromArray被定义为as follows：
export function pipeFromArray<T,R>(fns: Array<UnaryFunction<T,R>>): UnaryFunction<T,R> {
if (fns.length === 0) {
return identity as UnaryFunction<any,any>;
}
if (fns.length === 1) {
return fns[0];
}
return function piped(input: T): R {
return fns.reduce((prev: any,fn: UnaryFunction<T,R>) => fn(prev),input as any);
};
}
在弄清上面片段中发生的事情之前，重要的是要知道 operators 。运算符是一个函数，它返回另一个函数，该函数的单个参数为Observable<T>，返回类型为Observable<R>。有时，T和R可以相同（例如，使用filter()，debounceTime() ...时）。
例如，map是defined like this：
export function map<T,R>(project: (value: T,index: number) => R,thisArg?: any): OperatorFunction<T,R> {
return operate((source,subscriber) => {
// The index of the value from the source. Used with projection.
let index = 0;
// Subscribe to the source,all errors and completions are sent along
// to the consumer.
source.subscribe(
new OperatorSubscriber(subscriber,(value: T) => {
// Call the projection function with the appropriate this context,// and send the resulting value to the consumer.
subscriber.next(project.call(thisArg,value,index++));
})
);
});
}
export function operate<T,R>(
init: (liftedSource: Observable<T>,subscriber: Subscriber<R>) => (() => void) | void
): OperatorFunction<T,R> {
return (source: Observable<T>) => {
if (hasLift(source)) {
return source.lift(function (this: Subscriber<R>,liftedSource: Observable<T>) {
try {
return init(liftedSource,this);
} catch (err) {
this.error(err);
}
});
}
throw new TypeError('Unable to lift unknown Observable type');
};
}
因此，operate将返回一个功能。注意其参数：source: Observable<T>。返回类型是从Subscriber<R>派生的。
Observable.lift仅创建一个新的Observable。就像在喜欢的列表中创建节点一样。
protected lift<R>(operator?: Operator<T,R>): Observable<R> {
const observable = new Observable<R>();
// it's important to keep track of the source !
observable.source = this;
observable.operator = operator;
return observable;
}
因此，运算符（如map）将返回一个函数。调用该函数的是pipeFromArray函数：
export function pipeFromArray<T,any>;
}
if (fns.length === 1) {
return fns[0];
}
return function piped(input: T): R {
// here the functions returned by the operators are being called
return fns.reduce((prev: any,input as any);
};
}
在以上代码段中，fn函数返回的内容是operate：
return (source: Observable<T>) => {
if (hasLift(source)) { // has `lift` method
return source.lift(function (this: Subscriber<R>,liftedSource: Observable<T>) {
try {
return init(liftedSource,this);
} catch (err) {
this.error(err);
}
});
}
throw new TypeError('Unable to lift unknown Observable type');
};
也许最好还是看一个例子。我建议您自己尝试使用调试器。
const src$ = new Observable(subscriber => {subscriber.next(1),subscriber.complete()});
subscriber => {}回调fn将分配给Observable._subscribe属性。
constructor(subscribe?: (this: Observable<T>,subscriber: Subscriber<T>) => TeardownLogic) {
if (subscribe) {
this._subscribe = subscribe;
}
}
接下来，让我们尝试添加一个运算符：
const src2$ = src$.pipe(map(num => num ** 2))
在这种情况下，它将从pipeFromArray调用此块：
// `pipeFromArray`
if (fns.length === 1) {
return fns[0];
}
// `Observable.pipe`
pipe(...operations: OperatorFunction<any,any>[]): Observable<any> {
return operations.length ? pipeFromArray(operations)(this) : this;
}
因此，Observable.pipe将调用(source: Observable<T>) => { ... }，其中source是src$ Observable。通过调用该函数（其结果存储在src2$中），它还将调用Observable.lift方法。
return source.lift(function (this: Subscriber<R>,liftedSource: Observable<T>) {
try {
return init(liftedSource,this);
} catch (err) {
this.error(err);
}
});
/* ... */
protected lift<R>(operator?: Operator<T,R>): Observable<R> {
const observable = new Observable<R>();
observable.source = this;
observable.operator = operator;
return observable;
}
此时，src$是一个Observable实例，其中source设置为src$，而operator设置为function (this: Subscriber<R>,liftedSource: Observable<T>) ...
从我的角度来看，这完全是关于链接列表的。创建Observable链（通过添加运算符）时，列表是从上到下创建的。
当 tail节点调用其subscribe方法时，将创建另一个列表，这次是从下到上。我喜欢将第一个称为Observable list，将第二个称为Subscribers list。
src2$.subscribe(console.log)
调用subscribe方法时会发生以下情况：
const subscriber = isSubscriber(observerOrNext) ? observerOrNext : new SafeSubscriber(observerOrNext,error,complete);
const { operator,source } = this;
subscriber.add(
operator
? operator.call(subscriber,source)
: source || config.useDeprecatedSynchronousErrorHandling
? this._subscribe(subscriber)
: this._trySubscribe(subscriber)
);
return subscriber;
在这种情况下，src2$有一个operator，因此它将调用它。 operator定义为：
function (this: Subscriber<R>,this);
} catch (err) {
this.error(err);
}
}
其中init取决于所使用的运算符。再一次，这里是map的{{1}}
init export function map<T,R> {
return operate( /* THIS IS `init()` */(source,subscriber) => {
// The index of the value from the source. Used with projection.
let index = 0;
// Subscribe to the source,index++));
})
);
});
}
实际上是source。调用src$时，它将最终调用提供给source.subscribe()的回调。调用new Observable(subscriber => { ... })将从上方调用subscriber.next(1)，后者将调用(value: T) => { ... }（subscriber.next(project.call(thisArg,index++));-提供给project的回调）。最后，map指的是subscriber.next。
回到console.log，这是在调用_subscribe方法时发生的情况：
Subject因此，这就是protected _subscribe(subscriber: Subscriber<T>): Subscription {
this._throwIfClosed(); // if unsubscribed
this._checkFinalizedStatuses(subscriber); // `error` or `complete` notifications
return this._innerSubscribe(subscriber);
}
protected _innerSubscribe(subscriber: Subscriber<any>) {
const { hasError,isStopped,observers } = this;
return hasError || isStopped
? EMPTY_SUBSCRIPTION
: (observers.push(subscriber),new Subscription(() => arrRemove(this.observers,subscriber)));
}
个订户列表的填充方式。通过返回Subject's，它可以确保随后的订阅者取消订阅（由于new Subscription(() => arrRemove(this.observers,subscriber)) / complete通知或仅仅是error），无效订阅者将成为从subscriber.unsubscribe()的列表中删除。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。