AMAZON SQS(2)Java Client Consumer and Producer

AMAZON SQS(2)Java Client Consumer and Producer

Configuration File build.sbt
  "com.amazonaws" % "aws-java-sdk" % "1.10.6",                //

First class, SQS Client builder, SQSQueue.scala
package com.sillycat.jobsconsumer.messagequeue

import com.sillycat.jobsconsumer.utilities.{IncludeLogger, IncludeConfig}

import com.amazonaws.services.sqs.buffered.{QueueBufferConfig, AmazonSQSBufferedAsyncClient}
import com.amazonaws.services.sqs.{AmazonSQSClient, AmazonSQSAsyncClient}
import scala.collection.JavaConverters._
import com.amazonaws.services.sqs.model.{DeleteMessageRequest, ReceiveMessageRequest, SendMessageBatchRequestEntry}
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.auth.BasicAWSCredentials

object SQSQueue extends IncludeLogger with IncludeConfig{

  private def getCredential = {
    new BasicAWSCredentials(
      config.getString(envStr("sqs.keyId")),
      config.getString(envStr("sqs.accessKey")))
  }

  def getAsyncClient = {
    val client = new AmazonSQSAsyncClient(getCredential)
    client.setRegion(Region.getRegion(Regions.fromName(config.getString(envStr("sqs.region")))))
    client
  }

}

At first, I am thinking that I can have a embedded SQS server, actually I found one, but there is AKKA system conflict there. So I do not want to speed more time on that.
package com.sillycat.jobsconsumer.messagequeue

import com.amazonaws.services.sqs.AmazonSQSAsyncClient
import com.amazonaws.services.sqs.model.{ReceiveMessageRequest}
import com.sillycat.jobsconsumer.utilities.{IncludeLogger, IncludeConfig}
import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}

class SQSQueueSpec extends FunSpec with Matchers with BeforeAndAfterAll with IncludeConfig with IncludeLogger{

  val queueName = config.getString(envStr("sqs.queueName.rawjobs"))

  var client: AmazonSQSAsyncClient= _

  var queueUrl: String = _

  override def beforeAll(): Unit = {
    //build client

    if(config.getString("build.env").equals("test")){
      logger.info("No embedded SQS, ignore the test")
    }else{
      client = SQSQueue.getAsyncClient
      queueUrl = client.getQueueUrl(queueName).getQueueUrl
    }
  }

  override def afterAll() {
  }

  describe("IncludeSQSProducer") {
    describe("sendMessage"){
      it("Directly Send String") {
        if(config.getString("build.env").equals("test")){
          logger.info("No embedded SQS, ignore the test")
        }else{
          val expect = "134343143"
          client.sendMessage(queueUrl ,expect)
          val msgs = client
            .receiveMessage(new ReceiveMessageRequest((
            queueUrl
            )).withMaxNumberOfMessages(10)).getMessages
          msgs.size() should be (1)
          msgs.get(0).getBody should be (expect)
          client.deleteMessage(queueUrl,msgs.get(0).getReceiptHandle)
        }
      }
    }
  }

}

Write some code in the trait to make the consumer and producer easier.
The consumer trait, IncludeSQSConsumer.scala
package com.sillycat.jobsconsumer.messagequeue.consumer

import com.amazonaws.services.sqs.model.Message

import com.amazonaws.services.sqs.model.{DeleteMessageRequest, ReceiveMessageRequest}
import com.sillycat.jobsconsumer.messagequeue.SQSQueue
import com.sillycat.jobsconsumer.utilities.{IncludeConfig, IncludeLogger}
import scala.collection.JavaConverters._

trait IncludeSQSConsumer extends IncludeLogger with IncludeConfig{

  protected def queueName = "default"

  while(true){
    SQSQueue.getAsyncClient
      .receiveMessage(new ReceiveMessageRequest((getQueueUrl)).withMaxNumberOfMessages(10))
      .getMessages
      .asScala
      .map(handleMessage)
      .foreach(deleteMessage)
  }

  protected def getQueueUrl:String = {
    SQSQueue.getAsyncClient.getQueueUrl(queueName).getQueueUrl
  }

  protected def handleMessage(msg: Message): Message

  protected def deleteMessage(msg: Message) = {
    logger.debug("Deleting Message after operation " + msg.getReceiptHandle)
    SQSQueue.getAsyncClient.deleteMessageAsync(
      new DeleteMessageRequest(getQueueUrl,msg.getReceiptHandle))
  }

}

Simple Consumer, RawjobConsumer.scala
package com.sillycat.jobsconsumer.messagequeue.consumer

import com.amazonaws.services.sqs.model.{Message}
import com.sillycat.jobsconsumer.utilities.{IncludeConfig, IncludeLogger}

object RawJobConsumer extends App with IncludeSQSConsumer with IncludeLogger with IncludeConfig{

  override protected def queueName = {
    val queueName = config.getString(envStr("sqs.queueName.rawjobs"))
    logger.debug("Polling the Raw Job Message from Queue: " + queueName)
    queueName
  }

  protected def handleMessage(msg: Message): Message = {
    logger.debug("Receiving Message from SQS " + msg.getBody)
    msg
  }

}

trait of producer, IncludeSQSProducer.scala
package com.sillycat.jobsconsumer.messagequeue.producer

import com.amazonaws.services.sqs.model.{SendMessageRequest, SendMessageBatchRequestEntry}
import com.sillycat.jobsconsumer.messagequeue.SQSQueue
import com.sillycat.jobsconsumer.utilities.{IncludeConfig, IncludeLogger}

trait IncludeSQSProducer extends IncludeLogger with IncludeConfig{

  protected def queueName = "default"

  def sendMessage(sourceMsg:Any): Unit = {
    logger.debug("Send the Message to Queue " + getQueueUrl)
    val msg = new SendMessageRequest(getQueueUrl,prepareMessage(sourceMsg))
    SQSQueue.getAsyncClient.sendMessage(msg)
  }

  protected def prepareMessage(sourceMsg:Any):String

  private def getQueueUrl:String = {
    SQSQueue.getAsyncClient.getQueueUrl(queueName).getQueueUrl
  }

}

Simple Producer, ClassifiedJobProducer.scala
package com.sillycat.jobsconsumer.messagequeue.producer

object ClassifiedJobProducer extends IncludeSQSProducer{

  override protected def queueName = {
    val queueName = config.getString(envStr("sqs.queueName.classifiedjobs"))
    logger.debug("Send the classified Job Message to Queue: " + queueName)
    queueName
  }

  def prepareMessage(sourceMsg:Any):String = {
    sourceMsg.toString
  }

}



References:
JAVA or SCALA
http://aws.amazon.com/java/
https://github.com/adamw/mqperf/tree/master/src/main/scala/com/softwaremill/mqperf/mq

old blog
https://github.com/luohuazju/sillycat-analyzer-java
https://github.com/adamw/mqperf

Message Compress
kryo
http://blog.csdn.net/rocklee/article/details/26706145 

messagepack
http://web2.0coder.com/archives/347
http://www.cnblogs.com/peiandsky/archive/2012/04/24/2467766.html

protobuf
https://github.com/google/protobuf
https://github.com/drslump/Protobuf-PHP

serializing and deserializing XML
http://alvinalexander.com/scala/serializing-deserializing-xml-scala-classes

Testing SQS Server
http://maciejb.me/2012/10/17/testing-your-java-amazon-sqs-code-with-elasticmq/
https://github.com/adamw/elasticmq