[Amazon Lex] MQTTで通信するLexクライアントを作ってみました。

1 はじめに

AIソリューション部の平内(SIN)です。

今回は、Amazon Lex(以下、Lex)を、AWS IoT のMQTTで使用してみました。

最初に、実際に利用しているようすをご確認ください。時間が空いて、Lambdaのインスタンスが無い場合、一回目のレスポンスだけ一呼吸遅れますが、それ以外は、殆ど違和感なく使えそうです。

2 構成

構成は、以下のようになっています。

  1. クライアントへは、CognitoPoolIDでAWSIoTの特定のトピックへのPublishSubscribeの権限を渡しています
  2. 送信メッセージは、MQTTへのPublishの形で行います
  3. AWS IoTのルールでヒットしたメッセージは、Lambdaに送られます
  4. Lambdaは、送られてきたメッセージを入力としてLexとのやり取りします
  5. Lexからのレスポンスは、MQTTPublishします
  6. トピックをSubscribeしているクライアントは、メッセージが到着すると、それを表示しています

3 Cognito

Cognitoで発行されるPoolIdには、特定のトピック(Sample_Topic)でのSubscribePublishのパーミッションが付与されています。(ここでは、clientを省略していますが、名前で絞ると、より厳格になります)

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "iot:Connect",
                "iot:Receive",
                "iot:Subscribe",
                "iot:Publish"
            ],
            "Resource": [
                "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:client/*",
                "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topic/Sample_Topic",
                "arn:aws:iot:ap-northeast-1:xxxxxxxxxxxx:topicfilter/Sample_Topic"
            ]
        }
    ]
}

4 クライアントの実装

(1) メッセージ

やり取りするメッセージの形式を下記のように定めました。送信も受信も同じトピックで行っているので、自分が送信したメッセージも受信してしまいます。

そこで、typeタグで、送信・受信を表現しています。

const param = {
    type: 'req',  // req or res
    body: 'message',
    userId: 'xxxxxx'
}

(2) 実装

クライアントは、とりあえず、html(JavaScript)で作成しています。

  1. 最初に、トピックをSubscribeしています。(Mqttクラスのinit()で実装)
  2. メッセージの入力があった場合、MQTTpublishしています。(sendMessage())
  3. メッセージが到着した場合は、その内容を、表示しています。(iot.onRecive())
  4. 表示する際の送信・受信メッセージの区別は、typeタグで行われています。(appendLog(param.body, param.type))
<script>
    $('#message').focus();
    const userId = 'id-' + Date.now(); // ブラウザの更新時に初期化する
    const PoolId = 'ap-northeast-1:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx';
    const region = 'ap-northeast-1';
    const endpoint = 'xxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com';
    const topic = "Sample_Topic";
    const clientId = "sample_id";

    const iot = new Mqtt();
    iot.init(PoolId, region, endpoint, topic, clientId);
    iot.onRecive = async data => { 
        const param = JSON.parse(data);
        appendLog(param.body, param.type); 
    }
    function sendMessage() {
        const message = $('#message').val().trim();
        if(message.length > 0) {
            $('#message').val('');
            iot.Send({ type: 'req', userId: userId, body: message });
        }
        return false;
    }
    function appendLog(message, className) {
      $('<log>', { class:className, text:message }).appendTo('#logView');
      $('#logView').scrollTop(self.innerHeight);
    }

  </script>

すべてのコードは、下記に置きました。
furuya02/MQTT_LexClient_index.html
furuya02/MQTT_LexClient_mqtt.js

(3) テスト

AWSIoTのコンソール(テスト)で動作確認している様子です。

送信

受信

5 ルール

AWS IoTに到着したMQTTのメッセージは、ルールによりLambdaへ送られます。ここでは、トピック名(Sample_Toipc)のメッセージを、すべて送っています。

6 Lambda

AWS IoTからキックされるLambdaでは、そのメッセージをLexに送信し、そのレスポンスをMQTTPublishしています。

const aws = require('aws-sdk');

exports.handler = async (event) => {
    if(event.type != 'req') {
        return;
    }
    const message = await lexClient(event.body, event.userId);
    await sendMqtt(message);
};

async function lexClient(message, userId) {
    // 下記で紹介します(7 Lex CLient)
}

async function sendMqtt(message) {
    // 下記で紹介します(8 Publish)
}

すべてのコードは、下記に置きました。
furuya02/MQTT_LexClient_index.js

7 Lex CLient

LambdaからpostText()を使用してLexへアクセスするコードは、以下のとおりです。 Lexにアクセスするためのポリシーを追加する必要があります。

async function lexClient(message, userId) {
    const botAlias = '$LATEST';
    const botName = 'OrderFlowers';
    const sessionAttributes = {};

    const lexruntime = new aws.LexRuntime( {region: 'us-east-1'});

    const params = {
        botAlias: botAlias,
        botName: botName,
        inputText: message,
        userId: userId,
        sessionAttributes: sessionAttributes
    };

    const data = await lexruntime.postText(params).promise();
    console.log(JSON.stringify(data));
    return data.message;
}

8 Publish

LambdaからMQTTPublishするコードは、以下のとおりです。 AWS IoTへPublishできるポリシーを追加する必要があります。

async function sendMqtt(message) {

    const endpoint = 'xxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com';
    const topic = "Sample_Topic"
    const region = 'ap-northeast-1';

    var iotdata = new aws.IotData({
        endpoint: endpoint,
        region: region
    });

    const param = {
        type: 'res',
        body: message
    }

    var iotParams = {
        topic: topic,
        payload: JSON.stringify(param),
        qos: 0
    };

    let result = await iotdata.publish(iotParams).promise();
}

9 セッション

MQTTでは、1つ1つが独立したメッセージですが、Lexでは、セッションの管理が必要です。

Lexでは、userIdでセッションを区別しているため、今回は、クライアントのJavaScriptでブラウザを 更新した際に、新たにuserIdを採番するように実装しました。

const userId = 'id-' + Date.now(); // ブラウザの更新時に初期化する

10 最後に

最初に、紹介したとおり、時間が空いて、Lambdaのインスタンスがいない場合、一回目のレスポンスだけ一呼吸遅れますが、それ以外は、殆ど違和感なく使えそうです。

これで、クライアントにLexへのパーミッションを付与しなくても、Lambda側の制御で実装が可能になります。

11 参考リンク


[Amazon Lex] HTML+JavaScriptでLexクライアントを作ってみました
[AWS IoT] MQTTを使用して、Lambdaからブラウザを更新する方法〜aws-iot-device-sdk(aws-iot-sdk-browser-bundle.js)を使用する場合〜

コメントは受け付けていません。