@Test public void messages() throws Exception { doReturn(1).when(options).getInitialRequestAmount(); doReturn(1).when(options).getReplenishingRequestAmount(); Disposable subscribe = Flux.from(underTest.messages()).subscribe(); subscribe.dispose(); InOrder inOrder = Mockito.inOrder(socket); inOrder.verify(socket).send(new Request(1)); inOrder.verify(socket).send(new Cancel()); }
@Override public Publisher<Message<T>> messages() { Flux<Message<T>> flux = Flux.from(messages); if (options.getInitialRequestAmount() != Integer.MAX_VALUE) { flux = flux.limitRate(options.getInitialRequestAmount(), options.getReplenishingRequestAmount()); } return flux.retryWhen(retryStrategy()); }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Message<T>> messages() { Flux<Message<T>> flux = Flux.from(messages); if (options.getInitialRequestAmount() != Integer.MAX_VALUE) { flux = flux.limitRate(options.getInitialRequestAmount(), options.getReplenishingRequestAmount()); } return flux.retryWhen(retryStrategy()); } }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Message<T>> messages() { Flux<Message<T>> flux = Flux.from(messages); if (options.getInitialRequestAmount() != Integer.MAX_VALUE) { flux = flux.limitRate(options.getInitialRequestAmount(), options.getReplenishingRequestAmount()); } return flux.retryWhen(retryStrategy()); } OfframpClientImpl(OfframpOptions<T> options);   OfframpClientImpl(
      OfframpOptions<T> options,
      WebSocket.Factory socketFactory,
      CommitHandler.Factory commitHandlerFactory); }
OfframpClientImpl implements OfframpClient<T> { @Override public Publisher<Message<T>> messages() { Flux<Message<T>> flux = Flux.from(messages); if (options.getInitialRequestAmount() != Integer.MAX_VALUE) { flux = flux.limitRate(options.getInitialRequestAmount(), options.getReplenishingRequestAmount()); } return flux.retryWhen(retryStrategy()); } 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<Message<T>> messages() { Flux<Message<T>> flux = Flux.from(messages); if (options.getInitialRequestAmount() != Integer.MAX_VALUE) { flux = flux.limitRate(options.getInitialRequestAmount(), options.getReplenishingRequestAmount()); } return flux.retryWhen(retryStrategy()); } 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();  }