    @Test
    public void should_create_nsq_consumer_on_ha_proxy_register_action_and_forward_nsq_consumers_messages_to_subscriber() throws Exception {

        // given
        @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);

        // when
        actionToCommitEventConsumer.onNext(HAProxyAction.register("haProxy1"));
        actionToCommitEventConsumer.onNext(HAProxyAction.register("haProxy2"));

        // then
        subscriber.hasSubscription();
        subscriber.assertSubscribed();
        subscriber.assertValueSet(IntStream.range(0, 20).mapToObj(Integer::valueOf).collect(Collectors.toSet()));
    }