@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<>(); 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) { } @Override public void onComplete() { } }); 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())); }
PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; }
JmsSource { PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; } }
JmsSource { PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; } JmsSource(JMSContext context, JmsConnectorIncomingConfiguration config, Jsonb json, Executor executor); }
JmsSource { PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; } JmsSource(JMSContext context, JmsConnectorIncomingConfiguration config, Jsonb json, Executor executor);  }
JmsSource { PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; } JmsSource(JMSContext context, JmsConnectorIncomingConfiguration config, Jsonb json, Executor executor);   }