简单来说
给定一个尚未完成的现有Observable,是否有一种方法可以检索相关的订阅者(传递给subscribe的函数)以使它们订阅另一个Observable?
背景
我的应用程序中的服务帮助创建SeverEvent连接,返回一个ConnectableObservable以代理连接并使用publish运算符允许多播。该服务通过内部存储跟踪现有的连接:
store: {[key: string]: ConnectionTracker};
// …
interface ConnectionTracker {
url: string;
eventSource: EventSource;
observable: rx.ConnectableObservable<any>;
subscription: rx.Subscription;
observer: rx.Observer<any>;
data?: any; // Arbitrary data
}
在连接创建时,如果已经存在相关联的跟踪器(使用连接端点进行身份验证),则该服务应该:
- 关闭现有跟踪器的 ServerEvent 连接。
- 打开新的 ServerEvent 连接(因此是一个新的 ConnectableObservable)。
- 用新的 Observable 替换现有的跟踪器 Observable,但现有的订阅者现在应该订阅新的 Observable 而不是旧的。
下面是创建 ConnectionTracker 的代码部分:
/**
* Create/Update a ServerEvent connection tracker
*/
createTracker<T>(endpoint: string, queryString: string = null): ConnectionTracker
{
let fullUri = endpoint + (queryString ? `?${queryString}` : '')
, tracker = this.findTrackerByEndpoint(endpoint) || {
observable: null,
fullUri: fullUri,
eventSource: null,
observer: null,
subscription: null
}
;
// Tracker exists
if (tracker.observable !== null) {
// If fullUri hasn't changed, use the tracker as is
if (tracker.fullUri === fullUri) {
return tracker;
}
// At this point, we know "fullUri" has changed, the tracker's
// connection should be replaced with a fresh one
// ⇒ TODO
// ⇒ Gather old tracker.observable's subscribers/subscriptions to make
// them subscribe to the new Observable instead (created down below)
// Terminate previous connection and clean related resouces
tracker.observer.complete();
tracker.eventSource.close();
}
tracker.eventSource = new EventSource(<any>fullUri, {withCredentials: true});
tracker.observable = rx.Observable.create((observer: rx.Observer<T>) => {
// Executed once
tracker.eventSource.onmessage = e => observer.next(JSON.parse(e.data));
tracker.eventSource.onerror = e => observer.error(e);
// Keep track of the observer
tracker.observer = observer;
})
// Transform Observable into a ConnectableObservable for multicast
.publish()
;
// Start emitting right away and also keep a reference to
// proxy subscription for later disposal
tracker.subscription = tracker.observable.connect();
return tracker;
}
谢谢你。
switchMap
来返回“新”的 Observable。订阅者将保持不变,但从这个“新”的 Observable 接收值。 - martin