forked from akka/alpakka
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add attribute parameters to sqs source settings akka#302
- Loading branch information
1 parent
2d0ec27
commit 62faa51
Showing
5 changed files
with
120 additions
and
26 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
76 changes: 76 additions & 0 deletions
76
sqs/src/main/scala/akka/stream/alpakka/sqs/SqsSourceSettings.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package akka.stream.alpakka.sqs | ||
|
|
||
| import java.util | ||
|
|
||
| import scala.collection.JavaConverters._ | ||
|
|
||
| object SqsSourceSettings { | ||
| val Defaults = SqsSourceSettings(20, 100, 10) | ||
|
|
||
| def create(waitTimeSeconds: Int, maxBufferSize: Int, maxBatchSize: Int): SqsSourceSettings = | ||
| SqsSourceSettings(waitTimeSeconds, maxBufferSize, maxBatchSize) | ||
|
|
||
| def create(waitTimeSeconds: Int, | ||
| maxBufferSize: Int, | ||
| maxBatchSize: Int, | ||
| attributeNames: util.List[AttributeName], | ||
| messageAttributeNames: util.List[MessageAttributeName]): SqsSourceSettings = | ||
| SqsSourceSettings(waitTimeSeconds, | ||
| maxBufferSize, | ||
| maxBatchSize, | ||
| attributeNames.asScala, | ||
| messageAttributeNames.asScala) | ||
|
|
||
| } | ||
|
|
||
| //#SqsSourceSettings | ||
| final case class SqsSourceSettings( | ||
| waitTimeSeconds: Int, | ||
| maxBufferSize: Int, | ||
| maxBatchSize: Int, | ||
| attributeNames: Seq[AttributeName] = Seq(), | ||
| messageAttributeNames: Seq[MessageAttributeName] = Seq() | ||
| ) { | ||
| require(maxBatchSize <= maxBufferSize, "maxBatchSize must be lower or equal than maxBufferSize") | ||
| // SQS requirements | ||
| require(0 <= waitTimeSeconds && waitTimeSeconds <= 20, | ||
| s"Invalid value ($waitTimeSeconds) for waitTimeSeconds. Requirement: 0 <= waitTimeSeconds <= 20 ") | ||
| require(1 <= maxBatchSize && maxBatchSize <= 10, | ||
| s"Invalid value ($maxBatchSize) for maxBatchSize. Requirement: 1 <= maxBatchSize <= 10 ") | ||
| } | ||
| //#SqsSourceSettings | ||
|
|
||
| final case class MessageAttributeName(name: String) { | ||
| require( | ||
| name.matches("[0-9a-zA-Z_\\-.]+"), | ||
| "MessageAttributeNames may only contain alphanumeric characters and the underscore (_), hyphen (-), and period (.)" | ||
| ) | ||
|
|
||
| require( | ||
| !name.matches("(^\\.[^*].*)|(.*\\.\\..*)|(.*\\.$)"), | ||
| "MessageAttributeNames cannot start or end with a period (.) or have multiple periods in succession (..)" | ||
| ) | ||
|
|
||
| require(name.length <= 256, "MessageAttributeNames may not be longer than 256 characters") | ||
| } | ||
|
|
||
| sealed abstract class AttributeName(val name: String) | ||
| case object All extends AttributeName("All") | ||
| case object Policy extends AttributeName("Policy") | ||
| case object VisibilityTimeout extends AttributeName("VisibilityTimeout") | ||
| case object MaximumMessageSize extends AttributeName("MaximumMessageSize") | ||
| case object MessageRetentionPeriod extends AttributeName("MessageRetentionPeriod") | ||
| case object ApproximateNumberOfMessages extends AttributeName("ApproximateNumberOfMessages") | ||
| case object ApproximateNumberOfMessagesNotVisible extends AttributeName("ApproximateNumberOfMessagesNotVisible") | ||
| case object CreatedTimestamp extends AttributeName("CreatedTimestamp") | ||
| case object LastModifiedTimestamp extends AttributeName("LastModifiedTimestamp") | ||
| case object QueueArn extends AttributeName("QueueArn") | ||
| case object ApproximateNumberOfMessagesDelayed extends AttributeName("ApproximateNumberOfMessagesDelayed") | ||
| case object DelaySeconds extends AttributeName("DelaySeconds") | ||
| case object ReceiveMessageWaitTimeSeconds extends AttributeName("ReceiveMessageWaitTimeSeconds") | ||
| case object RedrivePolicy extends AttributeName("RedrivePolicy") | ||
| case object FifoQueue extends AttributeName("FifoQueue") | ||
| case object ContentBasedDeduplication extends AttributeName("ContentBasedDeduplication") | ||
| case object KmsMasterKeyId extends AttributeName("KmsMasterKeyId") | ||
| case object KmsDataKeyReusePeriodSeconds extends AttributeName("KmsDataKeyReusePeriodSeconds") | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
39 changes: 39 additions & 0 deletions
39
sqs/src/test/scala/akka/stream/alpakka/sqs/scaladsl/MessageAttributeNameSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /* | ||
| * Copyright (C) 2016-2017 Lightbend Inc. <http://www.lightbend.com> | ||
| */ | ||
| package akka.stream.alpakka.sqs.scaladsl | ||
|
|
||
| import akka.stream.alpakka.sqs.{MessageAttributeName, SqsSourceSettings} | ||
| import org.scalatest.{FlatSpec, Matchers} | ||
|
|
||
| class MessageAttributeNameSpec extends FlatSpec with Matchers { | ||
|
|
||
| it should "not allow names which have periods at the beginning" in { | ||
| a[IllegalArgumentException] should be thrownBy { | ||
| MessageAttributeName(".failed") | ||
| } | ||
| } | ||
|
|
||
| it should "not allow names which have periods at the end" in { | ||
| a[IllegalArgumentException] should be thrownBy { | ||
| MessageAttributeName("failed.") | ||
| } | ||
|
|
||
| } | ||
|
|
||
| it should "reject names which are longer than 256 characters" in { | ||
| a[IllegalArgumentException] should be thrownBy { | ||
| MessageAttributeName( | ||
| "A.really.realy.long.attribute.name.that.is.longer.than.what.is.allowed.256.characters.are.allowed." + | ||
| "however.they.cannot.contain.anything.other.than.alphanumerics.hypens.underscores.and.periods.though" + | ||
| "you.cant.have.more.than.one.consecutive.period.they.are.also.case.sensitive" | ||
| ) | ||
| } | ||
| } | ||
| it should "reject names with multiple sequential periods" in { | ||
| a[IllegalArgumentException] should be thrownBy { | ||
| MessageAttributeName("multiple..periods") | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters