Receiver implements Subscriber<Message<TestMessage>>, AutoCloseable { public void start() throws InterruptedException { log.debug("Starting receiver"); CountDownLatch latch = new CountDownLatch(1); Flux.from(offramp.rebalances()).any(not(Set::isEmpty)).subscribe(b -> latch.countDown()); offramp.messages().subscribe(this); Commits.fromMessages(commitProcessor, Duration.ofSeconds(1)).flatMap(offramp::commit).subscribe(); if (!latch.await(30, SECONDS)) { throw new RuntimeException("Timed out waiting for a rebalance event to start the receiver"); } log.info("Receiver started"); } Receiver(
      @NonNull MessagesMetricSet metrics,
      @NonNull TestMessageContextManager contextManager,
      @NonNull @Value("${originName}") String origin,
      @NonNull OfframpClient<TestMessage> offramp); void start(); @Override void close(); @Override void onSubscribe(Subscription s); @Override void onNext(Message<TestMessage> message); @Override void onError(Throwable t); @Override void onComplete();  }
@Test public void worker_passes_to_context() throws Exception { try (Receiver receiver = new Receiver(metrics, contextManager, origin, offramp)) { receiver.start(); verify(context1).messageReceived(message1); verify(context2).messageReceived(message2); } }