JmsSink { SubscriberBuilder<Message<?>, Void> getSink() { return sink; } JmsSink(JMSContext context, JmsConnectorOutgoingConfiguration config, Jsonb jsonb, Executor executor);   }
@Test public void testDefaultConfiguration() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); 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(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); }
@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"); }
@Test public void testWithDeliveryDelayAndMode() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("delivery-delay", 1500L) .put("delivery-mode", "non_persistent") .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); subscriber = sink.getSink().build(); subscriber.onSubscribe(new Subscriptions.EmptySubscription()); AtomicBoolean acked = new AtomicBoolean(); subscriber.onNext(Message.of("hello", () -> CompletableFuture.runAsync(() -> acked.set(true)))); await() .atLeast(1, TimeUnit.SECONDS) .until(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client.messages.get(0).getJMSMessageID()).isNotNull(); assertThat(client.messages.get(0).getJMSTimestamp()).isPositive(); }
@Test public void testDisablingMessageIdAndTimestamp() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("disable-message-id", true) .put("disable-message-timestamp", true) .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); 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(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client.messages.get(0).getJMSMessageID()).isNull(); assertThat(client.messages.get(0).getJMSTimestamp()).isEqualTo(0L); }
@Test public void testWithCorrelationIdAndPriorityAndTTL() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("correlation-id", "my-correlation") .put("priority", 5) .put("ttl", 1000L) .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); 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(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client.messages.get(0).getJMSPriority()).isEqualTo(5); assertThat(client.messages.get(0).getJMSCorrelationID()).isEqualTo("my-correlation"); }
@Test public void testWithReplyTo() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("reply-to", "my-response") .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); 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(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client.messages.get(0).getJMSReplyTo()) .isInstanceOf(Queue.class); }
@Test public void testWithReplyToTopic() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("reply-to", "my-response") .put("reply-to-destination-type", "topic") .put("channel-name", "jms") .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); 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(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); assertThat(client.messages.get(0).getBody(String.class)).isEqualTo("hello"); assertThat(client.messages.get(0).getJMSReplyTo()) .isInstanceOf(Topic.class); }
@Test public void testPropagation() throws JMSException { MapBasedConfig config = new MapBasedConfig.Builder() .put("destination", "queue-one") .put("channel-name", "jms") .put("ttl", 10000L) .build(); JmsSink sink = new JmsSink(jms, new JmsConnectorOutgoingConfiguration(config), json, executor); MyJmsClient client = new MyJmsClient(jms.createQueue("queue-one")); subscriber = sink.getSink().build(); subscriber.onSubscribe(new Subscriptions.EmptySubscription()); AtomicBoolean acked = new AtomicBoolean(); Message<String> hello = Message.of("hello", () -> CompletableFuture.runAsync(() -> acked.set(true))); Destination rt = jms.createQueue("reply-to"); OutgoingJmsMessageMetadata metadata = OutgoingJmsMessageMetadata.builder() .withCorrelationId("my-correlation-id") .withReplyTo(rt) .withDeliveryMode(DeliveryMode.PERSISTENT) .withType(String.class.getName()) .build(); hello = hello.withMetadata(Collections.singleton(metadata)); subscriber.onNext(hello); await().until(() -> client.messages.size() >= 1); assertThat(acked).isTrue(); javax.jms.Message message = client.messages.get(0); assertThat(message.getBody(String.class)).isEqualTo("hello"); assertThat(message.getJMSCorrelationID()).isEqualTo("my-correlation-id"); assertThat(message.getJMSReplyTo()).isEqualTo(rt); assertThat(message.getJMSDeliveryMode()).isEqualTo(2); assertThat(message.getJMSType()).isEqualTo(String.class.getName()); }