scala - AWS Lambda - 如何获取来自 AWS IOT 的数据的主题名称

Question

我正在使用 AWS IOT 源测试 AWS Lambda。我的 mqtt 客户端在不同的主题中发布:设备 A 将数据发布到 streaming/A , 设备 B 将数据发布到 streaming/B所以在 AWS Lambda 中，我定义了一个 SQL 规则，选择来自主题 streaming/+ 的所有设备。 .问题是现在我没有设备来源的信息，因为我只有一个Array[Byte]]带有额外的信息。如果有人有访问带有主题信息的 mqtt 负载的解决方案，我会接受!

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]()
     }
   }
}

Answer 1

我一直在寻找同样的东西，有一种方法可以实现这一目标。在构建 SQL 时，您可以使用 topic() 函数来获取消息发送到的主题。这样你就可以放入属性部分

*, topic() as topic

所以你的最终 SQL 将如下所示:

SELECT *, topic() as topic FROM one/of/my/+/topics

然后，您的有效负载将包含一个新的属性主题，您可以在 lambda 函数中解析该主题。更多相关信息 https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html