@Test public void rebalances() throws Exception { Set<Integer> assignment = singleton(0); Rebalance rebalance = new Rebalance(assignment); Disposable subscribe = Flux.from(underTest.rebalances()).limitRequest(1).subscribe(); underTest.getRebalances().onNext(rebalance); subscribe.dispose(); }
@Override public Publisher<Set<Integer>> rebalances() { return rebalances.map(Rebalance::getAssignment); }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Set<Integer>> rebalances() { return rebalances.map(Rebalance::getAssignment); } }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Set<Integer>> rebalances() { return rebalances.map(Rebalance::getAssignment); } OfframpClientImpl(OfframpOptions<T> options);   OfframpClientImpl(
      OfframpOptions<T> options,
      WebSocket.Factory socketFactory,
      CommitHandler.Factory commitHandlerFactory); }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Set<Integer>> rebalances() { return rebalances.map(Rebalance::getAssignment); } 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> { @Override public Publisher<Set<Integer>> rebalances() { return rebalances.map(Rebalance::getAssignment); } 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();  }