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();  }
@Test public void should_create_nsq_consumer_on_ha_proxy_register_action_and_forward_nsq_consumers_messages_to_subscriber() throws Exception { @SuppressWarnings("unchecked") FlowableNSQConsumer<Integer> nsqConsumer1 = mock(FlowableNSQConsumer.class); when(nsqConsumer1.flowable()).thenReturn(Flowable.range(0, 10)); when(nsqConsumerBuilder.apply("haProxy1")).thenReturn(nsqConsumer1); @SuppressWarnings("unchecked") FlowableNSQConsumer<Integer> nsqConsumer2 = mock(FlowableNSQConsumer.class); when(nsqConsumer2.flowable()).thenReturn(Flowable.range(10, 10)); when(nsqConsumerBuilder.apply("haProxy2")).thenReturn(nsqConsumer2); actionToCommitEventConsumer.onNext(HAProxyAction.register("haProxy1")); actionToCommitEventConsumer.onNext(HAProxyAction.register("haProxy2")); subscriber.hasSubscription(); subscriber.assertSubscribed(); subscriber.assertValueSet(IntStream.range(0, 20).mapToObj(Integer::valueOf).collect(Collectors.toSet())); }
@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(); }