softwaremill#124 - Allow for setting the redrive policy through SetQu…

…eueAttributes
simong · Jan 12, 2019 · 68edafa · 68edafa
1 parent 69ac38f
commit 68edafa
Show file tree

Hide file tree

Showing 8 changed files with 62 additions and 10 deletions.
diff --git a/...sticmq/rest/sqs/model/RedrivePolicy.scala → ...n/scala/org/elasticmq/RedrivePolicy.scala b/...sticmq/rest/sqs/model/RedrivePolicy.scala → ...n/scala/org/elasticmq/RedrivePolicy.scala
@@ -1,3 +1,3 @@
-package org.elasticmq.rest.sqs.model
+package org.elasticmq
 
 case class RedrivePolicy(queueName: String, maxReceiveCount: Int)
diff --git a/core/src/main/scala/org/elasticmq/actor/queue/QueueActorQueueOps.scala b/core/src/main/scala/org/elasticmq/actor/queue/QueueActorQueueOps.scala
@@ -19,6 +19,9 @@ trait QueueActorQueueOps extends Logging {
     case UpdateQueueReceiveMessageWait(newReceiveMessageWait) =>
       logger.info(s"${queueData.name}: Updating receive message wait to $newReceiveMessageWait")
       queueData = queueData.copy(receiveMessageWait = newReceiveMessageWait)
+    case UpdateQueueRedrivePolicy(newPolicy) =>
+      val dlqData = DeadLettersQueueData(newPolicy.queueName, newPolicy.maxReceiveCount)
+      queueData = queueData.copy(deadLettersQueue = Some(dlqData))
     case ClearQueue() =>
       messageQueue.clear()
     case GetQueueStatistics(deliveryTime) => getQueueStatistics(deliveryTime)

diff --git a/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala b/core/src/main/scala/org/elasticmq/msg/QueueMsg.scala
@@ -14,6 +14,7 @@ case class UpdateQueueDefaultVisibilityTimeout(newDefaultVisibilityTimeout: Mill
     extends QueueQueueMsg[Unit]
 case class UpdateQueueDelay(newDelay: Duration) extends QueueQueueMsg[Unit]
 case class UpdateQueueReceiveMessageWait(newReceiveMessageWait: Duration) extends QueueQueueMsg[Unit]
+case class UpdateQueueRedrivePolicy(newRedrivePolicy: RedrivePolicy) extends QueueQueueMsg[Unit]
 case class GetQueueStatistics(deliveryTime: Long) extends QueueQueueMsg[QueueStatistics]
 case class ClearQueue() extends QueueQueueMsg[Unit]
 

diff --git a/...esting-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala b/...esting-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
@@ -16,7 +16,6 @@ import scala.util.control.Exception._
 import com.amazonaws.AmazonServiceException
 import org.elasticmq.util.Logging
 import org.elasticmq._
-import org.elasticmq.rest.sqs.model.RedrivePolicy
 
 class AmazonJavaSdkTestSuite extends FunSuite with Matchers with BeforeAndAfter with Logging {
   val visibilityTimeoutAttribute = "VisibilityTimeout"
@@ -1456,7 +1455,7 @@ class AmazonJavaSdkTestSuite extends FunSuite with Matchers with BeforeAndAfter
     }
   }
 
-  test("should validate redrive policy json") {
+  test("should validate the redrive policy json when creating a queue") {
     // Then
     a[AmazonSQSException] shouldBe thrownBy {
       client.createQueue(
@@ -1470,6 +1469,38 @@ class AmazonJavaSdkTestSuite extends FunSuite with Matchers with BeforeAndAfter
     }
   }
 
+  test("should be able to set the redrive policy on an existing queue") {
+    import spray.json._
+    import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
+    // Given two queues
+    client.createQueue(new CreateQueueRequest("dlq1")).getQueueUrl
+    val mainQueueUrl = client.createQueue(new CreateQueueRequest("q1")).getQueueUrl
+
+    // When setting the redrive policy
+    val policyStr = RedrivePolicy("dlq1", 1).toJson.toString()
+    client.setQueueAttributes(
+      new SetQueueAttributesRequest(mainQueueUrl,
+                                    Map(
+                                      redrivePolicyAttribute -> policyStr
+                                    )))
+
+    // The the redrive policy should be returned when getting the queue attributes
+    val queueAttributes = client.getQueueAttributes(mainQueueUrl, List("All")).getAttributes
+    queueAttributes.get(redrivePolicyAttribute) should be(policyStr)
+  }
+
+  test("should validate the redrive policy json when updating the attributes of a queue") {
+    client.createQueue(new CreateQueueRequest("dlq1")).getQueueUrl
+    val mainQueueUrl = client.createQueue(new CreateQueueRequest("q1")).getQueueUrl
+
+    a[AmazonSQSException] shouldBe thrownBy {
+      client.setQueueAttributes(mainQueueUrl, Map(redrivePolicyAttribute -> "not a proper json policy"))
+    }
+    a[AmazonSQSException] shouldBe thrownBy {
+      client.setQueueAttributes(mainQueueUrl, Map(redrivePolicyAttribute -> """{"wrong": "json"}"""))
+    }
+  }
+
   def queueVisibilityTimeout(queueUrl: String) = getQueueLongAttribute(queueUrl, visibilityTimeoutAttribute)
 
   def queueDelay(queueUrl: String) = getQueueLongAttribute(queueUrl, delaySecondsAttribute)

diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/CreateQueueDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/CreateQueueDirectives.scala
@@ -6,13 +6,12 @@ import org.elasticmq.rest.sqs.Constants._
 import org.elasticmq.rest.sqs.CreateQueueDirectives._
 import org.elasticmq.rest.sqs.ParametersUtil._
 import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
-import org.elasticmq.{DeadLettersQueueData, MillisVisibilityTimeout, QueueData}
+import org.elasticmq.{DeadLettersQueueData, MillisVisibilityTimeout, QueueData, RedrivePolicy}
 import org.joda.time.{DateTime, Duration}
 import spray.json._
 import scala.async.Async._
 import scala.concurrent.Future
 
-import org.elasticmq.rest.sqs.model.RedrivePolicy
 import spray.json.JsonParser.ParsingException
 
 trait CreateQueueDirectives {

diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/QueueAttributesDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/QueueAttributesDirectives.scala
@@ -1,19 +1,22 @@
 package org.elasticmq.rest.sqs
 
-import org.elasticmq.MillisVisibilityTimeout
+import org.elasticmq.{MillisVisibilityTimeout, RedrivePolicy}
 import Constants._
 import org.joda.time.Duration
 import org.elasticmq.msg.{
   GetQueueStatistics,
+  LookupQueue,
   UpdateQueueDefaultVisibilityTimeout,
   UpdateQueueDelay,
-  UpdateQueueReceiveMessageWait
+  UpdateQueueReceiveMessageWait,
+  UpdateQueueRedrivePolicy
 }
 import org.elasticmq.actor.reply._
 import scala.concurrent.Future
+import scala.util.Try
 
 import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
-import org.elasticmq.rest.sqs.model.RedrivePolicy
+import org.elasticmq.rest.sqs.model.RedrivePolicyJson
 import spray.json._
 
 trait QueueAttributesDirectives {
@@ -30,7 +33,7 @@ trait QueueAttributesDirectives {
     val RedrivePolicyAttribute = "RedrivePolicy"
 
     val AllUnsupportedAttributeNames = PolicyAttribute :: MaximumMessageSizeAttribute ::
-      MessageRetentionPeriodAttribute :: RedrivePolicyAttribute :: Nil
+      MessageRetentionPeriodAttribute :: Nil
   }
 
   object QueueReadableAttributeNames {
@@ -132,6 +135,20 @@ trait QueueAttributesDirectives {
               case ReceiveMessageWaitTimeSecondsAttribute => {
                 queueActor ? UpdateQueueReceiveMessageWait(Duration.standardSeconds(attributeValue.toLong))
               }
+              case RedrivePolicyParameter =>
+                import RedrivePolicyJson._
+
+                for {
+                  newRedrivePolicy <- Future.fromTry(Try(attributeValue.parseJson.convertTo[RedrivePolicy]))
+                  _ = if (newRedrivePolicy.maxReceiveCount < 1 || newRedrivePolicy.maxReceiveCount > 1000) {
+                    throw SQSException.invalidParameterValue
+                  }
+                  dlq <- queueManagerActor ? LookupQueue(newRedrivePolicy.queueName)
+                  _ = if (dlq.isEmpty) {
+                    throw SQSException.nonExistentQueue
+                  }
+                  _ <- queueActor ? UpdateQueueRedrivePolicy(newRedrivePolicy)
+                } yield ()
               case attr
                   if UnsupportedAttributeNames.AllUnsupportedAttributeNames
                     .contains(attr) => {

diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/model/RedrivePolicyJson.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/model/RedrivePolicyJson.scala
@@ -1,5 +1,6 @@
 package org.elasticmq.rest.sqs.model
 
+import org.elasticmq.RedrivePolicy
 import spray.json.{
   DefaultJsonProtocol,
   JsNumber,

diff --git a/rest/rest-sqs/src/test/scala/org/elasticmq/rest/sqs/RedrivePolicyJsonTest.scala b/rest/rest-sqs/src/test/scala/org/elasticmq/rest/sqs/RedrivePolicyJsonTest.scala
@@ -1,7 +1,7 @@
 package org.elasticmq.rest.sqs
 
 import akka.http.scaladsl.testkit.ScalatestRouteTest
-import org.elasticmq.rest.sqs.model.RedrivePolicy
+import org.elasticmq.RedrivePolicy
 import org.scalatest.{FlatSpec, Matchers}
 import spray.json._