Repeated identical subscription use case not handled

[ijager]

I am using the meteor-rxjs API in an Angular Service like in the snippet below. The function returns an Observable holding all results for a certain query. Since the query may be different every time this function is called, we need to do the subscription every time to make sure the correct documents are synced with the client.

  getSensors(someQuery: Object): Observable<Sensor[]> {
    return Observable.create(observer => {
      MeteorObservable.subscribe('sensors', someQuery).subscribe(() => {
        MeteorObservable.autorun().zone().subscribe(() => {
          observer.next(SensorCollection.find().fetch());
        });
      });
    });
  }

While the query can be different, it can also be identical to a previous query. If the previous subscription is still live, Meteor will detect that the subscription already exists and will not execute the onReady() callback. As a result, the observable returned by MeteorObservable.subscribe()will never emit an event and the component requesting the data will not get it.

When an identical subscription is already live, Meteor will still return a subscription handler. This handler contains the subscriptionId which is the same as the id of the existing subscription. However MeteorObservable does not use this information yet. If it did MeteorObservable could recognize a duplicate subscriptionId and issue the observer.next() event anyway.

I implemented a solution here. This works by storing an array of live subscriptions globally and checking whether the new meteor-subscription already exists. If so, a next()-event is emitted on the observer.

let liveSubscriptions = [];

export class MeteorObservable {

public static subscribe<T>(name: string, ...args: any[]): Observable<T> {

    .... 
    // some lines removed to save spave
    ....

    let subHandler = null;
    return Observable.create((observer: Subscriber<Meteor.Error | T>) => {
      observers.push(observer);
      // Execute subscribe lazily.
      if (subHandler === null) {
        subHandler = subscribe();
        if (liveSubscriptions.find(sub => sub === subHandler.subscriptionId)) {
          // subscription already exists, call observer.next() since Meteor won't.
          observer.next();
        } else {
          liveSubscriptions.push(subHandler.subscriptionId);
        }
      }
      return () => {
        removeObserver(observers,
          observer, () => {
            // remove subscription from liveSubscriptions list
            let i = liveSubscriptions.findIndex(
              sub => sub === subHandler.subscriptionId
            );

            if (i > -1) {
              liveSubscriptions.splice(i, 1);
            }

            subHandler.stop();

          });
      };
    });
  }

}

I tested this, and it seems to work fine.

The only things is, Is there a better solution for the global array that now lives in this file. Maybe it would be nice to have a local context of live subscriptions. For example per Angular Service.

[dotansimha]

Fixed in 0.4.8