Repository: softwaremill/elasticmq (scala)

Issue:
Add Missing Actions
`ListDeadLetterSourceQueues Action` https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ListDeadLetterSourceQueues.html

`RemovePermission Action` https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_RemovePermission.html

Pull request:
Adds empty shell for `RemovePermission` action and fully functional (sans pagination) `ListDeadLetterSourceQueues`. Implementation of `ListDeadLetterSourceQueues` is rough because it requires loading `QueueData` for every existing queue. However I'm not aware of any better way to do this inside the elasticmq framework.

Branched off of open commit to use JSON responses, so this can be merged afterwards. Review last two commits for new changes. Tested and working locally.

Open to all feedback as this is my first contribution to the project. NBD if not wanted, mainly did this for my own exploration.

Closes #818

Patch:
diff --git a/core/src/main/scala/org/elasticmq/actor/QueueManagerActor.scala b/core/src/main/scala/org/elasticmq/actor/QueueManagerActor.scala
index 6c742208f..baf3a3543 100644
--- a/core/src/main/scala/org/elasticmq/actor/QueueManagerActor.scala
+++ b/core/src/main/scala/org/elasticmq/actor/QueueManagerActor.scala
@@ -58,6 +58,11 @@ class QueueManagerActor(nowProvider: NowProvider, limits: Limits, queueEventList
         result
 
       case ListQueues() => queues.keySet.toSeq
+
+      case ListDeadLetterSourceQueues(queueName) =>
+        queues.collect {
+          case (name, actor) if actor.queueData.deadLettersQueue.exists(_.name == queueName) => name
+        }.toList
     }
 
   protected def createQueueActor(
diff --git a/core/src/main/scala/org/elasticmq/msg/QueueManagerMsg.scala b/core/src/main/scala/org/elasticmq/msg/QueueManagerMsg.scala
index 15d3b1e43..85b62dd0c 100644
--- a/core/src/main/scala/org/elasticmq/msg/QueueManagerMsg.scala
+++ b/core/src/main/scala/org/elasticmq/msg/QueueManagerMsg.scala
@@ -10,3 +10,4 @@ case class CreateQueue(request: CreateQueueData) extends QueueManagerMsg[Either[
 case class DeleteQueue(queueName: String) extends QueueManagerMsg[Unit]
 case class LookupQueue(queueName: String) extends QueueManagerMsg[Option[ActorRef]]
 case class ListQueues() extends QueueManagerMsg[Seq[String]]
+case class ListDeadLetterSourceQueues(queueName: String) extends QueueManagerMsg[List[String]]
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/Action.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/Action.scala
index 3b6462b58..c2570c360 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/Action.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/Action.scala
@@ -9,12 +9,14 @@ object Action extends Enumeration {
   val DeleteMessage = Value("DeleteMessage")
   val DeleteQueue = Value("DeleteQueue")
   val GetQueueUrl = Value("GetQueueUrl")
+  val ListDeadLetterSourceQueues = Value("ListDeadLetterSourceQueues")
   val ListQueueTags = Value("ListQueueTags")
   val ListQueues = Value("ListQueues")
   val PurgeQueue = Value("PurgeQueue")
   val GetQueueAttributes = Value("GetQueueAttributes")
   val SetQueueAttributes = Value("SetQueueAttributes")
   val ReceiveMessage = Value("ReceiveMessage")
+  val RemovePermission = Value("RemovePermission")
   val SendMessageBatch = Value("SendMessageBatch")
   val SendMessage = Value("SendMessage")
   val TagQueue = Value("TagQueue")
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ListDeadLetterSourceQueuesDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ListDeadLetterSourceQueuesDirectives.scala
new file mode 100644
index 000000000..d10b87529
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/ListDeadLetterSourceQueuesDirectives.scala
@@ -0,0 +1,69 @@
+package org.elasticmq.rest.sqs
+
+import org.elasticmq.actor.reply._
+import org.elasticmq.msg.ListDeadLetterSourceQueues
+import org.elasticmq.rest.sqs.Action.{ListDeadLetterSourceQueues => ListDeadLetterSourceQueuesAction}
+import org.elasticmq.rest.sqs.Constants._
+import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
+import org.elasticmq.rest.sqs.model.RequestPayload
+import spray.json.DefaultJsonProtocol._
+import spray.json.RootJsonFormat
+
+import scala.xml.Elem
+
+trait ListDeadLetterSourceQueuesDirectives { this: ElasticMQDirectives with QueueURLModule with ResponseMarshaller =>
+  def listDeadLetterSourceQueues(p: RequestPayload)(implicit marshallerDependencies: MarshallerDependencies) = {
+    p.action(ListDeadLetterSourceQueuesAction) {
+      val payload = p.as[ListDeadLetterSourceQueuesActionRequest]
+      (getQueueNameFromQueueUrl(payload.QueueUrl) & baseQueueURL)
+        .tmap { case (deadLetterQueueName, baseURL) =>
+          for {
+            queueNames <- queueManagerActor ? ListDeadLetterSourceQueues(deadLetterQueueName)
+            queueUrls = queueNames.map(queueName => s"$baseURL/$queueName")
+          } yield {
+            complete(ListDeadLetterSourceQueuesResponse(queueUrls))
+          }
+        }
+        .tapply(f => f._1)
+    }
+  }
+}
+
+case class ListDeadLetterSourceQueuesActionRequest(
+    MaxResults: Option[Int],
+    NextToken: Option[String],
+    QueueUrl: String
+)
+object ListDeadLetterSourceQueuesActionRequest {
+  implicit val requestJsonFormat: RootJsonFormat[ListDeadLetterSourceQueuesActionRequest] = jsonFormat3(ListDeadLetterSourceQueuesActionRequest.apply)
+
+  implicit val requestParamReader: FlatParamsReader[ListDeadLetterSourceQueuesActionRequest] =
+    new FlatParamsReader[ListDeadLetterSourceQueuesActionRequest] {
+      override def read(params: Map[String, String]): ListDeadLetterSourceQueuesActionRequest = {
+        new ListDeadLetterSourceQueuesActionRequest(
+          params.get(MaxResultsParameter).map(_.toInt),
+          params.get(NextTokenParameter),
+          requiredParameter(params)(QueueUrlParameter)
+        )
+      }
+    }
+}
+
+case class ListDeadLetterSourceQueuesResponse(queueUrls: List[String])
+object ListDeadLetterSourceQueuesResponse {
+  implicit val format: RootJsonFormat[ListDeadLetterSourceQueuesResponse] = jsonFormat1(ListDeadLetterSourceQueuesResponse.apply)
+
+  implicit val xmlSerializer: XmlSerializer[ListDeadLetterSourceQueuesResponse] = new XmlSerializer[ListDeadLetterSourceQueuesResponse] {
+    override def toXml(t: ListDeadLetterSourceQueuesResponse): Elem = {
+
+      <ListDeadLetterSourceQueuesResponse>
+        <ListDeadLetterSourceQueuesResult>
+          {t.queueUrls.map(queueUrl => <QueueUrl>{queueUrl}</QueueUrl>)}
+        </ListDeadLetterSourceQueuesResult>
+        <ResponseMetadata>
+          <RequestId>{EmptyRequestId}</RequestId>
+        </ResponseMetadata>
+      </ListDeadLetterSourceQueuesResponse>
+    }
+  }
+}
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/RemovePermissionDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/RemovePermissionDirectives.scala
new file mode 100644
index 000000000..8959a5e29
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/RemovePermissionDirectives.scala
@@ -0,0 +1,13 @@
+package org.elasticmq.rest.sqs
+
+import org.elasticmq.rest.sqs.Action.RemovePermission
+import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
+import org.elasticmq.rest.sqs.model.RequestPayload
+
+trait RemovePermissionDirectives { this: ElasticMQDirectives with QueueURLModule with ResponseMarshaller =>
+  def removePermission(p: RequestPayload)(implicit marshallerDependencies: MarshallerDependencies) = {
+    p.action(RemovePermission) {
+      emptyResponse("RemovePermissionResponse")
+    }
+  }
+}
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SQSRestServerBuilder.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SQSRestServerBuilder.scala
index eaed1c5ac..231890fa5 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SQSRestServerBuilder.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/SQSRestServerBuilder.scala
@@ -158,12 +158,14 @@ case class TheSQSRestServerBuilder(
       with GetQueueUrlDirectives
       with PurgeQueueDirectives
       with AddPermissionDirectives
+      with RemovePermissionDirectives
       with AttributesModule
       with TagQueueDirectives
       with TagsModule
       with UnmatchedActionRoutes
       with ResponseMarshaller
-      with QueueAttributesOps {
+      with QueueAttributesOps
+      with ListDeadLetterSourceQueuesDirectives {
 
       def serverAddress = currentServerAddress.get()
 
@@ -199,9 +201,11 @@ case class TheSQSRestServerBuilder(
         getQueueAttributes(p) ~
         setQueueAttributes(p) ~
         addPermission(p) ~
+        removePermission(p) ~
         tagQueue(p) ~
         untagQueue(p) ~
         listQueueTags(p) ~
+        listDeadLetterSourceQueues(p) ~
         //4. Unmatched action
         unmatchedAction(p)
 
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/directives/QueueDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/directives/QueueDirectives.scala
index 8cfebf780..d88c11231 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/directives/QueueDirectives.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/directives/QueueDirectives.scala
@@ -36,7 +36,7 @@ trait QueueDirectives {
     getQueueNameFromQueueUrl(queueUrl)(queueName => queueActor(queueName, qa => queueData(qa, qd => body(qa, qd))))
   }
 
-  private def getQueueNameFromQueueUrl(queueUrl: String): Directive1[String] = {
+  protected def getQueueNameFromQueueUrl(queueUrl: String): Directive1[String] = {
 
     val matcher =
       if (contextPath.nonEmpty) {


Test patch:
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonCliTestSuite.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonCliTestSuite.scala
index 779773a20..9222e96da 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonCliTestSuite.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonCliTestSuite.scala
@@ -1,6 +1,8 @@
 package org.elasticmq.rest.sqs
 
 import org.elasticmq.StringMessageAttribute
+import org.elasticmq.rest.sqs.model.RedrivePolicyJson.format
+import org.elasticmq.rest.sqs.model.RedrivePolicy
 import org.scalatest.{Inside, LoneElement, OptionValues, Tag}
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.prop.TableDrivenPropertyChecks
@@ -27,9 +29,10 @@ class AmazonCliTestSuite
     AWSCli(Option(System.getenv("AWS_CLI_V2_EXECUTABLE")).getOrElse("aws"), "aws version 2")
   )
 
-  def createQueue(name: String)(implicit cli: AWSCli): CreateQueueResponse = {
+  def createQueue(name: String, attributesJson: Option[String] = None)(implicit cli: AWSCli): CreateQueueResponse = {
+    val attributesOption = attributesJson.fold("")(json => s"--attributes='$json'")
     val result =
-      s"""${cli.executable} sqs create-queue --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-name=$name""" !!
+      s"""${cli.executable} sqs create-queue --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-name=$name $attributesOption""" !!
 
     result.parseJson.convertTo[CreateQueueResponse]
   }
@@ -291,6 +294,15 @@ class AmazonCliTestSuite
       s"""${version.executable} sqs add-permission --label l --aws-account-ids=$awsAccountId --actions=get --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-url=$url""" !!
     }
 
+    test(s"should remove permission ${version.name}", Only213) {
+      // given
+      val url = createQueue("permission-test").QueueUrl
+      s"""${version.executable} sqs add-permission --label l --aws-account-ids=$awsAccountId --actions=get --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-url=$url""" !
+
+      // when ~> then
+      s"""${version.executable} sqs remove-permission --label l --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-url=$url""" !!
+    }
+
     test(s"should purge queue ${version.name}") {
       // given
       val url = createQueue("permission-test").QueueUrl
@@ -509,6 +521,21 @@ class AmazonCliTestSuite
 
       result.parseJson.convertTo[GetQueueAttributesResponse].Attributes.get("VisibilityTimeout") shouldBe Some("99")
     }
+
+    test(s"should get source queues of a dlq with ${version.name}", Only213) {
+      // given
+      val dlq = createQueue("testDlq")
+      val redrivePolicy = RedrivePolicy("testDlq", awsRegion, awsAccountId, 1).toJson.toString.replaceAll("\"", "\\\\\"")
+      val main = createQueue("main", Some(s"""{"RedrivePolicy":"$redrivePolicy"}"""))
+
+      // when
+      val response =
+        s"""${version.executable} sqs list-dead-letter-source-queues --endpoint=$ServiceEndpoint --region=us-west-1 --no-sign-request --queue-url ${dlq.QueueUrl}""" !!
+
+      // then
+      val result = response.parseJson.convertTo[ListDeadLetterSourceQueuesResponse]
+      result.queueUrls should contain only(main.QueueUrl)
+    }
   }
 }
 
diff --git a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
index d28b6265c..252e5fd8b 100644
--- a/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
+++ b/rest/rest-sqs-testing-amazon-java-sdk/src/test/scala/org/elasticmq/rest/sqs/AmazonJavaSdkTestSuite.scala
@@ -2095,6 +2095,30 @@ class AmazonJavaSdkTestSuite extends SqsClientServerCommunication with Matchers
     secondReceiveResult.getMessages shouldBe empty
   }
 
+  test("should list all source queues for a dlq") {
+    // given
+    val dlqUrl = client.createQueue(new CreateQueueRequest("testDlq")).getQueueUrl
+    val redrivePolicy = RedrivePolicy("testDlq", awsRegion, awsAccountId, 1).toJson.toString()
+
+    val qUrls = for (i <- 1 to 3) yield {
+      client
+        .createQueue(
+          new CreateQueueRequest("q" + i)
+            .withAttributes(
+              Map(redrivePolicyAttribute -> redrivePolicy).asJava
+            )
+        )
+        .getQueueUrl
+    }
+
+    // when
+    val request = new ListDeadLetterSourceQueuesRequest(dlqUrl)
+    val result = client.listDeadLetterSourceQueues(request).getQueueUrls.asScala
+
+    // then
+    result should contain theSameElementsAs(qUrls)
+  }
+
   def queueDelay(queueUrl: String): Long = getQueueLongAttribute(queueUrl, delaySecondsAttribute)
 
   def getQueueLongAttribute(queueUrl: String, attributeName: String): Long = {