@Test public void testDefaultConfigurationAgainstTopic() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "my-topic") .put("destination-type", "topic") .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client1 = new MyJmsClient(jms.createTopic("my-topic")); MyJmsClient client2 = new MyJmsClient(jms.createTopic("my-topic")); subscriber = sink.getSink().build(); subscriber.onSubscribe(new Subscriptions.EmptySubscription()); AtomicBoolean acked = new AtomicBoolean(); subscriber.onNext(Message.of("hello", () -> CompletableFuture.runAsync(() -> acked.set(true)))); await().until(() -> client1.messages.size() >= 1); await().until(() -> client2.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client1.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client2.messages.get(0).getBody(String.class)).isEqualTo("hello"); }
SubscriberBuilder<Message<?>, Void> getSink() { return sink; }
JmsSink { SubscriberBuilder<Message<?>, Void> getSink() { return sink; } }
JmsSink { SubscriberBuilder<Message<?>, Void> getSink() { return sink; } JmsSink(JMSContext context, JmsConnectorOutgoingConfiguration config, Jsonb jsonb, Executor executor); }
JmsSink { SubscriberBuilder<Message<?>, Void> getSink() { return sink; } JmsSink(JMSContext context, JmsConnectorOutgoingConfiguration config, Jsonb jsonb, Executor executor);  }
JmsSink { SubscriberBuilder<Message<?>, Void> getSink() { return sink; } JmsSink(JMSContext context, JmsConnectorOutgoingConfiguration config, Jsonb jsonb, Executor executor);   }