@Test
public void testMultipleRequests() {

    JmsSource source = new JmsSource(jms,
            new JmsConnectorIncomingConfiguration(new MapBasedConfig.Builder().put("channel-name", "queue").build()),
            null, null);
    Publisher<IncomingJmsMessage<?>> publisher = source.getSource().buildRs();
    new Thread(() -> {
        JMSContext context = factory.createContext();
        JMSProducer producer = context.createProducer();
        Queue q = context.createQueue("queue");
        for (int i = 0; i < 50; i++) {
            producer.send(q, i);
        }
    }).start();
    List<IncomingJmsMessage<?>> list = new CopyOnWriteArrayList<>();
    AtomicReference<Subscription> upstream = new AtomicReference<>();
    //noinspection SubscriberImplementation
    publisher.subscribe(new Subscriber<IncomingJmsMessage<?>>() {
        @Override
        public void onSubscribe(Subscription s) {
            upstream.set(s);
        }
        @Override
        public void onNext(IncomingJmsMessage<?> incomingJmsMessage) {
            list.add(incomingJmsMessage);
        }
        @Override
        public void onError(Throwable t) {
            // ignored
        }
        @Override
        public void onComplete() {
            // ignored
        }
    });
    await().untilAtomic(upstream, is(notNullValue()));
    upstream.get().request(10);
    await().until(() -> list.size() == 10);
    upstream.get().request(4);
    await().until(() -> list.size() == 14);
    upstream.get().request(Long.MAX_VALUE);
    await().until(() -> list.size() == 50);
    assertThat(list.stream().map(r -> (Integer) r.getPayload()).collect(Collectors.toList()))
            .containsAll(IntStream.of(49).boxed().collect(Collectors.toList()));
}