added methods in base actor

src/main/java/io/appform/dropwizard/actors/actor/BaseActor.java

@@ -138,6 +138,12 @@ public final void publishWithExpiry(final Message message, final long expiryInMs
         actorImpl.publishWithExpiry(message, expiryInMs);
     }
 
+    public final void publishWithDelayAndExpiry(final Message message,
+                                                final long expiryInMs,
+                                                final long delayMilliseconds) throws Exception {
+        actorImpl.publishWithDelayAndExpiry(message, expiryInMs, delayMilliseconds);
+    }
+
     public final void publish(final Message message) throws Exception {
         val properties = new AMQP.BasicProperties.Builder()
                 .deliveryMode(2)

src/main/java/io/appform/dropwizard/actors/actor/UnmanagedBaseActor.java

@@ -122,6 +122,12 @@ public final void publishWithExpiry(final Message message, final long expiryInMs
         publishActor().publishWithExpiry(message, expiryInMs);
     }
 
+    public final void publishWithDelayAndExpiry(final Message message,
+                                                final long expiryInMs,
+                                                final long delayMilliseconds) throws Exception {
+        publishActor().publishWithDelayAndExpiry(message, expiryInMs, delayMilliseconds);
+    }
+
     public final void publish(final Message message) throws Exception {
         publishActor().publish(message);
     }

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

@@ -148,9 +148,14 @@ private AMQP.BasicProperties getPropertiesWithExpiry(final AMQP.BasicProperties
         if (expiryInMs <= 0) {
             return properties;
         }
+        HashMap<String, Object> enrichedHeaders = new HashMap<>();
+        if (properties.getHeaders() != null) {
+            enrichedHeaders.putAll(properties.getHeaders());
+        }
         val expiresAt = Instant.now().toEpochMilli() + expiryInMs;
-        return new AMQP.BasicProperties.Builder()
-                .headers(ImmutableMap.of(MESSAGE_EXPIRY_TEXT, expiresAt))
+        enrichedHeaders.put(MESSAGE_EXPIRY_TEXT, expiresAt);
+        return properties.builder()
+                .headers(Collections.unmodifiableMap(enrichedHeaders))
                 .build();
     }
 

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

@@ -25,6 +25,7 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
+import org.apache.commons.lang3.StringUtils;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
@@ -342,6 +343,35 @@ private MessageHandlingFunction<Map, Boolean> handleExpectedMessageForReDelivery
         };
     }
 
+    @Test
+    public void testPublishWithExpiryAndDelay() throws Exception {
+        val queueName = "queue-1";
+        val objectMapper = new ObjectMapper();
+        val actorConfig = new ActorConfig();
+        actorConfig.setExchange("test-exchange-delay");
+        val publisher = new UnmanagedPublisher<>(
+                queueName, actorConfig, connection, objectMapper);
+        publisher.start();
+
+        val message = ImmutableMap.of(
+                "key", "test"
+        );
+        publisher.publishWithDelayAndExpiry(message, 1500, 500);
+
+        Thread.sleep(1510);
+
+
+        val expiredDeliveryCount = new AtomicInteger();
+        val consumer = new UnmanagedConsumer<>(
+                queueName, actorConfig, connection, objectMapper, new RetryStrategyFactory(), new ExceptionHandlingFactory(),
+                Map.class, this::handleForNoExpectedMsg, handleExpiredMessageWithDelay(expiredDeliveryCount), (x) -> true);
+        consumer.start();
+
+        Thread.sleep(1000);
+
+        Assertions.assertEquals(1, expiredDeliveryCount.getAndIncrement());
+    }
+
     @NotNull
     private MessageHandlingFunction<Map, Boolean> handleExpiredMessage(AtomicInteger expiredDeliveryCount) {
         return (msg, meta) -> {
@@ -352,6 +382,18 @@ private MessageHandlingFunction<Map, Boolean> handleExpiredMessage(AtomicInteger
         };
     }
 
+    @NotNull
+    private MessageHandlingFunction<Map, Boolean> handleExpiredMessageWithDelay(AtomicInteger expiredDeliveryCount) {
+        return (msg, meta) -> {
+            Assertions.assertTrue(meta.getDelayInMs() > 1500);
+            Map<String, Object> msgHeaders = meta.getHeaders();
+            Assertions.assertNotNull(msgHeaders);
+            Assertions.assertTrue(StringUtils.equals("500", String.valueOf(msgHeaders.get("x-delay"))));
+            expiredDeliveryCount.getAndIncrement();
+            return true;
+        };
+    }
+
     @NotNull
     private MessageHandlingFunction<Map, Boolean> handleDelayedMessageConsumption(AtomicInteger normalDeliveryCount) {
         return (msg, meta) -> {