DynamoDB Backend

    After EMQX version 3.1, a powerful rule engine is introduced to replace plug-ins. It is recommended that you use it. See Save data to DynamoDB to setup Save data to DynamoDB in rule engine.

    Config file: etc/plugins/emqx_backend_dynamo.conf

    Description of DynamoDB Persistence Hooks

    Create DynamoDB DB

    TIP

    DB name is free of choice

    mqtt_client stores client connection states:

    1. {
    2.     "TableName": "mqtt_client",
    3.     "KeySchema": [
    4.         { "AttributeName": "clientid", "KeyType": "HASH" }
    5.     ],
    6.     "AttributeDefinitions": [
    7.         { "AttributeName": "clientid", "AttributeType": "S" }
    8.     ],
    9.     "ProvisionedThroughput": {
    10.         "ReadCapacityUnits": 5,
    11.         "WriteCapacityUnits": 5
    12.     }
    13. }
    1. aws dynamodb scan --table-name mqtt_client --region us-west-2  --endpoint-url http://localhost:8000
    3. {
    4.     "Items": [
    5.         {
    6.             "offline_at": { "N": "0" },
    7.             "node": { "S": "emqx@127.0.0.1" },
    8.             "clientid": { "S": "mqttjs_384b9c73a9" },
    9.             "connect_state": { "N": "1" },
    10.             "online_at": { "N": "1562224940" }
    11.         }
    12.     ],
    13.     "Count": 1,
    14.     "ScannedCount": 1,
    15.     "ConsumedCapacity": null
    16. }

    DynamoDB Subscription Table

    mqtt_sub table stores MQTT subscriptions of clients:

    1. {
    2.     "TableName": "mqtt_sub",
    3.     "KeySchema": [
    4.         { "AttributeName": "clientid", "KeyType": "HASH" },
    5.         { "AttributeName": "topic", "KeyType": "RANGE" }
    6.     ],
    7.     "AttributeDefinitions": [
    8.         { "AttributeName": "clientid", "AttributeType": "S" },
    9.         { "AttributeName": "topic", "AttributeType": "S" }
    10.     "ProvisionedThroughput": {
    11.         "ReadCapacityUnits": 5,
    12.         "WriteCapacityUnits": 5
    13.     }
    14. }

    Query topics subscribed by the client named “test-dynamo”:

    DynamoDB Message Table

    mqtt_msg stores MQTT messages:

    2.     "TableName": "mqtt_msg",
    3.     "KeySchema": [
    4.         { "AttributeName": "msgid", "KeyType": "HASH" }
    5.     ],
    6.     "AttributeDefinitions": [
    7.         { "AttributeName": "msgid", "AttributeType": "S" }
    8.     ],
    9.     "ProvisionedThroughput": {
    10.         "ReadCapacityUnits": 5,
    11.         "WriteCapacityUnits": 5
    12.     }
    13. }

    mqtt_topic_msg_map stores the mapping between topics and messages:

    1. {
    2.     "TableName": "mqtt_topic_msg_map",
    3.     "KeySchema": [
    4.         { "AttributeName": "topic", "KeyType": "HASH" }
    5.     ],
    6.     "AttributeDefinitions": [
    7.         { "AttributeName": "topic", "AttributeType": "S" }
    8.     ],
    9.     "ProvisionedThroughput": {
    10.         "ReadCapacityUnits": 5,
    11.         "WriteCapacityUnits": 5
    12.     }
    13. }

    Query mqtt_msg and mqtt_topic_msg_map after a client publishes a message to the “test” topic:

    1. aws dynamodb scan --table-name mqtt_msg --region us-west-2  --endpoint-url http://localhost:8000
    1. >   - {
    2. >       - "Items": [
    3. >           - {
    4. >             "arrived": { "N": "1562308553" }, "qos": { "N": "1" },
    5. >             "sender": { "S": "mqttjs\_231b962d5c" }, "payload": { "S":
    6. >             "{ "msg": "Hello, World\!" }"}, "retain": { "N": "0" },
    7. >             "msgid": { "S":
    8. >             "Mjg4MTk1MDYwNTk0NjYwNzYzMTg4MDk3OTQ2MDU2Nzg1OTD" },
    9. >             "topic": { "S": "test" }
    10. >         }
    11. >     ], "Count": 1, "ScannedCount": 1, "ConsumedCapacity": null
    12. > }

    Query mqtt_topic_msg_map:

    1. >   - {
    2. >       - "Items": \[
    3. >           - {
    4. >             "topic": { "S": "test" }, "MsgId": { "SS": \[
    5. >         }
    7. > }

    mqtt_retain stores retained messages:

    1. {
    2.     "TableName": "mqtt_retain",
    3.     "KeySchema": [
    4.         { "AttributeName": "topic", "KeyType": "HASH" }
    5.     ],
    6.     "AttributeDefinitions": [
    7.         { "AttributeName": "topic", "AttributeType": "S" }
    8.     ],
    9.     "ProvisionedThroughput": {
    10.         "ReadCapacityUnits": 5,
    11.         "WriteCapacityUnits": 5
    12.     }
    13. }

    Query mqtt_retain after a client publishes a message to the “test” topic:

    1. {
    2.     "Items": [
    3.         {
    4.             "arrived": { "N": "1562312113" },
    5.             "qos": { "N": "1" },
    6.             "sender": { "S": "mqttjs_d0513acfce" },
    7.             "payload": { "S": "test" },
    8.             "retain": { "N": "1" },
    9.             "msgid": { "S": "Mjg4MTk1NzE3MTY4MjYxMjA5MDExMDg0NTk5ODgzMjAyNTH" },
    10.             "topic": { "S": "testtopic" }
    11.         }
    12.     ],
    13.     "Count": 1,
    14.     "ScannedCount": 1,
    15.     "ConsumedCapacity": null
    16. }

    DynamoDB Acknowledgement Table

    mqtt_acked stores acknowledgements from the clients:

    1. {
    2.     "TableName": "mqtt_acked",
    3.     "KeySchema": [
    4.         { "AttributeName": "topic", "KeyType": "HASH" },
    5.         { "AttributeName": "clientid", "KeyType": "RANGE" }
    6.     ],
    7.     "AttributeDefinitions": [
    8.         { "AttributeName": "topic", "AttributeType": "S" },
    9.         { "AttributeName": "clientid", "AttributeType": "S" }
    10.     ],
    11.     "ProvisionedThroughput": {
    12.         "ReadCapacityUnits": 5,
    13.         "WriteCapacityUnits": 5
    14.     }

    Query mqtt_acked after a client publishes a message to the “test” topic:

    Enable DynamoDB Backend

    1. ./bin/emqx_ctl plugins load emqx_backend_dynamo