Repository: softwaremill/elasticmq
Pull request #424 (2020-11-16 17:15:58)

REST server for queues statistics 
It is an expansion of statistics gathered internally. On the one hand, they are served as mbeans (on branch ui-mbeans). On the second hand, these commits expose REST endpoints with queues statistics and their attributes.




Patch:
diff --git a/build.sbt b/build.sbt
index 758d9cc3..389bb797 100644
--- a/build.sbt
+++ b/build.sbt
@@ -88,6 +88,7 @@ val akka2Streams = "com.typesafe.akka" %% "akka-stream" % akkaVersion
 val akka2Testkit = "com.typesafe.akka" %% "akka-testkit" % akkaVersion % "test"
 val akka2Http = "com.typesafe.akka" %% "akka-http" % akkaHttpVersion
 val sprayJson = "io.spray" %% "spray-json" % "1.3.5"
+val akkaHttpSprayJson = "com.typesafe.akka" %% "akka-http-spray-json" % akkaHttpVersion
 val akka2HttpTestkit = "com.typesafe.akka" %% "akka-http-testkit" % akkaHttpVersion % "test"
 
 val scalaAsync = "org.scala-lang.modules" %% "scala-async" % "0.10.0"
@@ -135,11 +136,12 @@ lazy val restSqs: Project = (project in file("rest/rest-sqs"))
                                 akka2Http,
                                 akka2Streams,
                                 sprayJson,
+                                akkaHttpSprayJson,
                                 akka2Testkit,
                                 akka2HttpTestkit,
                                 scalaAsync) ++ common
   ))
-  .dependsOn(core, commonTest % "test")
+  .dependsOn(core % "compile->compile;test->test", commonTest % "test")
 
 lazy val restSqsTestingAmazonJavaSdk: Project =
   (project in file("rest/rest-sqs-testing-amazon-java-sdk"))
@@ -195,7 +197,7 @@ lazy val server: Project = (project in file("server"))
      */
     credentials += Credentials(Path.userHome / ".s3_elasticmq_credentials"),
     // docker
-    dockerExposedPorts := Seq(9324),
+    dockerExposedPorts := Seq(9324,9325),
     dockerBaseImage := "openjdk:8u212-b04-jdk-stretch",
     packageName in Docker := "elasticmq",
     dockerUsername := Some("softwaremill"),
@@ -249,7 +251,7 @@ lazy val nativeServer: Project = (project in file("native-server"))
     ),
     dockerEntrypoint := Seq("/sbin/tini", "--", "/opt/docker/bin/elasticmq-native-server", "-Dconfig.file=/opt/elasticmq.conf"),
     dockerUpdateLatest := true,
-    dockerExposedPorts := Seq(9324),
+    dockerExposedPorts := Seq(9324,9325),
     dockerCommands := {
       val commands = dockerCommands.value
       val index = commands.indexWhere {
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/AttributesModule.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/AttributesModule.scala
index eaa5e380..09ca3821 100644
--- a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/AttributesModule.scala
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/AttributesModule.scala
@@ -10,9 +10,11 @@ trait AttributesModule {
   val attributeNameAndValuesReader = new AttributeNameAndValuesReader
 
   class AttributeNamesReader {
+    private val attributeNameKey = "AttributeName"
+
     def read(parameters: Map[String, String], allAttributeNames: List[String]) = {
       def collect(suffix: Int, acc: List[String]): List[String] = {
-        parameters.get("AttributeName." + suffix) match {
+        parameters.get(makeAttributeMapKeyAtPosition(suffix)) match {
           case None     => acc
           case Some(an) => collect(suffix + 1, an :: acc)
         }
@@ -26,11 +28,21 @@ trait AttributesModule {
         }
       }
 
-      val rawAttributeNames = collect(1, parameters.get("AttributeName").toList)
+      val rawAttributeNames = collect(1, parameters.get(attributeNameKey).toList)
       val attributeNames = unfoldAllAttributeIfRequested(rawAttributeNames)
 
       attributeNames
     }
+
+    def prepareParametersForRead(attributesToRead: List[String]): Map[String, String] = {
+      attributesToRead.zipWithIndex
+        .map { case (item, index) => (makeAttributeMapKeyAtPosition(index + 1), item) }
+        .toMap
+    }
+
+    private def makeAttributeMapKeyAtPosition(position: Int): String = {
+      s"$attributeNameKey.${position}"
+    }
   }
 
   class AttributesToXmlConverter {
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
index f477d223..2809ceb8 100644
--- 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,133 +1,20 @@
 package org.elasticmq.rest.sqs
 
 import akka.http.scaladsl.server.Route
-import org.elasticmq.actor.reply._
-import org.elasticmq.msg._
 import org.elasticmq.rest.sqs.Action.{GetQueueAttributes, SetQueueAttributes}
 import org.elasticmq.rest.sqs.Constants._
 import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
-import org.elasticmq.rest.sqs.model.RedrivePolicy.BackwardCompatibleRedrivePolicy
-import org.elasticmq.rest.sqs.model.RedrivePolicy
-import org.elasticmq.{DeadLettersQueueData, MillisVisibilityTimeout}
-import org.joda.time.Duration
-import spray.json.JsonParser.ParsingException
-import spray.json._
 
-import scala.async.Async.{await, _}
 import scala.concurrent.Future
 
 trait QueueAttributesDirectives {
-  this: ElasticMQDirectives with AttributesModule =>
-
-  def awsRegion: String
-  def awsAccountId: String
-
-  val attributeValuesCalculator = new AttributeValuesCalculator
-
-  object QueueWriteableAttributeNames {
-    val AllWriteableAttributeNames: List[String] = VisibilityTimeoutParameter :: DelaySecondsAttribute ::
-      ReceiveMessageWaitTimeSecondsAttribute :: RedrivePolicyParameter :: Nil
-  }
-
-  object UnsupportedAttributeNames {
-    val PolicyAttribute = "Policy"
-    val MaximumMessageSizeAttribute = "MaximumMessageSize"
-    val MessageRetentionPeriodAttribute = "MessageRetentionPeriod"
-    val FifoQueueAttribute = "FifoQueue"
-
-    val AllUnsupportedAttributeNames: List[String] = PolicyAttribute :: MaximumMessageSizeAttribute ::
-      MessageRetentionPeriodAttribute :: FifoQueueAttribute :: Nil
-  }
-
-  object FifoAttributeNames {
-    val ContentBasedDeduplication = "ContentBasedDeduplication"
-    val FifoQueue = "FifoQueue"
-
-    val AllFifoAttributeNames: Seq[String] = Seq(
-      ContentBasedDeduplication,
-      FifoQueue
-    )
-  }
-
-  object QueueReadableAttributeNames {
-    val ApproximateNumberOfMessagesAttribute = "ApproximateNumberOfMessages"
-    val ApproximateNumberOfMessagesNotVisibleAttribute =
-      "ApproximateNumberOfMessagesNotVisible"
-    val ApproximateNumberOfMessagesDelayedAttribute =
-      "ApproximateNumberOfMessagesDelayed"
-    val CreatedTimestampAttribute = "CreatedTimestamp"
-    val LastModifiedTimestampAttribute = "LastModifiedTimestamp"
-
-    val AllAttributeNames: List[String] = QueueWriteableAttributeNames.AllWriteableAttributeNames ++
-      (ApproximateNumberOfMessagesAttribute ::
-        ApproximateNumberOfMessagesNotVisibleAttribute ::
-        ApproximateNumberOfMessagesDelayedAttribute ::
-        CreatedTimestampAttribute ::
-        LastModifiedTimestampAttribute ::
-        QueueArnAttribute :: Nil) ++ FifoAttributeNames.AllFifoAttributeNames
-  }
+  this: ElasticMQDirectives with QueueAttributesOps =>
 
   def getQueueAttributes(p: AnyParams): Route = {
     p.action(GetQueueAttributes) {
       queueActorAndDataFromRequest(p) { (queueActor, queueData) =>
-        import QueueReadableAttributeNames._
-        import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
-
-        def calculateAttributeValues(attributeNames: List[String]): List[(String, Future[String])] = {
-          lazy val stats = queueActor ? GetQueueStatistics(System.currentTimeMillis())
-
-          import AttributeValuesCalculator.Rule
-
-          val alwaysAvailableParameterRules = Seq(
-            Rule(
-              VisibilityTimeoutParameter,
-              () => Future.successful(queueData.defaultVisibilityTimeout.seconds.toString)
-            ),
-            Rule(DelaySecondsAttribute, () => Future.successful(queueData.delay.getStandardSeconds.toString)),
-            Rule(ApproximateNumberOfMessagesAttribute, () => stats.map(_.approximateNumberOfVisibleMessages.toString)),
-            Rule(
-              ApproximateNumberOfMessagesNotVisibleAttribute,
-              () => stats.map(_.approximateNumberOfInvisibleMessages.toString)
-            ),
-            Rule(
-              ApproximateNumberOfMessagesDelayedAttribute,
-              () => stats.map(_.approximateNumberOfMessagesDelayed.toString)
-            ),
-            Rule(CreatedTimestampAttribute, () => Future.successful((queueData.created.getMillis / 1000L).toString)),
-            Rule(
-              LastModifiedTimestampAttribute,
-              () => Future.successful((queueData.lastModified.getMillis / 1000L).toString)
-            ),
-            Rule(
-              ReceiveMessageWaitTimeSecondsAttribute,
-              () => Future.successful(queueData.receiveMessageWait.getStandardSeconds.toString)
-            ),
-            Rule(QueueArnAttribute, () => Future.successful(s"arn:aws:sqs:$awsRegion:$awsAccountId:${queueData.name}"))
-          )
-
-          val optionalRules = Seq(
-            queueData.deadLettersQueue
-              .map(dlq => RedrivePolicy(dlq.name, awsRegion, awsAccountId, dlq.maxReceiveCount))
-              .map(redrivePolicy =>
-                Rule(RedrivePolicyParameter, () => Future.successful(redrivePolicy.toJson.toString))
-              )
-          )
-
-          val fifoRules = if (queueData.isFifo) {
-            Seq(
-              Rule(FifoAttributeNames.FifoQueue, () => Future.successful(queueData.isFifo.toString)),
-              Rule(
-                FifoAttributeNames.ContentBasedDeduplication,
-                () => Future.successful(queueData.hasContentBasedDeduplication.toString)
-              )
-            )
-          } else {
-            Seq()
-          }
 
-          val rules = alwaysAvailableParameterRules ++ optionalRules.flatten ++ fifoRules
-          attributeValuesCalculator.calculate(attributeNames, rules: _*)
-        }
+        val attributesFuture = getQueueAttributes(p, queueActor, queueData)
 
         def responseXml(attributes: List[(String, String)]) = {
           <GetQueueAttributesResponse>
@@ -140,10 +27,6 @@ trait QueueAttributesDirectives {
           </GetQueueAttributesResponse>
         }
 
-        val attributeNames = attributeNamesReader.read(p, AllAttributeNames)
-        val attributesFuture =
-          Future.sequence(calculateAttributeValues(attributeNames).map(p => p._2.map((p._1, _))))
-
         attributesFuture.map { attributes =>
           respondWith {
             responseXml(attributes)
@@ -156,55 +39,7 @@ trait QueueAttributesDirectives {
   def setQueueAttributes(p: AnyParams): Route = {
     p.action(SetQueueAttributes) {
       queueActorFromRequest(p) { queueActor =>
-        val attributes = attributeNameAndValuesReader.read(p)
-
-        val result = attributes.map({ case (attributeName, attributeValue) =>
-          attributeName match {
-            case VisibilityTimeoutParameter =>
-              queueActor ? UpdateQueueDefaultVisibilityTimeout(
-                MillisVisibilityTimeout.fromSeconds(attributeValue.toLong)
-              )
-            case DelaySecondsAttribute =>
-              queueActor ? UpdateQueueDelay(Duration.standardSeconds(attributeValue.toLong))
-            case ReceiveMessageWaitTimeSecondsAttribute =>
-              queueActor ? UpdateQueueReceiveMessageWait(Duration.standardSeconds(attributeValue.toLong))
-            case RedrivePolicyParameter =>
-              val redrivePolicy =
-                try {
-                  import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
-                  attributeValue.parseJson.convertTo[BackwardCompatibleRedrivePolicy]
-                } catch {
-                  case e: DeserializationException =>
-                    logger.warn("Cannot deserialize the redrive policy attribute", e)
-                    throw new SQSException("MalformedQueryString")
-                  case e: ParsingException =>
-                    logger.warn("Cannot parse the redrive policy attribute", e)
-                    throw new SQSException("MalformedQueryString")
-                }
-              async {
-                val deadLettersQueueActor = await(queueManagerActor ? LookupQueue(redrivePolicy.queueName))
-                if (deadLettersQueueActor.isEmpty) {
-                  throw SQSException.nonExistentQueue
-                }
-
-                if (redrivePolicy.maxReceiveCount < 1 || redrivePolicy.maxReceiveCount > 1000) {
-                  throw SQSException.invalidParameterValue
-                }
-                queueActor ? UpdateQueueDeadLettersQueue(
-                  Some(DeadLettersQueueData(redrivePolicy.queueName, redrivePolicy.maxReceiveCount)),
-                  deadLettersQueueActor
-                )
-              }
-            case attr
-                if UnsupportedAttributeNames.AllUnsupportedAttributeNames
-                  .contains(attr) =>
-              logger.warn("Ignored attribute \"" + attr + "\" (supported by SQS but not ElasticMQ)")
-              Future.successful(())
-            case attr =>
-              logger.warn("Unsupported attribute \"" + attr + "\" (failing on ElasticMQ)")
-              Future.failed(new SQSException("InvalidAttributeName"))
-          }
-        })
+        val result = setQueueAttributes(p, queueActor, queueManagerActor)
 
         Future.sequence(result).map { _ =>
           respondWith {
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/QueueAttributesOps.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/QueueAttributesOps.scala
new file mode 100644
index 00000000..bbe6cef3
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/sqs/QueueAttributesOps.scala
@@ -0,0 +1,193 @@
+package org.elasticmq.rest.sqs
+
+import akka.actor.ActorRef
+import akka.util.Timeout
+import org.elasticmq.actor.reply._
+import org.elasticmq.msg._
+import org.elasticmq.rest.sqs.Constants._
+import org.elasticmq.rest.sqs.model.RedrivePolicy
+import org.elasticmq.rest.sqs.model.RedrivePolicy.BackwardCompatibleRedrivePolicy
+import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
+import org.elasticmq.util.Logging
+import org.elasticmq.{DeadLettersQueueData, MillisVisibilityTimeout, QueueData}
+import org.joda.time.Duration
+import spray.json.JsonParser.ParsingException
+import spray.json._
+
+import scala.async.Async.{async, await}
+import scala.concurrent.{ExecutionContext, Future}
+
+trait QueueAttributesOps extends AttributesModule {
+  this: Logging =>
+
+  def awsRegion: String
+
+  def awsAccountId: String
+
+  val attributeValuesCalculator = new AttributeValuesCalculator
+
+  object QueueWriteableAttributeNames {
+    val AllWriteableAttributeNames: List[String] = VisibilityTimeoutParameter :: DelaySecondsAttribute ::
+      ReceiveMessageWaitTimeSecondsAttribute :: RedrivePolicyParameter :: Nil
+  }
+
+  object UnsupportedAttributeNames {
+    val PolicyAttribute = "Policy"
+    val MaximumMessageSizeAttribute = "MaximumMessageSize"
+    val MessageRetentionPeriodAttribute = "MessageRetentionPeriod"
+    val FifoQueueAttribute = "FifoQueue"
+
+    val AllUnsupportedAttributeNames: List[String] = PolicyAttribute :: MaximumMessageSizeAttribute ::
+      MessageRetentionPeriodAttribute :: FifoQueueAttribute :: Nil
+  }
+
+  object FifoAttributeNames {
+    val ContentBasedDeduplication = "ContentBasedDeduplication"
+    val FifoQueue = "FifoQueue"
+
+    val AllFifoAttributeNames: Seq[String] = Seq(
+      ContentBasedDeduplication,
+      FifoQueue
+    )
+  }
+
+  object QueueReadableAttributeNames {
+    val ApproximateNumberOfMessagesAttribute = "ApproximateNumberOfMessages"
+    val ApproximateNumberOfMessagesNotVisibleAttribute =
+      "ApproximateNumberOfMessagesNotVisible"
+    val ApproximateNumberOfMessagesDelayedAttribute =
+      "ApproximateNumberOfMessagesDelayed"
+    val CreatedTimestampAttribute = "CreatedTimestamp"
+    val LastModifiedTimestampAttribute = "LastModifiedTimestamp"
+
+    val AllAttributeNames: List[String] = QueueWriteableAttributeNames.AllWriteableAttributeNames ++
+      (ApproximateNumberOfMessagesAttribute ::
+        ApproximateNumberOfMessagesNotVisibleAttribute ::
+        ApproximateNumberOfMessagesDelayedAttribute ::
+        CreatedTimestampAttribute ::
+        LastModifiedTimestampAttribute ::
+        QueueArnAttribute :: Nil) ++ FifoAttributeNames.AllFifoAttributeNames
+  }
+
+  def getAllQueueAttributes(queueActor: ActorRef, queueData: QueueData)(implicit timeout: Timeout, executionContext: ExecutionContext): Future[List[(String, String)]] = {
+    val attributesParams = attributeNamesReader.prepareParametersForRead(QueueReadableAttributeNames.AllAttributeNames)
+    getQueueAttributes(attributesParams, queueActor, queueData)
+  }
+
+  def getQueueAttributes(p: AnyParams, queueActor: ActorRef, queueData: QueueData)(implicit timeout: Timeout, executionContext: ExecutionContext): Future[List[(String, String)]] = {
+    import QueueReadableAttributeNames._
+    import org.elasticmq.rest.sqs.model.RedrivePolicyJson._
+
+    def calculateAttributeValues(attributeNames: List[String]): List[(String, Future[String])] = {
+      lazy val stats = queueActor ? GetQueueStatistics(System.currentTimeMillis())
+
+      val alwaysAvailableParameterRules = Seq(
+        AttributeValuesCalculator.Rule(
+          VisibilityTimeoutParameter,
+          () => Future.successful(queueData.defaultVisibilityTimeout.seconds.toString)
+        ),
+        AttributeValuesCalculator.Rule(DelaySecondsAttribute, () => Future.successful(queueData.delay.getStandardSeconds.toString)),
+        AttributeValuesCalculator.Rule(ApproximateNumberOfMessagesAttribute, () => stats.map(_.approximateNumberOfVisibleMessages.toString)),
+        AttributeValuesCalculator.Rule(
+          ApproximateNumberOfMessagesNotVisibleAttribute,
+          () => stats.map(_.approximateNumberOfInvisibleMessages.toString)
+        ),
+        AttributeValuesCalculator.Rule(
+          ApproximateNumberOfMessagesDelayedAttribute,
+          () => stats.map(_.approximateNumberOfMessagesDelayed.toString)
+        ),
+        AttributeValuesCalculator.Rule(CreatedTimestampAttribute, () => Future.successful((queueData.created.getMillis / 1000L).toString)),
+        AttributeValuesCalculator.Rule(
+          LastModifiedTimestampAttribute,
+          () => Future.successful((queueData.lastModified.getMillis / 1000L).toString)
+        ),
+        AttributeValuesCalculator.Rule(
+          ReceiveMessageWaitTimeSecondsAttribute,
+          () => Future.successful(queueData.receiveMessageWait.getStandardSeconds.toString)
+        ),
+        AttributeValuesCalculator.Rule(QueueArnAttribute, () => Future.successful(s"arn:aws:sqs:$awsRegion:$awsAccountId:${queueData.name}"))
+      )
+
+      val optionalRules = Seq(
+        queueData.deadLettersQueue
+          .map(dlq => RedrivePolicy(dlq.name, awsRegion, awsAccountId, dlq.maxReceiveCount))
+          .map(redrivePolicy =>
+            AttributeValuesCalculator.Rule(RedrivePolicyParameter, () => Future.successful(redrivePolicy.toJson.toString))
+          )
+      )
+
+      val fifoRules = if (queueData.isFifo) {
+        Seq(
+          AttributeValuesCalculator.Rule(FifoAttributeNames.FifoQueue, () => Future.successful(queueData.isFifo.toString)),
+          AttributeValuesCalculator.Rule(
+            FifoAttributeNames.ContentBasedDeduplication,
+            () => Future.successful(queueData.hasContentBasedDeduplication.toString)
+          )
+        )
+      } else {
+        Seq()
+      }
+
+      val rules = alwaysAvailableParameterRules ++ optionalRules.flatten ++ fifoRules
+      attributeValuesCalculator.calculate(attributeNames, rules: _*)
+    }
+
+
+    val attributeNames = attributeNamesReader.read(p, AllAttributeNames)
+    Future.sequence(calculateAttributeValues(attributeNames).map(p => p._2.map((p._1, _))))
+
+
+  }
+
+  def setQueueAttributes(p: AnyParams, queueActor: ActorRef, queueManagerActor: ActorRef)(implicit timeout: Timeout, executionContext: ExecutionContext) = {
+    val attributes = attributeNameAndValuesReader.read(p)
+
+    attributes.map({ case (attributeName, attributeValue) =>
+      attributeName match {
+        case VisibilityTimeoutParameter =>
+          queueActor ? UpdateQueueDefaultVisibilityTimeout(
+            MillisVisibilityTimeout.fromSeconds(attributeValue.toLong)
+          )
+        case DelaySecondsAttribute =>
+          queueActor ? UpdateQueueDelay(Duration.standardSeconds(attributeValue.toLong))
+        case ReceiveMessageWaitTimeSecondsAttribute =>
+          queueActor ? UpdateQueueReceiveMessageWait(Duration.standardSeconds(attributeValue.toLong))
+        case RedrivePolicyParameter =>
+          val redrivePolicy =
+            try {
+              attributeValue.parseJson.convertTo[BackwardCompatibleRedrivePolicy]
+            } catch {
+              case e: DeserializationException =>
+                logger.warn("Cannot deserialize the redrive policy attribute", e)
+                throw new SQSException("MalformedQueryString")
+              case e: ParsingException =>
+                logger.warn("Cannot parse the redrive policy attribute", e)
+                throw new SQSException("MalformedQueryString")
+            }
+          async {
+            val deadLettersQueueActor = await(queueManagerActor ? LookupQueue(redrivePolicy.queueName))
+            if (deadLettersQueueActor.isEmpty) {
+              throw SQSException.nonExistentQueue
+            }
+
+            if (redrivePolicy.maxReceiveCount < 1 || redrivePolicy.maxReceiveCount > 1000) {
+              throw SQSException.invalidParameterValue
+            }
+            queueActor ? UpdateQueueDeadLettersQueue(
+              Some(DeadLettersQueueData(redrivePolicy.queueName, redrivePolicy.maxReceiveCount)),
+              deadLettersQueueActor
+            )
+          }
+        case attr
+          if UnsupportedAttributeNames.AllUnsupportedAttributeNames
+            .contains(attr) =>
+          logger.warn("Ignored attribute \"" + attr + "\" (supported by SQS but not ElasticMQ)")
+          Future.successful(())
+        case attr =>
+          logger.warn("Unsupported attribute \"" + attr + "\" (failing on ElasticMQ)")
+          Future.failed(new SQSException("InvalidAttributeName"))
+      }
+    })
+  }
+
+}
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 32e1fd74..c56620e3 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
@@ -144,7 +144,8 @@ case class TheSQSRestServerBuilder(
       with AttributesModule
       with TagQueueDirectives
       with TagsModule
-      with UnmatchedActionRoutes {
+      with UnmatchedActionRoutes
+      with QueueAttributesOps {
 
       def serverAddress = currentServerAddress.get()
 
@@ -235,7 +236,7 @@ case class TheSQSRestServerBuilder(
 
     SQSRestServer(
       appStartFuture,
-      () => appStartFuture.flatMap(_.unbind()).flatMap(_ => stopActorSystem())
+      () => appStartFuture.flatMap(_.terminate(1.minute))
     )
   }
 
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 db6c4a07..ceed4d16 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
@@ -30,6 +30,10 @@ trait QueueDirectives {
     queueNameFromRequest(p) { queueName => queueActor(queueName, qa => body(qa, queueName)) }
   }
 
+  def queueActorAndDataFromQueueName(queueName: String)(body: (ActorRef, QueueData) => Route): Route = {
+    queueActor(queueName, qa => queueData(qa, qd => body(qa, qd)))
+  }
+
   private val queueUrlParameter = "QueueUrl"
 
   private def queueUrlFromParams(p: AnyParams): Directive1[String] =
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsDirectives.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsDirectives.scala
new file mode 100644
index 00000000..db37c6ff
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsDirectives.scala
@@ -0,0 +1,91 @@
+package org.elasticmq.rest.stats
+
+import akka.actor.ActorRef
+import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
+import akka.http.scaladsl.model.StatusCodes.NotFound
+import akka.http.scaladsl.server.Route
+import org.elasticmq.metrics.QueueMetricsOps
+import org.elasticmq.rest.sqs.QueueAttributesOps
+import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
+import org.elasticmq.util.NowProvider
+import org.elasticmq.{QueueData, QueueStatistics}
+import spray.json._
+
+import scala.concurrent.{Future}
+import scala.util.{Failure, Success}
+
+final case class QueuesResponse(
+                                 name: String,
+                                 statistics: QueueStatisticsResponse
+                               )
+
+final case class QueueResponse(
+                                name: String,
+                                attributes: Attributes
+                              )
+
+final case class QueueStatisticsResponse(
+                                          approximateNumberOfVisibleMessages: Long,
+                                          approximateNumberOfInvisibleMessages: Long,
+                                          approximateNumberOfMessagesDelayed: Long
+                                        )
+
+
+trait JsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
+  implicit val queueStatisticsFormat = jsonFormat3(QueueStatisticsResponse)
+  implicit val queuesFormat = jsonFormat2(QueuesResponse)
+  implicit val queueFormat = jsonFormat2(QueueResponse)
+}
+
+trait StatisticsDirectives extends JsonSupport {
+  this: ElasticMQDirectives with QueueAttributesOps  =>
+
+  implicit val duration = timeout.duration
+
+  def statistics(implicit np: NowProvider) = {
+    pathPrefix("statistics" / "queues") {
+      concat(
+        pathEndOrSingleSlash {
+          complete {
+            gatherAllQueuesWithStats
+          }
+        },
+        path(Segment) { queueName =>
+          gatherSpecificQueueWithAttributesRoute(queueName)
+        }
+      )
+    }
+  }
+
+  def gatherAllQueuesWithStats(implicit np: NowProvider): Future[Iterable[QueuesResponse]] = {
+
+    QueueMetricsOps.getQueuesStatistics(queueManagerActor, np)
+      .map { x => x.map { case (name, stats) => mapToRestQueuesResponse(name, stats) } }
+  }
+
+  def gatherSpecificQueueWithAttributesRoute(queueName: String): Route = {
+    queueActorAndDataFromQueueName(queueName) { (queueActor, queueData) =>
+      onComplete(getQueryResponseWithAttributesFuture(queueName, queueActor, queueData)) {
+        case Success(value) => complete(value)
+        case Failure(ex) =>
+          logger.error(s"Error while loading statistics for queue ${queueName}", ex)
+          complete(NotFound, s"Can't load data for queue ${queueName}")
+      }
+    }
+  }
+
+  private def getQueryResponseWithAttributesFuture(name: String, queueActor: ActorRef, queueData: QueueData): Future[QueueResponse] = {
+    getAllQueueAttributes(queueActor, queueData)
+      .map(list => QueueResponse(name = name, list.toMap))
+  }
+
+  private def mapToRestQueuesResponse(queueName: String, statistics: QueueStatistics): QueuesResponse = {
+    QueuesResponse(queueName,
+      QueueStatisticsResponse(
+        approximateNumberOfInvisibleMessages = statistics.approximateNumberOfInvisibleMessages,
+        approximateNumberOfMessagesDelayed = statistics.approximateNumberOfMessagesDelayed,
+        approximateNumberOfVisibleMessages = statistics.approximateNumberOfVisibleMessages
+      )
+    )
+  }
+}
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsRestServerBuilder.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsRestServerBuilder.scala
new file mode 100644
index 00000000..037fbe69
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/StatisticsRestServerBuilder.scala
@@ -0,0 +1,150 @@
+package org.elasticmq.rest.stats
+
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
+
+import akka.actor.{ActorRef, ActorSystem}
+import akka.http.scaladsl.Http
+import akka.stream.ActorMaterializer
+import akka.util.Timeout
+import org.elasticmq._
+import org.elasticmq.rest.sqs.QueueAttributesOps
+import org.elasticmq.rest.sqs.directives.ElasticMQDirectives
+import org.elasticmq.util.{Logging, NowProvider}
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, Future}
+import scala.util.control.NonFatal
+
+case class TheStatisticsRestServerBuilder(
+                                           providedActorSystem: ActorSystem,
+                                           providedQueueManagerActor: ActorRef,
+                                           interface: String,
+                                           port: Int,
+                                           serverAddress: NodeAddress,
+                                           generateServerAddress: Boolean,
+                                           _awsRegion: String,
+                                           _awsAccountId: String
+                                         ) extends Logging {
+
+  /** @param _actorSystem Optional actor system. If one is provided, it will be used to create ElasticMQ and Spray
+    *                     actors, but its lifecycle (shutdown) will be not managed by the server. If one is not
+    *                     provided, an actor system will be created, and its lifecycle will be bound to the server's
+    *                     lifecycle.
+    */
+  def withActorSystem(_actorSystem: ActorSystem) =
+    this.copy(providedActorSystem = _actorSystem)
+
+  /** @param _queueManagerActor Optional "main" ElasticMQ actor.
+    */
+  def withQueueManagerActor(_queueManagerActor: ActorRef) =
+    this.copy(providedQueueManagerActor = _queueManagerActor)
+
+  /** @param _interface Hostname to which the server will bind.
+    */
+  def withInterface(_interface: String) = this.copy(interface = _interface)
+
+  /** @param _port Port to which the server will bind.
+    */
+  def withPort(_port: Int) = this.copy(port = _port)
+
+  /** Will assign port automatically (uses port 0). The port to which the socket binds will be logged on successful startup.
+    */
+  def withDynamicPort() = withPort(0)
+
+  /** @param _serverAddress Address which will be returned as the queue address. Requests to this address
+    *                       should be routed to this server.
+    */
+  def withServerAddress(_serverAddress: NodeAddress) =
+    this.copy(serverAddress = _serverAddress, generateServerAddress = false)
+
+  /** @param region Region which will be included in ARM resource ids.
+    */
+  def withAWSRegion(region: String) =
+    this.copy(_awsRegion = region)
+
+  /** @param accountId AccountId which will be included in ARM resource ids.
+    */
+  def withAWSAccountId(accountId: String) =
+    this.copy(_awsAccountId = accountId)
+
+  def start(): StatisticsRestServer = {
+
+    implicit val nowProvider = new NowProvider()
+
+    val theServerAddress =
+      if (generateServerAddress)
+        NodeAddress(host = if (interface.isEmpty) "localhost" else interface, port = port)
+      else serverAddress
+
+    implicit val implicitActorSystem = providedActorSystem
+    implicit val implicitMaterializer = ActorMaterializer()
+
+    val currentServerAddress =
+      new AtomicReference[NodeAddress](theServerAddress)
+
+    val env = new StatisticsDirectives
+      with QueueAttributesOps
+      with ElasticMQDirectives {
+
+      def serverAddress = currentServerAddress.get()
+
+      lazy val actorSystem = providedActorSystem
+      lazy val materializer = implicitMaterializer
+      lazy val queueManagerActor = providedQueueManagerActor
+      lazy val timeout = Timeout(21, TimeUnit.SECONDS) // see application.conf
+
+      lazy val awsRegion: String = _awsRegion
+      lazy val awsAccountId: String = _awsAccountId
+
+    }
+
+    import env._
+
+    val routes =
+      handleServerExceptions {
+        statistics
+      }
+
+    val appStartFuture = Http().newServerAt(interface, port).bindFlow(routes)
+
+    appStartFuture.foreach { sb: Http.ServerBinding =>
+      if (generateServerAddress && port != sb.localAddress.getPort) {
+        currentServerAddress.set(theServerAddress.copy(port = sb.localAddress.getPort))
+      }
+
+      TheStatisticsRestServerBuilder.this.logger.info(
+        "Started statistics rest server, bind address %s:%d, visible server address %s"
+          .format(
+            interface,
+            sb.localAddress.getPort,
+            if (env.serverAddress.isWildcard)
+              "* (depends on incoming request path) "
+            else env.serverAddress.fullAddress
+          )
+      )
+    }
+
+    appStartFuture.failed.foreach { case NonFatal(e) =>
+      TheStatisticsRestServerBuilder.this.logger
+        .error("Cannot start statistics rest server, bind address %s:%d".format(interface, port), e)
+    }
+
+    StatisticsRestServer(
+      appStartFuture,
+      () => appStartFuture.flatMap(_.terminate(1.minute))
+    )
+  }
+}
+
+case class StatisticsRestServer(startFuture: Future[Http.ServerBinding], stopAndGetFuture: () => Future[Any]) {
+  def waitUntilStarted() = {
+    Await.result(startFuture, 1.minute)
+  }
+
+  def stopAndWait() = {
+    val stopFuture = stopAndGetFuture()
+    Await.result(stopFuture, 1.minute)
+  }
+}
+
diff --git a/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/package.scala b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/package.scala
new file mode 100644
index 00000000..5a59f115
--- /dev/null
+++ b/rest/rest-sqs/src/main/scala/org/elasticmq/rest/stats/package.scala
@@ -0,0 +1,6 @@
+package org.elasticmq.rest
+
+package object stats {
+  type Attributes = Map[String, String]
+
+}
diff --git a/server/src/main/resources/reference.conf b/server/src/main/resources/reference.conf
index ac2e4679..49772325 100644
--- a/server/src/main/resources/reference.conf
+++ b/server/src/main/resources/reference.conf
@@ -21,6 +21,12 @@ rest-sqs {
     sqs-limits = strict
 }
 
+rest-stats {
+    enabled = true
+    bind-port = 9325
+    bind-hostname = "0.0.0.0"
+}
+
 // Should the node-address be generated from the bind port/hostname
 // Set this to true e.g. when assigning port automatically by using port 0.
 generate-node-address = false
diff --git a/server/src/main/scala/org/elasticmq/server/ElasticMQServer.scala b/server/src/main/scala/org/elasticmq/server/ElasticMQServer.scala
index e84d9480..67f56c85 100644
--- a/server/src/main/scala/org/elasticmq/server/ElasticMQServer.scala
+++ b/server/src/main/scala/org/elasticmq/server/ElasticMQServer.scala
@@ -5,13 +5,14 @@ import akka.util.Timeout
 import org.elasticmq.actor.QueueManagerActor
 import org.elasticmq.actor.reply._
 import org.elasticmq.rest.sqs.{CreateQueueDirectives, SQSRestServer, TheSQSRestServerBuilder}
+import org.elasticmq.rest.stats.{StatisticsRestServer, TheStatisticsRestServerBuilder}
 import org.elasticmq.server.config.{CreateQueue, ElasticMQServerConfig}
 import org.elasticmq.util.{Logging, NowProvider}
 import org.elasticmq.{DeadLettersQueueData, ElasticMQError, MillisVisibilityTimeout, QueueData}
 import org.joda.time.{DateTime, Duration}
 
-import scala.concurrent.Await
 import scala.concurrent.duration.Duration.Inf
+import scala.concurrent.{Await, ExecutionContextExecutor, Future}
 
 class ElasticMQServer(config: ElasticMQServerConfig) extends Logging {
   val actorSystem = ActorSystem("elasticmq")
@@ -19,10 +20,20 @@ class ElasticMQServer(config: ElasticMQServerConfig) extends Logging {
   def start() = {
     val queueManagerActor = createBase()
     val restServerOpt = optionallyStartRestSqs(queueManagerActor)
+    val restStatisticsServerOpt = optionallyStartRestStatistics(queueManagerActor)
 
     val shutdown = () => {
-      restServerOpt.map(_.stopAndGetFuture())
-      Await.result(actorSystem.terminate(), Inf)
+      implicit val ec: ExecutionContextExecutor = actorSystem.dispatcher
+
+      val futureTerminationRestSQS = restServerOpt.map(_.stopAndGetFuture()).getOrElse(Future.unit)
+      val futureTerminationRestStats = restStatisticsServerOpt.map(_.stopAndGetFuture()).getOrElse(Future.unit)
+      val eventualTerminated = for {
+        _ <- futureTerminationRestSQS
+        _ <- futureTerminationRestStats
+        ac <- actorSystem.terminate()
+      } yield ac
+      Await.result(eventualTerminated, Inf)
+
     }
 
     createQueues(queueManagerActor) match {
@@ -65,6 +76,28 @@ class ElasticMQServer(config: ElasticMQServerConfig) extends Logging {
     }
   }
 
+  private def optionallyStartRestStatistics(queueManagerActor: ActorRef): Option[StatisticsRestServer] = {
+    if (config.restStatisticsConfiguration.enabled) {
+
+      val server = TheStatisticsRestServerBuilder(
+        actorSystem,
+        queueManagerActor,
+        config.restStatisticsConfiguration.bindHostname,
+        config.restStatisticsConfiguration.bindPort,
+        config.nodeAddress,
+        config.generateNodeAddress,
+        config.awsRegion,
+        config.awsAccountId
+      ).start()
+
+      server.waitUntilStarted()
+
+      Some(server)
+    } else {
+      None
+    }
+  }
+
   private def createQueues(queueManagerActor: ActorRef): List[ElasticMQError] = {
     implicit val timeout = {
       import scala.concurrent.duration._
diff --git a/server/src/main/scala/org/elasticmq/server/config/ElasticMQServerConfig.scala b/server/src/main/scala/org/elasticmq/server/config/ElasticMQServerConfig.scala
index d73c0c65..99ea721f 100644
--- a/server/src/main/scala/org/elasticmq/server/config/ElasticMQServerConfig.scala
+++ b/server/src/main/scala/org/elasticmq/server/config/ElasticMQServerConfig.scala
@@ -59,6 +59,16 @@ class ElasticMQServerConfig(config: Config) extends Logging {
 
   val restSqs = new RestSqsConfiguration
 
+  class RestStatisticsConfiguration {
+    private val subConfig = config.getConfig("rest-stats")
+    val enabled = subConfig.getBoolean("enabled")
+    val bindPort = subConfig.getInt("bind-port")
+    val bindHostname = subConfig.getString("bind-hostname")
+  }
+
+  val restStatisticsConfiguration = new RestStatisticsConfiguration
+
+
   val createQueues: List[CreateQueue] = {
     def getOptionalBoolean(c: Config, k: String) = if (c.hasPath(k)) Some(c.getBoolean(k)) else None
     def getOptionalDuration(c: Config, k: String) = if (c.hasPath(k)) Some(c.getDuration(k, TimeUnit.SECONDS)) else None


Test patch:
diff --git a/core/src/test/scala/org/elasticmq/actor/queue/ReceiveRequestAttemptCacheTest.scala b/core/src/test/scala/org/elasticmq/actor/queue/ReceiveRequestAttemptCacheTest.scala
index d57caaa9..722e6567 100644
--- a/core/src/test/scala/org/elasticmq/actor/queue/ReceiveRequestAttemptCacheTest.scala
+++ b/core/src/test/scala/org/elasticmq/actor/queue/ReceiveRequestAttemptCacheTest.scala
@@ -3,7 +3,7 @@ package org.elasticmq.actor.queue
 import scala.collection.mutable
 import org.elasticmq.NeverReceived
 import org.elasticmq.actor.queue.ReceiveRequestAttemptCache.ReceiveFailure.Invalid
-import org.elasticmq.actor.test.MutableNowProvider
+import org.elasticmq.util.MutableNowProvider
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
 
diff --git a/core/src/test/scala/org/elasticmq/actor/test/QueueManagerForEachTest.scala b/core/src/test/scala/org/elasticmq/actor/test/QueueManagerForEachTest.scala
index a699f0b2..0bcaf802 100644
--- a/core/src/test/scala/org/elasticmq/actor/test/QueueManagerForEachTest.scala
+++ b/core/src/test/scala/org/elasticmq/actor/test/QueueManagerForEachTest.scala
@@ -3,6 +3,7 @@ package org.elasticmq.actor.test
 import akka.actor.{ActorRef, ActorSystem, Props}
 import org.elasticmq.StrictSQSLimits
 import org.elasticmq.actor.QueueManagerActor
+import org.elasticmq.util.MutableNowProvider
 import org.scalatest.{BeforeAndAfterEach, Suite}
 
 trait QueueManagerForEachTest extends BeforeAndAfterEach {
diff --git a/core/src/test/scala/org/elasticmq/actor/test/MutableNowProvider.scala b/core/src/test/scala/org/elasticmq/util/MutableNowProvider.scala
similarity index 51%
rename from core/src/test/scala/org/elasticmq/actor/test/MutableNowProvider.scala
rename to core/src/test/scala/org/elasticmq/util/MutableNowProvider.scala
index 2fc85a35..4e3844e9 100644
--- a/core/src/test/scala/org/elasticmq/actor/test/MutableNowProvider.scala
+++ b/core/src/test/scala/org/elasticmq/util/MutableNowProvider.scala
@@ -1,12 +1,15 @@
-package org.elasticmq.actor.test
+package org.elasticmq.util
 
-import org.joda.time.DateTime
-import org.elasticmq.util.NowProvider
 import java.util.concurrent.atomic.AtomicLong
 
-class MutableNowProvider extends NowProvider {
-  var mutableNowMillis = new AtomicLong(100L)
+import org.joda.time.DateTime
+
+class MutableNowProvider(initalValue: Long = 100L) extends NowProvider {
+  var mutableNowMillis = new AtomicLong(initalValue)
 
   override def nowMillis = mutableNowMillis.get()
+
   override def now = new DateTime(nowMillis)
 }
+
+
diff --git a/rest/rest-sqs/src/test/scala/org/elasticmq/rest/stats/StatisticsDirectivesTest.scala b/rest/rest-sqs/src/test/scala/org/elasticmq/rest/stats/StatisticsDirectivesTest.scala
new file mode 100644
index 00000000..21e1bf9f
--- /dev/null
+++ b/rest/rest-sqs/src/test/scala/org/elasticmq/rest/stats/StatisticsDirectivesTest.scala
@@ -0,0 +1,113 @@
+package org.elasticmq.rest.stats
+
+import akka.actor.{ActorRef, ActorSystem, Props}
+import akka.http.scaladsl.server.Directives
+import akka.http.scaladsl.testkit.ScalatestRouteTest
+import akka.util.Timeout
+import org.elasticmq.actor.QueueManagerActor
+import org.elasticmq.actor.reply.ReplyActorRef
+import org.elasticmq.msg.{CreateQueue, DeleteQueue}
+import org.elasticmq.rest.sqs.directives._
+import org.elasticmq.rest.sqs.{ActorSystemModule, QueueAttributesOps, QueueManagerActorModule}
+import org.elasticmq.util.MutableNowProvider
+import org.elasticmq.{MillisVisibilityTimeout, QueueData, StrictSQSLimits}
+import org.joda.time.{DateTime, Duration}
+import org.scalatest._
+import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.concurrent.duration.DurationInt
+
+class StatisticsDirectivesTest extends AnyFlatSpec
+  with Matchers
+  with ScalatestRouteTest
+  with Directives
+  with QueueDirectives
+  with AnyParamDirectives
+  with QueueManagerActorModule
+  with FutureDirectives
+  with ActorSystemModule
+  with ExceptionDirectives
+  with RespondDirectives
+  with StatisticsDirectives
+  with ElasticMQDirectives
+  with QueueAttributesOps
+  with BeforeAndAfter
+  with ScalaFutures
+  with IntegrationPatience {
+
+  def awsAccountId: String = "id"
+
+  def awsRegion: String = "region"
+
+  implicit lazy val timeout: Timeout = 1.minute
+  implicit lazy val actorSystem: ActorSystem = ActorSystem("test-actor-system")
+
+  implicit val nowProvider = new MutableNowProvider(
+    new DateTime().withDate(2020, 1, 1).withTimeAtStartOfDay().getMillis)
+
+  lazy val queueManagerActor: ActorRef =
+    actorSystem.actorOf(Props(new QueueManagerActor(nowProvider, StrictSQSLimits)))
+
+
+  "statisticsRequestForAllQueues" should "return all queues statistics" in {
+
+    val route = {
+      statistics
+    }
+
+    Get("/statistics/queues") ~> route ~> check {
+      responseAs[List[QueuesResponse]] should contain theSameElementsAs List(
+        QueuesResponse("firstQueue", QueueStatisticsResponse(0, 0, 0)),
+        QueuesResponse("secondQueue", QueueStatisticsResponse(0, 0, 0))
+      )
+    }
+  }
+
+  "statisticsRequestForCertainQueue" should "return statistics for specified queue" in {
+
+    val route = {
+      statistics
+    }
+
+    Get("/statistics/queues/firstQueue") ~> route ~> check {
+      val queueResponse = responseAs[QueueResponse]
+      val expectedNowTimeInMillis = (nowProvider.nowMillis/1000L).toString
+      queueResponse.name shouldEqual "firstQueue"
+      queueResponse.attributes should contain theSameElementsAs Map("ApproximateNumberOfMessagesDelayed" -> "0",
+        "VisibilityTimeout" -> "0",
+        "ApproximateNumberOfMessagesNotVisible" -> "0",
+        "LastModifiedTimestamp" -> expectedNowTimeInMillis,
+        "QueueArn" -> "arn:aws:sqs:region:id:firstQueue",
+        "CreatedTimestamp" -> expectedNowTimeInMillis,
+        "ApproximateNumberOfMessages" -> "0",
+        "ReceiveMessageWaitTimeSeconds" -> "0",
+        "DelaySeconds" -> "0"
+      )
+    }
+  }
+  
+  before {
+    createQueueWithName("firstQueue")
+    createQueueWithName("secondQueue")
+
+  }
+
+  after {
+    deleteQueueWithName("firstQueue")
+    deleteQueueWithName("secondQueue")
+  }
+
+
+  private def createQueueWithName(name: String) = {
+    (queueManagerActor ? CreateQueue(
+      QueueData(name, MillisVisibilityTimeout(1L), Duration.ZERO, Duration.ZERO, nowProvider.now, nowProvider.now)
+    )).futureValue
+  }
+
+  private def deleteQueueWithName(name: String) = {
+    (queueManagerActor ? DeleteQueue(name)).futureValue
+
+  }
+}