AWS Lambda публикации в IOT Тема срабатывает бесконечно

Question

Проблема:

У меня есть функция node.js (8.10) AWS Lambda, которая принимает объект json и публикует его в теме IOT.Функция успешно публикуется в теме, однако после запуска она непрерывно вызывается до тех пор, пока я не уменьшу параллелизм до нуля, чтобы остановить дальнейший вызов функции.

Я пытаюсь выяснить, что я реализовалнеправильно, что вызывает более одного экземпляра функции для вызова.

Функция:

Вот моя функция:

var AWS = require('aws-sdk');

exports.handler = function (event, context) {
    var iotdata = new AWS.IotData({endpoint: 'xxxxxxxxxx.iot.us-east-1.amazonaws.com'});
    var params = {
        topic: '/PiDevTest/SyncDevice',
        payload: JSON.stringify(event),
        qos: 0
    };

    iotdata.publish(params, function(err, data) {
        if (err) {
          console.log(err, err.stack);
        } else {
            console.log("Message sent.");
            context.succeed();
        }     
    });
};

Мой тестовый json:

{
  "success": 1,
  "TccvID": "TestID01"
}

Тестовая консоль имеет ответ "null", но в разделе IOT отображаются данные из тестового json, публикуемые в теме примерно раз в секунду.

То, что я попробовал

-Я попытался определить обработчик в его собственной, неанонимной функции, называемой обработчик, и затем имел export.handler = handler;Это не вызвало никаких ошибок, но также не привело к успешной публикации в теме iot.

-Я подумал, что, возможно, проблемы были с обратным вызовом node.js.Я пытался реализовать это и опустить (текущая итерация выше), но ни один из способов не помог.Я где-то читал, что функция будет повторяться в случае ошибки, но я считаю, что это происходит только три раза, поэтому это не объясняет неопределенный вызов функции.

-Я также пытался вызвать функцию изеще одна лямбда, чтобы убедиться, что проблема не в инструменте тестирования aws.Это привело к тому же самому поведению.

Резюме:

Что я делаю неправильно, что заставляет эту функцию бесконечно публиковать данные json в теме iot?

Заранее спасибо за ваше время и опыт.

Answer 1

В то время как он только отправил это как комментарий, MarkB указал мне в правильном направлении.

Проблема была в том, что решение было связано с другой лямбдой, которая слушала ту же тему и вызывала лямбду, над которой я работална.Это привело к круговой логике, так как условие выхода не было выполнено.Исправление этого кода решило эту проблему.

Answer 2

Используйте aws-iot-device-sdk для создания MQTT-клиента и используйте его messageHandler и метод публикации для публикации ваших сообщений в теме IOT.Пример клиентского кода MQTT приведен ниже:

import * as DeviceSdk from 'aws-iot-device-sdk';
import * as AWS from 'aws-sdk';

let instance: any = null;

export default class IoTClient {

  client: any;
  /**
   * Constructor
   *
   * @params {boolean} createNewClient - Whether or not to use existing client instance
   */
  constructor(createNewClient = false, options = {}) {

  }

  async init(createNewClient, options) {
    if (createNewClient && instance) {
      instance.disconnect();
      instance = null;
    }

    if (instance) {
      return instance;
    }
    instance = this;
    this.initClient(options);
    this.attachDebugHandlers();
  }

  /**
   * Instantiate AWS IoT device object
   * Note that the credentials must be initialized with empty strings;
   * When we successfully authenticate to the Cognito Identity Pool,
   * the credentials will be dynamically updated.
   *
   * @params {Object} options - Options to pass to DeviceSdk
   */
  initClient(options) {
    const clientId = getUniqueId();

    this.client = DeviceSdk.device({
      region: options.region || getConfig('iotRegion'),

      // AWS IoT Host endpoint
      host: options.host || getConfig('iotHost'),

      // clientId created earlier
      clientId: options.clientId || clientId,

      // Connect via secure WebSocket
      protocol: options.protocol || getConfig('iotProtocol'),

      // Set the maximum reconnect time to 500ms; this is a browser application
      // so we don't want to leave the user waiting too long for reconnection after
      // re-connecting to the network/re-opening their laptop/etc...
      baseReconnectTimeMs: options.baseReconnectTimeMs || 500,
      maximumReconnectTimeMs: options.maximumReconnectTimeMs || 1000,

      // Enable console debugging information
      debug: (typeof options.debug === 'undefined') ? true : options.debug,

      // AWS access key ID, secret key and session token must be
      // initialized with empty strings
      accessKeyId: options.accessKeyId,
      secretKey: options.secretKey,
      sessionToken: options.sessionToken,
      // Let redux handle subscriptions
      autoResubscribe: (typeof options.debug === 'undefined') ? false : options.autoResubscribe,
    });
  }

  disconnect() {
    this.client.end();
  }

  attachDebugHandlers() {
    this.client.on('reconnect', () => {
      logger.info('reconnect');
    });

    this.client.on('offline', () => {
      logger.info('offline');
    });

    this.client.on('error', (err) => {
      logger.info('iot client error', err);
    });

    this.client.on('message', (topic, message) => {
      logger.info('new message', topic, JSON.parse(message.toString()));
    });
  }

  updateWebSocketCredentials(accessKeyId, secretAccessKey, sessionToken) {
    this.client.updateWebSocketCredentials(accessKeyId, secretAccessKey, sessionToken);
  }

  attachMessageHandler(onNewMessageHandler) {
    this.client.on('message', onNewMessageHandler);
  }

  attachConnectHandler(onConnectHandler) {
    this.client.on('connect', (connack) => {
      logger.info('connected', connack);
      onConnectHandler(connack);
    });
  }

  attachCloseHandler(onCloseHandler) {
    this.client.on('close', (err) => {
      logger.info('close', err);
      onCloseHandler(err);
    });
  }

  publish(topic, message) {
    this.client.publish(topic, message);
  }

  subscribe(topic) {
    this.client.subscribe(topic);
  }

  unsubscribe(topic) {
    this.client.unsubscribe(topic);
    logger.info('unsubscribed from topic', topic);
  }
}

*** getConfig () предназначен для получения переменных среды из файла yml, иначе вы можете указать его здесь.