Skip to content

Instantly share code, notes, and snippets.

Created June 20, 2014 11:43
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
import scala.collection.JavaConverters._
import{DeleteMessageRequest, ReceiveMessageRequest, SendMessageBatchRequestEntry}
import com.softwaremill.mqperf.config.AWSCredentialsFromEnv
import com.amazonaws.regions.{Region, Regions}
class SqsMq(configMap: Map[String, String]) extends Mq {
private val asyncClient = {
val c = new AmazonSQSAsyncClient(AWSCredentialsFromEnv())
private val asyncBufferedClient = new AmazonSQSBufferedAsyncClient(asyncClient)
private val queueUrl = asyncClient.createQueue("mqperf-test-queue").getQueueUrl
override type MsgId = String
override def createSender() = new MqSender {
override def send(msgs: List[String]) = {
asyncClient.sendMessageBatch(queueUrl, { case (m, i) => new SendMessageBatchRequestEntry(i.toString, m)}.asJava
override def createReceiver() = new MqReceiver {
override def receive(maxMsgCount: Int) = {
.receiveMessage(new ReceiveMessageRequest(queueUrl).withMaxNumberOfMessages(maxMsgCount))
.map(m => (m.getReceiptHandle, m.getBody))
override def ack(ids: List[MsgId]) = {
ids.foreach { id => asyncBufferedClient.deleteMessageAsync(new DeleteMessageRequest(queueUrl, id))}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment