Pulling on a socket

PullApplicationActorLess.scala

object PullApplicationActorLess extends App {

  val TOTAL_MESSAGES = 1000000
  val diagnostics = actorOf(new Diagnostics.DiagnosticsActor).start
  var context: ZMQ.Context = null
  var pullSocket: ZMQ.Socket = null

  context = ZMQ.context(1)
  pullSocket = context.socket(ZMQ.PULL)
  pullSocket.connect("tcp://127.0.0.1:5555")

  println("Starting consumer ...")

  while (true) {

    val request = pullSocket.recv(0)
    val requestString = new String(request)
    diagnostics ! "done"
  }

  object Diagnostics {
    var messagesReceived = 0
    class DiagnosticsActor extends Actor {
      def receive = {
        case msg: String =>
          messagesReceived = messagesReceived + 1
          if (messagesReceived >= TOTAL_MESSAGES) {
            // val time = System.currentTimeMillis - PushApplicationActorLess.startTime
            println("Message Processing Time: " + System.currentTimeMillis)
            //println("Throughput [Msg/s]: " + (TOTAL_MESSAGES / time) * 1000)
          }
      }
    }

  }
}

PushApplicationActorLess.scala

object PushApplicationActorLess extends App {
  val TOTAL_MESSAGES = 1000000
  val xmlString = XML.load("src/test/resources/benchmarks/message/message-one-s1.xml").toString

  var context: ZMQ.Context = null
  var pushSocket: ZMQ.Socket = null

  context = ZMQ.context(5)
  pushSocket = context.socket(ZMQ.PUSH)
  pushSocket.bind("tcp://127.0.0.1:5555")
  Thread.sleep(10)

  val startTime = System.currentTimeMillis
  println("Start time - " + startTime)
  for (i <- 1 to TOTAL_MESSAGES) {
    pushSocket.send(xmlString.getBytes, 0)
  }
  val endTime = System.currentTimeMillis

  println("Publisher elapsed time " + (endTime - startTime))

}