CODEWITHSUNDEEP
javascalaamazon-web-servicesmessage-queueamazon-sqs
jack miao
jack miao

Reputation: 1488

How to properly send a message to amazon sqs queue with scala?

Iv seen this common-aws on github for how to use it, and this is their example(only for the sender cause this is what i need):

import com.amazonaws.services.sqs.AmazonSQSAsyncClient 
import com.pellucid.wrap.sqs.AmazonSQSScalaClient
import com.mfglabs.commons.aws.sqs._

val sqs = new AmazonSQSScalaClient(new AmazonSQSAsyncClient(), ec)
val builder = SQSStreamBuilder(sqs)

val sender: Flow[String, SendMessageResult, Unit] =
  Flow[String].map { body =>
    val req = new SendMessageRequest()
    req.setMessageBody(body)
    req.setQueueUrl(queueUrl)
    req
  }
  .via(builder.sendMessageAsStream())

but I get some errors and I dont really understand this example, what I need is to create a func that takes a list of case classes, serialise each one of the list to json and send it to a sqs queue...thats it, so this is what i tried so far:

val queueUrl = "the url to my queue"

//the objects here are of case class type ObjectUploadRequest
val listOfObjects = List(Obj1, Obj2, Obj3, Obj4, Obj5)

def pushListToSQS(listOfObjectsRequests: List[ObjectUploadRequest]): Future[SendMessageRequest] = {
  listOfObjectsRequests.map(objReq => {
    val ser = swrite(objReq)
    val sender: Flow[String, SendMessageResult, Unit] =
      Flow[String].map { body =>
        val req = new SendMessageRequest()
        req.setMessageBody(body)
        req.setQueueUrl(queueUrl)
        req
      }.via(builder.sendMessageAsStream())
  })
}

and im getting this error:

enter image description here

would appreciate if someone can help, thanks

Upvotes: 0

Views: 3796

Answers (1)

Eric Eijkelenboom
Eric Eijkelenboom

Reputation: 7021

If you don't mind using the good old AWS Java SDK and its synchronous SQS client, then this works for me: 

import com.amazonaws.services.sqs.AmazonSQSClient
import com.amazonaws.services.sqs.model.SendMessageRequest  

val sqs = new AmazonSQSClient()

listOfObjects.foreach { obj =>
    val json = // convert obj to json

    sqs.sendMessage(new SendMessageRequest()
        .withQueueUrl("your queue url")
        .withMessageBody(json))
}

Upvotes: 1

Related Questions