CheckOrdering.scala

package io.scalac.amqp

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
//import scala.concurrent.ExecutionContext.Implicits.global

object CheckOrdering extends App {

  val max = 100000
  val connection = Connection()
  //E, Q1, Q2, and binding should be created before this 'test'.
  val q1 = connection.consume(queue = "Q1")
  val q2 = connection.consume(queue = "Q2")
  val e1 = connection.publish(exchange = "E", routingKey = "q1")
  val e2 = connection.publish(exchange = "E", routingKey = "q2")

  implicit val system = ActorSystem()
  implicit val mat = ActorMaterializer()

  var expected =  BigInt(1)
  Source.fromPublisher(q2).map(_.message).runForeach(checkExpected)
  Source.fromPublisher(q1).map(_.message).runWith(Sink.fromSubscriber(e2))
  Source.fromIterator(() => (1 to max).map(makeMsg).iterator).runWith(Sink.fromSubscriber(e1))

  def checkExpected(m: Message): Unit = {
    val actual = BigInt(m.body.toArray)
    if (actual % 100 == 0) println(actual)
    if(expected != actual){
      println(s"Expected: $expected, got: $actual")
      //stop()
    }
    if(expected == BigInt(max)){
      println(s"Finished!")
      stop()
    }
    expected = expected + 1
  }

  def stop(): Unit = {
    println("Shutting down connection")
    connection.shutdown()
    println("Shutting down system")
    system.shutdown()
    println("System exit")
    System.exit(0)
  }

  def makeMsg(i: Int): Message = Message(body = BigInt(i).toByteArray)
}