@Test public void onNext_CommitResponse() throws Exception { CommitResponse commitResponse = new CommitResponse("foo", true); underTest.onNext(commitResponse); verify(commitHandler).complete(commitResponse); }
@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 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 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); }
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(); }
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();  }