Use delay in batch publishing

commercetools · Mar 7, 2024 · 448dcb3 · 448dcb3
1 parent 29bd4bc
commit 448dcb3
Showing 1 changed file with 10 additions and 2 deletions.
diff --git a/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPusher.scala b/...ervice-bus/src/main/scala/com/commercetools/queue/azure/servicebus/ServiceBusPusher.scala
@@ -30,14 +30,22 @@ class ServiceBusPusher[F[_], Data](sender: ServiceBusSenderClient)(implicit seri
 
   override def push(message: Data, delay: Option[FiniteDuration]): F[Unit] = {
     val sbMessage = new ServiceBusMessage(serializer.serialize(message))
-    delay.traverse_(delay =>
+    delay.traverse_ { delay =>
       F.realTimeInstant
-        .map(now => sbMessage.setScheduledEnqueueTime(now.plusMillis(delay.toMillis).atOffset(ZoneOffset.UTC)))) *>
+        .map(now => sbMessage.setScheduledEnqueueTime(now.plusMillis(delay.toMillis).atOffset(ZoneOffset.UTC)))
+    } *>
       F.blocking(sender.sendMessage(sbMessage)).void
   }
 
   override def push(messages: List[Data], delay: Option[FiniteDuration]): F[Unit] = {
     val sbMessages = messages.map(msg => new ServiceBusMessage(serializer.serialize(msg)))
+    delay.traverse_ { delay =>
+      F.realTimeInstant.map { now =>
+        sbMessages.foreach { msg =>
+          msg.setScheduledEnqueueTime(now.plusMillis(delay.toMillis).atOffset(ZoneOffset.UTC))
+        }
+      }
+    }
     F.blocking(sender.sendMessages(sbMessages.asJava)).void
   }