<commit_before>package uk.ac.wellcome.messaging.message

import java.util.concurrent.ConcurrentLinkedQueue

import akka.stream.QueueOfferResult
import akka.stream.scaladsl.Flow
import org.mockito.Matchers.{endsWith, eq => equalTo}
import org.mockito.Mockito.{never, times, verify}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{Assertion, FunSpec, Matchers}
import uk.ac.wellcome.messaging.test.fixtures.Messaging
import uk.ac.wellcome.messaging.test.fixtures.SQS.QueuePair
import uk.ac.wellcome.monitoring.fixtures.MetricsSenderFixture
import uk.ac.wellcome.storage.ObjectLocation
import uk.ac.wellcome.test.fixtures.Akka
import uk.ac.wellcome.test.utils.ExtendedPatience
import uk.ac.wellcome.json.JsonUtil._

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

class MessageStreamTest
    extends FunSpec
    with Matchers
    with Messaging
    with Akka
    with ScalaFutures
    with ExtendedPatience
    with MetricsSenderFixture {

  def process(list: ConcurrentLinkedQueue[ExampleObject])(o: ExampleObject) = {
    list.add(o)
    Future.successful(())
  }

  it("reads messages off a queue, processes them and deletes them") {
    withMessageStreamFixtures[ExampleObject, Assertion] {
      case (bucket, messageStream, QueuePair(queue, dlq), _) =>
        val exampleObject1 = ExampleObject("some value 1")
        sendMessage(bucket, queue, exampleObject1)
        val exampleObject2 = ExampleObject("some value 2")
        sendMessage(bucket, queue, exampleObject2)

        val received = new ConcurrentLinkedQueue[ExampleObject]()
<commit_msg>Start working on MessageStreamTest<commit_after>package uk.ac.wellcome.messaging.message

import java.util.concurrent.ConcurrentLinkedQueue

import akka.stream.QueueOfferResult
import akka.stream.scaladsl.Flow
import org.mockito.Matchers.{endsWith, eq => equalTo}
import org.mockito.Mockito.{never, times, verify}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{Assertion, FunSpec, Matchers}
import uk.ac.wellcome.messaging.test.fixtures.Messaging
import uk.ac.wellcome.messaging.test.fixtures.SQS.{Queue, QueuePair}
import uk.ac.wellcome.monitoring.fixtures.MetricsSenderFixture
import uk.ac.wellcome.storage.ObjectLocation
import uk.ac.wellcome.test.fixtures.Akka
import uk.ac.wellcome.test.utils.ExtendedPatience
import uk.ac.wellcome.json.JsonUtil._
import uk.ac.wellcome.storage.fixtures.S3.Bucket

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

class MessageStreamTest
    extends FunSpec
    with Matchers
    with Messaging
    with Akka
    with ScalaFutures
    with ExtendedPatience
    with MetricsSenderFixture {

  def process(list: ConcurrentLinkedQueue[ExampleObject])(o: ExampleObject) = {
    list.add(o)
    Future.successful(())
  }

  describe("small messages (<256KB)") {
    it("reads messages off a queue, processes them and deletes them") {
      withMessageStreamFixtures[ExampleObject, Assertion] {
        case (bucket, messageStream, QueuePair(queue, dlq), _) =>
          val messages = createMessages(count = 3)

          messages.foreach { msg =>
            sendInlineNotification(bucket, queue, msg)
          }

          val received = new ConcurrentLinkedQueue[ExampleObject]()

          messageStream.foreach(
            streamName = "test-stream",
            process = process(received))

          eventually {
            received should contain theSameElementsAs messages

            assertQueueEmpty(queue)
            assertQueueEmpty(dlq)
          }
      }
    }
  }

  private def sendInlineNotification(bucket: Bucket, queue: Queue, msg: ExampleObject): Unit =
    sendMessage[MessageNotification[ExampleObject]](
      queue = queue,
      obj = InlineNotification(msg)
    )

  private def createMessages(count: Int): List[ExampleObject] =
    (1 to count)
      .map { idx => ExampleObject("a" * idx) }
      .toList

  it("reads messages off a queue, processes them and deletes them") {
    withMessageStreamFixtures[ExampleObject, Assertion] {
      case (bucket, messageStream, QueuePair(queue, dlq), _) =>
        val exampleObject1 = ExampleObject("some value 1")
        sendMessage(bucket, queue, exampleObject1)
        val exampleObject2 = ExampleObject("some value 2")
        sendMessage(bucket, queue, exampleObject2)

        val received = new ConcurrentLinkedQueue[ExampleObject]()

{'fraction_non_alphanumeric': 0.09138086642599277, 'fraction_numerical': 0.004061371841155234, 'mean_word_length': 3.947544642857143, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7254698', 'n_tokens_mistral': 1321, 'n_tokens_neox': 1286, 'n_words': 277}