SingleThreadInMemSubscriber extends InMemorySubscriber implements Logging { @SuppressWarnings("FutureReturnValueIgnored") @Override public void onMessage(ExternalMessage message) { CompletableFuture.runAsync(() -> callObservers(message), executor) .exceptionally(throwable -> logError(throwable, message)); } SingleThreadInMemSubscriber(ChannelId id); @SuppressWarnings("FutureReturnValueIgnored") // Error handling is done manually. @Override void onMessage(ExternalMessage message); @Override void close();  }
@Test @MuteLogging @DisplayName("not halt after observer throws an error") void recoverFromObserverError() throws InterruptedException { SingleThreadInMemSubscriber subscriber = new SingleThreadInMemSubscriber(ChannelId.getDefaultInstance()); ThrowingObserver throwingObserver = new ThrowingObserver(); subscriber.addObserver(throwingObserver); Any id = Identifier.pack(newUuid()); ExternalMessage externalMessage = ExternalMessage .newBuilder() .setId(id) .build(); subscriber.onMessage(externalMessage); waitForMessage(); assertThat(throwingObserver.onNextCalled()) .isTrue(); subscriber.removeObserver(throwingObserver); MemoizingObserver<ExternalMessage> memoizingObserver = new MemoizingObserver<>(); subscriber.addObserver(memoizingObserver); subscriber.onMessage(externalMessage); waitForMessage(); assertThat(memoizingObserver.firstResponse()) .isEqualTo(externalMessage); }