ObservableGroup { public <T> ObservableTransformer<? super T, T> transform(Observer<? super T> observer, String observableTag) { return new GroupSubscriptionTransformer<>(this, Utils.getObserverTag(observer), observableTag); } ObservableGroup(long groupId); long id(); void initializeAutoTaggingAndResubscription(final Object target); void initializeAutoTaggingAndResubscriptionInTargetClassOnly(T target,
      Class<T> targetClass); ObservableTransformer<? super T, T> transform(Observer<? super T> observer,
      String observableTag); ObservableTransformer<? super T, T> transform(Observer<? super T> observer); void lock(); void unlock(); void dispose(); Observable<T> observable(Observer<? super T> observer); Observable<T> observable(Observer<? super T> observer, String observableTag); SourceSubscription subscription(Observer<?> observer, String observableTag); SourceSubscription subscription(Observer<?> observer); void resubscribeAll(TaggedObserver<? super T> observer); void resubscribe(TaggedObserver<? super T> observer, String observableTag); void resubscribe(TaggedObserver<? super T> observer); void cancelAndRemove(Observer<?> observer, String observableTag); void cancelAllObservablesForObserver(Observer<?> observer); boolean hasObservable(Observer<?> observer, String observableTag); boolean hasObservables(Observer<?> observer); boolean isDestroyed(); @Override String toString();  }
@Test public void shouldDeliverSuccessfulEvent() throws Exception { ObservableGroup group = observableManager.newGroup(); PublishSubject<String> sourceObservable = PublishSubject.create(); TestObserver<String> subscriber = new TestObserver<>(); sourceObservable.compose(group.transform(subscriber)).subscribe(subscriber); subscriber.assertNotComplete(); sourceObservable.onNext("Foo Bar"); sourceObservable.onComplete(); subscriber.assertComplete(); subscriber.assertValue("Foo Bar"); }
@Test public void shouldDeliverError() throws Exception { ObservableGroup group = observableManager.newGroup(); TestObserver<String> testObserver = new TestObserver<>(); Observable<String> sourceObservable = Observable.error(new RuntimeException("boom")); sourceObservable.compose(group.transform(testObserver)).subscribe(testObserver); testObserver.assertError(RuntimeException.class); }
@Test public void shouldOverrideExistingSubscriber() throws Exception { ObservableGroup group = observableManager.newGroup(); PublishSubject<String> sourceObservable = PublishSubject.create(); TestAutoResubscribingObserver testSubscriber1 = new TestAutoResubscribingObserver("foo"); TestAutoResubscribingObserver testSubscriber2 = new TestAutoResubscribingObserver("foo"); sourceObservable.compose(group.transform(testSubscriber1)).subscribe(testSubscriber1); sourceObservable.compose(group.transform(testSubscriber2)).subscribe(testSubscriber2); sourceObservable.onNext("Ruben Aguirre"); sourceObservable.onComplete(); testSubscriber1.assertionTarget.assertNotComplete(); testSubscriber1.assertionTarget.assertNoValues(); testSubscriber2.assertionTarget.assertComplete(); testSubscriber2.assertionTarget.assertValue("Ruben Aguirre"); }
@Test public void shouldNotBeCompleted() { ObservableGroup group = observableManager.newGroup(); TestObserver<Object> subscriber = new TestObserver<>(); Observable<String> sourceObservable = Observable.never(); sourceObservable.compose(group.transform(fooObserver)).subscribe(fooObserver); subscriber.assertNotComplete(); }