@Test public void should_shutdown_nsq_consumer_on_unregister_action() throws Exception { @SuppressWarnings("unchecked") FlowableNSQConsumer<Integer> nsqConsumer = mock(FlowableNSQConsumer.class); when(nsqConsumer.flowable()).thenReturn(Flowable.range(0, 10)); when(nsqConsumerBuilder.apply("haProxy1")).thenReturn(nsqConsumer); actionToCommitEventConsumer.onNext(HAProxyAction.register("haProxy1")); subscriber.hasSubscription(); subscriber.assertSubscribed(); subscriber.assertValueSet(IntStream.range(0, 20).mapToObj(Integer::valueOf).collect(Collectors.toSet())); actionToCommitEventConsumer.onNext(HAProxyAction.unregister("haProxy1")); verify(nsqConsumer).shutdown(); }
@Override public void onNext(HAProxyPublisher.HAProxyAction action) { if (action.isRegistration()) { createNSQConsumer(action.getId()); } else { deleteNSQConsumer(action.getId()); } }
HAProxySubscriber extends DefaultSubscriber<HAProxyPublisher.HAProxyAction> implements Managed { @Override public void onNext(HAProxyPublisher.HAProxyAction action) { if (action.isRegistration()) { createNSQConsumer(action.getId()); } else { deleteNSQConsumer(action.getId()); } } }
HAProxySubscriber extends DefaultSubscriber<HAProxyPublisher.HAProxyAction> implements Managed { @Override public void onNext(HAProxyPublisher.HAProxyAction action) { if (action.isRegistration()) { createNSQConsumer(action.getId()); } else { deleteNSQConsumer(action.getId()); } } HAProxySubscriber(Function<String, FlowableNSQConsumer<U>> nsqConsumerBuilder, Subscriber<U> subscriber); }
HAProxySubscriber extends DefaultSubscriber<HAProxyPublisher.HAProxyAction> implements Managed { @Override public void onNext(HAProxyPublisher.HAProxyAction action) { if (action.isRegistration()) { createNSQConsumer(action.getId()); } else { deleteNSQConsumer(action.getId()); } } HAProxySubscriber(Function<String, FlowableNSQConsumer<U>> nsqConsumerBuilder, Subscriber<U> subscriber); @Override void onComplete(); @Override void onError(Throwable e); @Override void onNext(HAProxyPublisher.HAProxyAction action); @Override void start(); @Override void stop(); }
HAProxySubscriber extends DefaultSubscriber<HAProxyPublisher.HAProxyAction> implements Managed { @Override public void onNext(HAProxyPublisher.HAProxyAction action) { if (action.isRegistration()) { createNSQConsumer(action.getId()); } else { deleteNSQConsumer(action.getId()); } } HAProxySubscriber(Function<String, FlowableNSQConsumer<U>> nsqConsumerBuilder, Subscriber<U> subscriber); @Override void onComplete(); @Override void onError(Throwable e); @Override void onNext(HAProxyPublisher.HAProxyAction action); @Override void start(); @Override void stop();  }