3

简单的说

给定一个现有的 Observable（尚未完成），有没有办法检索关联的订阅者（传递给 subscribe 的函数）以使他们订阅另一个 Observable？

语境

我的应用程序中的一项服务有助于创建SeverEvent连接，将ConnectableObservable返回到代理连接并允许使用发布运算符进行多播。该服务通过内部存储跟踪现有连接：

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
}

在创建连接时，如果关联的跟踪器已经存在（使用连接的端点生成身份），服务应该：

  • ok关闭现有跟踪器的ServerEvent连接
  • ok打开一个新的SerevrEvent连接（因此是一个新的 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;
}

谢谢你。

4

2 回答 2

2

与其尝试手动将订阅者从一个 Observable 转移到另一个 Observable，不如为侦听器提供一个 Observable，它会在需要时自动切换到不同的 Observable。

您可以通过使用始终切换到最新的内部 Observable 的高阶 Observable（发出 Observables 的 Observable）来做到这一点。

基本概念

// a BehaviorSubject is used so that late subscribers also immediately get the most recent inner Observable
const higherOrderObservable = new BehaviorSubject<Observable<any>>(EMPTY);

// pass new Observable to listeners
higherOrderObservable.next(new Observable(..));

// get most recent inner Observable
const currentObservable = higherOrderObservable.pipe(switchMap(obs => obs));
currentObservable.subscribe(valueFromInnerObservable => { .. })

在你的情况下

为每个端点创建一个BehaviorSubject（跟踪器供应商），它发出当前应该用于该端点的 Observable（跟踪器） 。当应该为给定端点使用不同的跟踪器时，将这个新的 Observable 传递给. 让您的听众订阅（跟踪器供应商）自动为他们提供正确的跟踪器，即切换到当前应该使用的 Observable。BehaviorSubjectBehaviorSubject

您的代码的简化版本可能如下所示。具体情况取决于您在createTracker整个应用程序中使用该功能的方式。

interface ConnectionTracker {
  fullUri: string;
  tracker$: ConnectableObservable<any>;
}

// Map an endpoint to a tracker supplier.
// This is your higher order Observable as it emits objects that wrap an Observable
store: { [key: string]: BehaviorSubject<ConnectionTracker> };
closeAllTrackers$ = new Subject();

// Creates a new tracker if necessary and returns a ConnectedObservable for that tracker. 
// The ConnectedObservable will always resemble the current tracker.
createTracker<T>(endpoint: string, queryString: string = null): Observable<any> {
  const fullUri = endpoint + (queryString ? `?${queryString}` : '');
  // if no tracker supplier for the endpoint exists, create one
  if (!store[endpoint]) {
    store[endpoint] = new BehaviorSubject<ConnectionTracker>(null);
  }
  const currentTracker = store[endpoint].getValue();

  // if no tracker exists or the current one is obsolete, create a new one
  if (!currentTracker || currentTracker.fullUri !== fullUri) {
    const tracker$ = new Observable<T>(subscriber => {
      const source = new EventSource(fullUri, { withCredentials: true });
      source.onmessage = e => subscriber.next(JSON.parse(e.data));
      source.onerror = e => subscriber.error(e);
      return () => source.close(); // on unsubscribe close the source
    }).pipe(publish()) as ConnectableObservable<any>;
    tracker$.connect();
    // pass the new tracker to the tracker supplier
    store[endpoint].next({ fullUri, tracker$ });
  }
  // return the tracker supplier for the given endpoint that always switches to the current tracker
  return store[endpoint].pipe(
    switchMap(tracker => tracker ? tracker.tracker$ : EMPTY), // switchMap will unsubscribe from the previous tracker and thus close the connection if a new tracker comes in
    takeUntil(this.closeAllTrackers$) // complete the tracker supplier on emit
  );
}

// close all trackers and remove the tracker suppliers
closeAllTrackers() {
  this.closeAllTrackers$.next();
  this.store = {};
}

如果您想立即关闭所有跟踪器连接并且现有订阅者应该收到complete通知，请致电closeAllTrackers。如果您只想关闭一些跟踪器连接但不希望现有订阅者收到complete通知，以便他们继续侦听将来提供的新跟踪器，请调用store[trackerEndpoint].next(null)每个跟踪器。

于 2019-07-25T17:33:18.220 回答
1

如果你试图做一些事情，比如将订阅者移动到不同的 observable，那么你只是没有做 RxJS 中预期的事情。任何此类操纵基本上都是黑客行为。

如果您偶尔产生一个新的 observable（例如通过发出请求）并且您希望某些订阅者始终订阅其中最新的，那么解决方案如下：

  private observables: Subject<Observable<Data>> = new Subject();

  getData(): Observable<Data> {
    return this.observables.pipe(switchAll());
  }

  onMakingNewRequest(newObservable: Observable<Data>) {
    this.observables.push(newObservable);
  }

通过这种方式，您可以公开客户端订阅的单个可观察对象（通过getData()），但通过推送给this.observables您更改用户看到的实际数据源。

至于关闭连接和类似的东西，你的 observable（每个请求或其他东西创建的那个）基本上应该在取消订阅时负责释放和关闭东西，然后你不需要做任何额外的处理，前一个 observable从您推送新的那一刻起，将自动取消订阅。详细信息取决于您联系的实际后端。

于 2019-07-26T10:46:31.110 回答