Fix RabbitMqMessagePublisherShould test.

* Purge the queue before performing the test in order to avoid consuming previously published events such as the ones coming from the acceptance tests.
* Since the `MessagePublisher` publishes out the messages asynchronously, we were validating that the queue was empty even before publishing the message, so we were not really waiting for the message to be consumed before asserting that the consumed message was the expected one. Now we wait until the queue has messages before starting to consume.
* The previous assertion was running in another thread and, since the default exception handler used by the RabbitMQ library is an implementation which catches the exceptions and traduce them into a log error (rabbitmq/rabbitmq-java-client#74), the test was never failing (only logging errors). Now we inject from the test to the messages consumer a handler which extracts the event into a synchronized `mutable.Buffer` in order to perform the assertions proving that the consumed messages are the ones we're really expecting. Since that assertion is performed outside the consumer, the test will fail in case the expected messages doesn't match the consumed ones.

Author: JavierCane

src/test/tv/codely/scala_http_api/module/IntegrationTestCase.scala

@@ -4,25 +4,21 @@ import com.typesafe.config.ConfigFactory
 import tv.codely.scala_http_api.module.shared.domain.MessagePublisher
 import tv.codely.scala_http_api.module.shared.infrastructure.config.{DbConfig, MessageBrokerConfig}
 import tv.codely.scala_http_api.module.shared.infrastructure.dependency_injection.SharedModuleDependencyContainer
-import tv.codely.scala_http_api.module.shared.infrastructure.message_broker.rabbitmq.{MessageConsumer, RabbitMqMessageConsumer}
 import tv.codely.scala_http_api.module.shared.infrastructure.persistence.doobie.DoobieDbConnection
 
 import scala.concurrent.ExecutionContext
 
 protected[scala_http_api] trait IntegrationTestCase extends UnitTestCase {
   private val actorSystemName = "scala-http-api-integration-test"
 
-  private val appConfig       = ConfigFactory.load("application")
-  private val dbConfig        = DbConfig(appConfig.getConfig("database"))
-  private val publisherConfig = MessageBrokerConfig(appConfig.getConfig("message-publisher"))
+  private val appConfig         = ConfigFactory.load("application")
+  private val dbConfig          = DbConfig(appConfig.getConfig("database"))
+  protected val publisherConfig = MessageBrokerConfig(appConfig.getConfig("message-publisher"))
 
   private val sharedDependencies = new SharedModuleDependencyContainer(actorSystemName, dbConfig, publisherConfig)
 
   implicit protected val executionContext: ExecutionContext = sharedDependencies.executionContext
 
   protected val doobieDbConnection: DoobieDbConnection = sharedDependencies.doobieDbConnection
   protected val messagePublisher: MessagePublisher     = sharedDependencies.messagePublisher
-
-  protected val videoCreatedQueueConsumer: MessageConsumer =
-    new RabbitMqMessageConsumer(publisherConfig)("codelytv_scala_api.video_created")
 }

src/test/tv/codely/scala_http_api/module/shared/infrastructure/message_broker/rabbitmq/MessageConsumer.scala

@@ -4,6 +4,6 @@ import tv.codely.scala_http_api.module.shared.domain.Message
 
 trait MessageConsumer {
   def startConsuming(handler: Message => Boolean): Unit
-  def hasMessages: Boolean
-  def isEmpty: Boolean = !hasMessages
+  def hasMessagesToConsume: Boolean
+  def isEmpty: Boolean = !hasMessagesToConsume
 }

src/test/tv/codely/scala_http_api/module/shared/infrastructure/message_broker/rabbitmq/MessagePurger.scala

@@ -0,0 +1,5 @@
+package tv.codely.scala_http_api.module.shared.infrastructure.message_broker.rabbitmq
+
+trait MessagePurger {
+  def purgeQueue(): Unit
+}

src/test/tv/codely/scala_http_api/module/shared/infrastructure/message_broker/rabbitmq/RabbitMqMessageConsumer.scala

@@ -38,5 +38,5 @@ final class RabbitMqMessageConsumer(brokerConfig: MessageBrokerConfig)(queueName
     channel.basicConsume(queueName, autoAckAfterConsume, consumer).map(_ => ())
   }
 
-  override def hasMessages: Boolean = channel.messageCount(queueName) > 0
+  override def hasMessagesToConsume: Boolean = channel.messageCount(queueName) > 0
 }

src/test/tv/codely/scala_http_api/module/shared/infrastructure/message_broker/rabbitmq/RabbitMqMessagePublisherShould.scala

@@ -5,31 +5,49 @@ import tv.codely.scala_http_api.module.IntegrationTestCase
 import tv.codely.scala_http_api.module.shared.domain.Message
 import tv.codely.scala_http_api.module.video.domain.VideoCreatedStub
 
+import scala.collection.mutable
 import scala.concurrent.duration._
 
 final class RabbitMqMessagePublisherShould extends IntegrationTestCase with Eventually {
   override implicit val patienceConfig: PatienceConfig = PatienceConfig(timeout = 1.second, interval = 50.millis)
 
-  // @ToDo: Fix this test. Since the `MessagePublisher` publishes out the event asynchronously,
-  // we validate that the queue is empty even before publishing the event.
-  // Furthermore, the `assertConsumes` assertion could contain an invalid actual result that it will not fail.
+  private val queueName                                  = "codelytv_scala_api.video_created"
+  private val videoCreatedQueuePurger: MessagePurger     = new RabbitMqMessagePurger(publisherConfig)(queueName)
+  private val videoCreatedQueueConsumer: MessageConsumer = new RabbitMqMessageConsumer(publisherConfig)(queueName)
+
+  private val consumedMessages: mutable.Buffer[Message] = mutable.Buffer.empty
+
   "publish VideoCreated domain events" in {
-    val videoCreated = VideoCreatedStub.random
+    videoCreatedQueuePurger.purgeQueue()
+    waitUntilQueueIsEmpty()
 
+    val videoCreated = VideoCreatedStub.random
     messagePublisher.publish(videoCreated)
+    waitUntilQueueHasMessages()
 
-    videoCreatedQueueConsumer.startConsuming(handler = assertConsumes(videoCreated))
-
+    videoCreatedQueueConsumer.startConsuming(extractConsumedMessagesHandler)
     waitUntilQueueIsEmpty()
+
+    consumedMessages.synchronized(consumedMessages shouldBe Seq(videoCreated))
   }
 
-  private def assertConsumes(expectedMessage: Message)(consumedMessage: Message): Boolean = {
-    consumedMessage shouldBe expectedMessage
-    true
+  private def extractConsumedMessagesHandler(consumedMessage: Message): Boolean = {
+    consumedMessages.synchronized(consumedMessages += consumedMessage)
+    val handledSuccessfully = true
+    handledSuccessfully
   }
 
+  private def waitUntilQueueHasMessages(): Unit = eventually(
+    if (videoCreatedQueueConsumer.hasMessagesToConsume) ()
+    else throw new RuntimeException("Queue has no messages. Waiting a little bit more…")
+  )
+
   private def waitUntilQueueIsEmpty(): Unit = eventually(
-    if (videoCreatedQueueConsumer.isEmpty) ()
-    else throw new RuntimeException("Queue is not empty. Waiting a little bit more…")
+    if (videoCreatedQueueConsumer.isEmpty) {
+      // If the RabbitMQ queue doesn't has any message, it doesn't mean we're not processing the last ones.
+      // Wait a little in order to let consuming and acknowledge these messages.
+      // More info under the `message-count` domain concept: https://www.rabbitmq.com/amqp-0-9-1-reference.html#domains
+      Thread.sleep(50)
+    } else throw new RuntimeException("Queue is not empty. Waiting a little bit more…")
   )
 }

src/test/tv/codely/scala_http_api/module/shared/infrastructure/message_broker/rabbitmq/RabbitMqMessagePurger.scala

@@ -0,0 +1,16 @@
+package tv.codely.scala_http_api.module.shared.infrastructure.message_broker.rabbitmq
+
+import com.newmotion.akka.rabbitmq.ConnectionFactory
+import tv.codely.scala_http_api.module.shared.infrastructure.config.MessageBrokerConfig
+
+final class RabbitMqMessagePurger(brokerConfig: MessageBrokerConfig)(queueName: String) extends MessagePurger {
+  private val factory = new ConnectionFactory()
+  factory.setHost(brokerConfig.host)
+  factory.setPort(brokerConfig.port)
+  factory.setUsername(brokerConfig.user)
+  factory.setPassword(brokerConfig.password)
+  private val connection = factory.newConnection()
+  private val channel    = connection.createChannel()
+
+  override def purgeQueue(): Unit = channel.queuePurge(queueName)
+}

src/test/tv/codely/scala_http_api/module/user/application/register/UserRegistererShould.scala

@@ -8,7 +8,7 @@ final class UserRegistererShould extends UserUnitTestCase with MessagePublisherM
   private val registerer = new UserRegisterer(repository, messagePublisher)
 
   "register a user" in {
-    val user = UserStub.random
+    val user           = UserStub.random
     val userRegistered = UserRegisteredStub(user)
 
     repositoryShouldSave(user)