This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new df3815732 SQS - Add delaySeconds to SQS batch publishing (#760)
df3815732 is described below
commit df3815732158d8b17319951862c8d477e5f7f29c
Author: Luka Lazanja <[email protected]>
AuthorDate: Tue Aug 6 11:44:31 2024 +0200
SQS - Add delaySeconds to SQS batch publishing (#760)
* Add delaySeconds to SQS batch publishing
Add missing delaySeconds property in the mapping from SendMessageRequest
to SendMessageBatchRequestEntry. Update integration tests.
* Apply scalafmt code style
---------
Co-authored-by: Luka Lazanja <[email protected]>
---
.../connectors/sqs/scaladsl/SqsPublishFlow.scala | 1 +
.../test/scala/docs/scaladsl/SqsPublishSpec.scala | 39 +++++++++++++++++++---
2 files changed, 36 insertions(+), 4 deletions(-)
diff --git
a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishFlow.scala
b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishFlow.scala
index 6ad0cfece..196acc10b 100644
---
a/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishFlow.scala
+++
b/sqs/src/main/scala/org/apache/pekko/stream/connectors/sqs/scaladsl/SqsPublishFlow.scala
@@ -93,6 +93,7 @@ object SqsPublishFlow {
.messageAttributes(r.messageAttributes())
.messageGroupId(r.messageGroupId())
.messageDeduplicationId(r.messageDeduplicationId())
+ .delaySeconds(r.delaySeconds())
.build()
}
diff --git a/sqs/src/test/scala/docs/scaladsl/SqsPublishSpec.scala
b/sqs/src/test/scala/docs/scaladsl/SqsPublishSpec.scala
index 00eb842c9..ff718be03 100644
--- a/sqs/src/test/scala/docs/scaladsl/SqsPublishSpec.scala
+++ b/sqs/src/test/scala/docs/scaladsl/SqsPublishSpec.scala
@@ -33,6 +33,7 @@ class SqsPublishSpec extends AnyFlatSpec with Matchers with
DefaultTestContext w
abstract class IntegrationFixture(fifo: Boolean = false) {
val queueUrl: String = if (fifo) randomFifoQueueUrl() else randomQueueUrl()
+ val messageReceiveMinimumTimeout: Int = 2
implicit val awsSqsClient: SqsAsyncClient = sqsClient
def receiveMessage(): Message =
@@ -43,18 +44,20 @@ class SqsPublishSpec extends AnyFlatSpec with Matchers with
DefaultTestContext w
.asScala
.head
- def receiveMessages(maxNumberOfMessages: Int): Seq[Message] = {
+ def receiveMessages(maxNumberOfMessages: Int, waitTimeSeconds: Int = 0):
Seq[Message] = {
// see
https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/sqs/model/ReceiveMessageRequest.html
require(maxNumberOfMessages > 0 && maxNumberOfMessages <= 10,
"maxNumberOfMessages must be in 1 to 10")
+ val timeoutSeconds = messageReceiveMinimumTimeout + waitTimeSeconds
val request =
ReceiveMessageRequest
.builder()
.queueUrl(queueUrl)
.maxNumberOfMessages(maxNumberOfMessages)
+ .waitTimeSeconds(waitTimeSeconds)
.build()
- awsSqsClient.receiveMessage(request).get(2,
TimeUnit.SECONDS).messages().asScala.toSeq
+ awsSqsClient.receiveMessage(request).get(timeoutSeconds,
TimeUnit.SECONDS).messages().asScala.toSeq
}
}
@@ -175,10 +178,16 @@ class SqsPublishSpec extends AnyFlatSpec with Matchers
with DefaultTestContext w
}
}
- it should "publish batch of SendMessageRequests and pull them" taggedAs
Integration in {
+ it should "publish batch of SendMessageRequests without individual delays
and pull them" taggedAs Integration in {
new IntegrationFixture {
// #batch-send-request
- val messages = for (i <- 0 until 10) yield
SendMessageRequest.builder().messageBody(s"Message - $i").build()
+ val messages =
+ for (i <- 0 until 10)
+ yield SendMessageRequest
+ .builder()
+ .messageBody(s"Message - $i")
+ .delaySeconds(0)
+ .build()
val future = Source
.single(messages)
@@ -191,6 +200,28 @@ class SqsPublishSpec extends AnyFlatSpec with Matchers
with DefaultTestContext w
}
}
+ it should "publish batch of SendMessageRequests with individual delays and
pull them" taggedAs Integration in {
+ new IntegrationFixture {
+ val messageDelaySeconds = 5
+ val messages =
+ for (i <- 0 until 10)
+ yield SendMessageRequest
+ .builder()
+ .messageBody(s"Message - $i")
+ .delaySeconds(messageDelaySeconds) // SQS queue should process
messages after 5 seconds
+ .build()
+
+ val future = Source
+ .single(messages)
+ .runWith(SqsPublishSink.batchedMessageSink(queueUrl))
+
+ future.futureValue shouldBe Done
+
+ receiveMessages(10) should have size 0 // delay hasn't elapsed
+ receiveMessages(10, messageDelaySeconds + 5) should have size 10 //
delay has elapsed
+ }
+ }
+
"PublishFlow" should "put message in a flow, then pass the result further"
taggedAs Integration in {
new IntegrationFixture {
val future =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
