Merge pull request santanusinha#54 from pvntohare/publish_bug_fix

Adding infinite expiry for publishWithExpiry(), NPE bug fix for publish()

Author: r0goyal

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.appform.dropwizard.actors</groupId>
     <artifactId>dropwizard-rabbitmq-actors</artifactId>
-    <version>2.0.28-4</version>
+    <version>2.0.28-5</version>
     <name>Dropwizard RabbitMQ Bundle</name>
     <url>https://github.com/santanusinha/dropwizard-rabbitmq-actors</url>
     <description>Provides actor abstraction on RabbitMQ for dropwizard based projects.</description>

src/main/java/io/appform/dropwizard/actors/base/UnmanagedPublisher.java

@@ -67,11 +67,15 @@ public final void publishWithDelay(final Message message, final long delayMillis
     }
 
     public final void publishWithExpiry(final Message message, final long expiryInMs) throws Exception {
-        val expiresAt = Instant.now().toEpochMilli() + expiryInMs;
-        val properties = new AMQP.BasicProperties.Builder()
+        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                 .deliveryMode(2)
-                .headers(ImmutableMap.of(MESSAGE_EXPIRY_TEXT, expiresAt))
                 .build();
+        if (expiryInMs > 0) {
+            val expiresAt = Instant.now().toEpochMilli() + expiryInMs;
+            properties = properties.builder()
+                    .headers(ImmutableMap.of(MESSAGE_EXPIRY_TEXT, expiresAt))
+                    .build();
+        }
         publish(message, properties);
     }
 
@@ -91,7 +95,10 @@ public final void publish(final Message message, final AMQP.BasicProperties prop
     }
 
     private AMQP.BasicProperties getEnrichedProperties(AMQP.BasicProperties properties) {
-        val enrichedHeaders = new HashMap<>(properties.getHeaders());
+        HashMap<String, Object> enrichedHeaders = new HashMap<>();
+        if (properties.getHeaders() != null) {
+            enrichedHeaders.putAll(properties.getHeaders());
+        }
         enrichedHeaders.put(MESSAGE_PUBLISHED_TEXT, Instant.now().toEpochMilli());
         return properties.builder()
                 .headers(Collections.unmodifiableMap(enrichedHeaders))

src/test/java/io/appform/dropwizard/actors/connectivity/actor/ExpiryMessagesTest.java

@@ -275,6 +275,66 @@ queueName, actorConfig, connection, objectMapper, new RetryStrategyFactory(), ne
         Assertions.assertEquals(1, normalDeliveryCount.get());
     }
 
+    /**
+     * This test does the following:
+     * - Publisher publishes the message with an expiry of -1ms with 1 consumer
+     * - Consumer would consume the message normally and delay in consumption is 1000ms
+     */
+    @Test
+    public void testWhenMessagesAreNotExpiredCase4() throws Exception {
+        val queueName = "queue-6";
+        val objectMapper = new ObjectMapper();
+        val actorConfig = new ActorConfig();
+        actorConfig.setExchange("test-exchange-1");
+        val publisher = new UnmanagedPublisher<>(
+                queueName, actorConfig, connection, objectMapper);
+        publisher.start();
+
+        val message = ImmutableMap.of(
+                "key", "value"
+        );
+
+        publisher.publishWithExpiry(message, -1);
+        Thread.sleep(1000);
+        val normalDeliveryCount = new AtomicInteger();
+        val consumer = new UnmanagedConsumer<>(
+                queueName, actorConfig, connection, objectMapper, new RetryStrategyFactory(), new ExceptionHandlingFactory(),
+                Map.class, handleExpectedMessageWithDelay(1000, normalDeliveryCount), this::handleForNoExpectedMsg, (x) -> true);
+        consumer.start();
+
+
+        Assertions.assertEquals(1, normalDeliveryCount.get());
+    }
+
+    /**
+     * This test does the following:
+     * - Publisher publishes the message with 1 consumer
+     * - Consumer would consume the message normally and delay in consumption is 1000ms
+     */
+    @Test
+    public void testRegressionForPublish() throws Exception {
+        val queueName = "queue-7";
+        val objectMapper = new ObjectMapper();
+        val actorConfig = new ActorConfig();
+        actorConfig.setExchange("test-exchange-1");
+        val publisher = new UnmanagedPublisher<>(
+                queueName, actorConfig, connection, objectMapper);
+        publisher.start();
+
+        val message = ImmutableMap.of(
+                "key", "value"
+        );
+
+        publisher.publish(message);
+        val normalDeliveryCount = new AtomicInteger();
+        val consumer = new UnmanagedConsumer<>(
+                queueName, actorConfig, connection, objectMapper, new RetryStrategyFactory(), new ExceptionHandlingFactory(),
+                Map.class, handleExpectedMessageWithDelay(0, normalDeliveryCount), this::handleForNoExpectedMsg, (x) -> true);
+        consumer.start();
+        Thread.sleep(500);
+
+        Assertions.assertEquals(1, normalDeliveryCount.get());
+    }
     @NotNull
     private MessageHandlingFunction<Map, Boolean> handleExpectedMessage(int maxDelay, AtomicInteger normalDeliveryCount) {
         return (msg, meta) -> {
@@ -284,6 +344,15 @@ private MessageHandlingFunction<Map, Boolean> handleExpectedMessage(int maxDelay
         };
     }
 
+    @NotNull
+    private MessageHandlingFunction<Map, Boolean> handleExpectedMessageWithDelay(int minDelay, AtomicInteger normalDeliveryCount) {
+        return (msg, meta) -> {
+            Assertions.assertTrue(meta.getDelayInMs() > minDelay);
+            normalDeliveryCount.getAndIncrement();
+            return true;
+        };
+    }
+
     @NotNull
     private MessageHandlingFunction<Map, Boolean> handleExpectedMessageForReDelivery(AtomicInteger expiredDeliveryCount) {
         return (msg, meta) -> {