CompositeObserver implements StreamObserver<T> { @Override public void onError(Throwable t) { observers.forEach(o -> o.onError(t)); } CompositeObserver(Iterable<StreamObserver<? super T>> observers); @Override void onNext(T value); @Override void onError(Throwable t); @Override void onCompleted();  }
@Test @DisplayName("delegate `onError` calls to wrapped observers") void delegateOnError() { MemoizingObserver<String> observer1 = StreamObservers.memoizingObserver(); MemoizingObserver<String> observer2 = StreamObservers.memoizingObserver(); StreamObserver<String> observer = new CompositeObserver<>(ImmutableList.of(observer1, observer2)); RuntimeException error = new RuntimeException("An error."); observer.onError(error); assertThat(observer1.getError()).isEqualTo(error); assertThat(observer2.getError()).isEqualTo(error); }