rxgroups/src/main/java/com/airbnb/rxgroups/ObservableGroup.java
<T> ManagedObservable<T> add(final String observerTag, final String observableTag,
      Observable<T> observable, ObservableEmitter<? super T> observer) {
    checkNotDestroyed();
    final Map<String, ManagedObservable<?>> existingObservables =
        getObservablesForObserver(observerTag);
    ManagedObservable<?> previousObservable = existingObservables.get(observableTag);

    if (previousObservable != null) {
      cancelAndRemove(observerTag, observableTag);
    }

    ManagedObservable<T> managedObservable =
        new ManagedObservable<>(observerTag, observableTag, observable, observer, new
            Action() {
              @Override
              public void run() {
                existingObservables.remove(observableTag);
              }
            });

    existingObservables.put(observableTag, managedObservable);

    if (!locked) {
      managedObservable.unlock();
    }
    return managedObservable;
  }
<T> ManagedObservable<T> add(final String observerTag, final String observableTag,
      Observable<T> observable, ObservableEmitter<? super T> observer) {
    checkNotDestroyed();
    final Map<String, ManagedObservable<?>> existingObservables =
        getObservablesForObserver(observerTag);
    ManagedObservable<?> previousObservable = existingObservables.get(observableTag);

    if (previousObservable != null) {
      cancelAndRemove(observerTag, observableTag);
    }

    ManagedObservable<T> managedObservable =
        new ManagedObservable<>(observerTag, observableTag, observable, observer, new
            Action() {
              @Override
              public void run() {
                existingObservables.remove(observableTag);
              }
            });

    existingObservables.put(observableTag, managedObservable);

    if (!locked) {
      managedObservable.unlock();
    }
    return managedObservable;
  }
< T > ManagedObservable < T > add ( final String observerTag , final String observableTag , Observable < T > observable , ObservableEmitter < ? super T > observer ) { checkNotDestroyed ( ) ; final Map < String , ManagedObservable < ? > > existingObservables = getObservablesForObserver ( observerTag ) ; ManagedObservable < ? > previousObservable = existingObservables . get ( observableTag ) ; if ( previousObservable != null ) { cancelAndRemove ( observerTag , observableTag ) ; } ManagedObservable < T > managedObservable = new ManagedObservable <> ( observerTag , observableTag , observable , observer , new Action ( ) { @ Override public void run ( ) { existingObservables . remove ( observableTag ) ; } } ) ; existingObservables . put ( observableTag , managedObservable ) ; if ( ! locked ) { managedObservable . unlock ( ) ; } return managedObservable ; }
Adds an {@link Observable} and {@link Observer} to this group and subscribes to it. If an
{@link Observable} with the same tag is already added, the previous one will be canceled and
removed before adding and subscribing to the new one.
Adds an {