JmsSource { PublisherBuilder<IncomingJmsMessage<?>> getSource() { return source; } JmsSource(JMSContext context, JmsConnectorIncomingConfiguration config, Jsonb json, Executor executor);   }
@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())); }
@Test public void testBroadcast() { JmsSource source = new JmsSource(jms, new JmsConnectorIncomingConfiguration(new MapBasedConfig.Builder() .put("channel-name", "queue").put("broadcast", true).build()), null, null); PublisherBuilder<IncomingJmsMessage<?>> publisher = source.getSource(); List<IncomingJmsMessage<?>> list1 = new ArrayList<>(); List<IncomingJmsMessage<?>> list2 = new ArrayList<>(); publisher.peek(list1::add).ignore().run(); 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(); publisher.peek(list2::add).ignore().run(); await().until(() -> list1.size() == 50); await().until(() -> list2.size() == 50); assertThat(list1.stream().map(r -> (Integer) r.getPayload()).collect(Collectors.toList())) .containsAll(IntStream.of(49).boxed().collect(Collectors.toList())); assertThat(list2.stream().map(r -> (Integer) r.getPayload()).collect(Collectors.toList())) .containsAll(IntStream.of(49).boxed().collect(Collectors.toList())); }