AWS Lambda - How to Get Topic Title Coming from AWS IOT
I am testing AWS Lambda with IOS AWS source. My mqtt clients publish different topics: device A publishes data to streaming/A
, device B publishes data to streaming/B
, so in AWS Lambda I defined a SQL rule by selecting all devices originating from topics streaming/+
. The fact is that now I did not have information about the source of the device, because I only Array[Byte]]
have additional information. If anyone has a solution to access the mqtt payload with topic information, I'll take that!
import java.io.{ByteArrayOutputStream, InputStream, OutputStream}
import com.amazonaws.services.lambda.runtime.{Context, RequestStreamHandler}
/**
* Created by alifirat on 24/04/17.
*/
class IOTConsumer extends RequestStreamHandler {
val BUFFER_SIZE = 1024 * 4
override def handleRequest(input: InputStream, output: OutputStream, context: Context): Unit = {
val bytes = toByteArray(input)
val logger= context.getLogger
logger.log("Receive following thing :" + new String(bytes))
output.write(bytes)
}
/**
* Reads and returns the rest of the given input stream as a byte array.
* Caller is responsible for closing the given input stream.
*/
def toByteArray(is : InputStream) : Array[Byte] = {
val output = new ByteArrayOutputStream()
try {
val b = new Array[Byte](BUFFER_SIZE);
var n = 0
var flag = true
while(flag) {
n = is.read(b)
if(n == -1) flag = false
else {
output.write(b, 0, n)
}
}
output.toByteArray();
} finally {
output.close();
Array[Byte]()
}
}
}
source to share
I was looking for the same, there is a way to achieve this. When building your SQL, you use the topic () function to get the topic the message was sent to. So you can put in the attributes section
*, topic() as topic
so your last SQL will look like:
SELECT *, topic() as topic FROM one/of/my/+/topics
your payload will contain a new attribute topic that you can parse in your lambda function.
source to share
If your trigger is SNS messages I would just read the JSON. This will work in Scala:
import com.amazonaws.services.lambda.runtime.events.SNSEvent
import scala.collection.JavaConverters._
object Example extends LambdaApp {
/** Convert Java lists (or nulls!) to Scala lists */
def safeList[A](xs: java.util.List[A]) =
Option(xs).map(_.asScala).getOrElse(List.empty[A])
/** Install the handler in AWS Lambda as `Example::handler`. */
def handler(e: SNSEvent) = {
val rs = for {
r <- safeList(e.getRecords)
} yield {
r.getSNS.getMessage
}
rs.asJava // Convert Scala list to Java.
}
}
In your build.sbt file, you need the following dependencies:
libraryDependencies ++= Seq(
"com.amazonaws" % "aws-lambda-java-core" % "1.1.0",
"com.amazonaws" % "aws-lambda-java-events" % "1.3.0"
)
If you are interested in the SNS theme name, you can get it from:
r.EventSubscriptionArn
AWD Lambda JDK library parses JSON SNS messages for you using jackson-core.
source to share