OfframpClientImpl implements OfframpClient<T> { @SuppressWarnings("unchecked") void onNext(Event event) { if (event instanceof Message) { messages.onNext((Message<T>) event); } else if (event instanceof CommitResponse) { commitHandler.complete((CommitResponse) event); } else if (event instanceof Rebalance) { rebalances.onNext((Rebalance) event); } else if (event instanceof Connection) { log.info("Connected to agent {}", ((Connection) event).getAgentName()); } else if (event instanceof Error) { log.error(((Error) event).getReason()); } else { log.warn("Unknown message type: {}", event); } } OfframpClientImpl(OfframpOptions<T> options);   OfframpClientImpl(
      OfframpOptions<T> options,
      WebSocket.Factory socketFactory,
      CommitHandler.Factory commitHandlerFactory); @Override Publisher<Message<T>> messages(); @Override Mono<Boolean> commit(Map<Integer, Long> offsets); @Override Publisher<Set<Integer>> rebalances(); @Override void close();  }
@Test public void onNext_CommitResponse() throws Exception { CommitResponse commitResponse = new CommitResponse("foo", true); underTest.onNext(commitResponse); verify(commitHandler).complete(commitResponse); }
@Test public void onNext_Error() throws Exception { Logger OfframpClientImplLogger = (Logger) LoggerFactory.getLogger(OfframpClientImpl.class); ListAppender<ILoggingEvent> listAppender = new ListAppender<>(); listAppender.start(); OfframpClientImplLogger.addAppender(listAppender); Error error = new Error("That thing was wrong"); underTest.onNext(error); List<ILoggingEvent> logsList = listAppender.list; assertThat(logsList.get(0).getMessage(), is("That thing was wrong")); assertEquals(Level.ERROR, logsList.get(0).getLevel()); }