@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()));
}