DynamoDB Backend

TIP

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.

Configure DynamoDB Cluster

Config file: etc/plugins/emqx_backend_dynamo.conf

  1. ## DynamoDB Region
  2. backend.dynamo.region = us-west-2
  4. ## DynamoDB Server
  5. backend.dynamo.pool1.server = http://localhost:8000
  7. ## DynamoDB Pool Size
  8. backend.dynamo.pool1.pool_size = 8
  10. ## AWS ACCESS KEY ID
  11. backend.dynamo.pool1.aws_access_key_id = AKIAU5IM2XOC7AQWG7HK
  13. ## AWS SECRET ACCESS KEY
  14. backend.dynamo.pool1.aws_secret_access_key = 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
  16. ## DynamoDB Backend Hooks
  17. backend.dynamo.hook.client.connected.1    = {"action": {"function": "on_client_connected"}, "pool": "pool1"}
  18. backend.dynamo.hook.session.created.1     = {"action": {"function": "on_subscribe_lookup"}, "pool": "pool1"}
  19. backend.dynamo.hook.client.disconnected.1 = {"action": {"function": "on_client_disconnected"}, "pool": "pool1"}
  20. backend.dynamo.hook.session.subscribed.1  = {"topic": "#", "action": {"function": "on_message_fetch_for_queue"}, "pool": "pool1"}
  21. backend.dynamo.hook.session.subscribed.2  = {"topic": "#", "action": {"function": "on_retain_lookup"}, "pool": "pool1"}
  22. backend.dynamo.hook.session.unsubscribed.1= {"topic": "#", "action": {"function": "on_acked_delete"}, "pool": "pool1"}
  23. backend.dynamo.hook.message.publish.1     = {"topic": "#", "action": {"function": "on_message_publish"}, "pool": "pool1"}
  24. backend.dynamo.hook.message.publish.2     = {"topic": "#", "action": {"function": "on_message_retain"}, "pool": "pool1"}
  25. backend.dynamo.hook.message.publish.3     = {"topic": "#", "action": {"function": "on_retain_delete"}, "pool": "pool1"}
  26. backend.dynamo.hook.message.acked.1       = {"topic": "#", "action": {"function": "on_message_acked_for_queue"}, "pool": "pool1"}
  28. # backend.dynamo.hook.message.publish.4   = {"topic": "#", "action": {"function": "on_message_store"}, "pool": "pool1"}

Description of DynamoDB Persistence Hooks

hooktopicactionDescription
client.connectedon_client_connectedStore client connected state
client.connectedon_subscribe_lookupSubscribed topics
client.disconnectedon_client_disconnectedStore client disconnected state
session.subscribed#on_message_fetch_for_queueFetch offline messages
session.subscribed#on_retain_lookupLookup retained messages
message.publish#on_message_publishStore published messages
message.publish#on_message_retainStore retained messages
message.publish#on_retain_deleteDelete retained messages
message.acked#on_message_acked_for_queueProcess ACK

Create DynamoDB DB

  1. ./test/dynamo_test.sh

TIP

DB name is free of choice

DynamoDB Client Connection Table

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. }

Query the client connection state:

  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.     ],
  11.     "ProvisionedThroughput": {
  12.         "ReadCapacityUnits": 5,
  13.         "WriteCapacityUnits": 5
  14.     }
  15. }

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

  1. aws dynamodb scan --table-name mqtt_sub --region us-west-2  --endpoint-url http://localhost:8000
  3. {
  4.     "Items": [{"qos": { "N": "2" }, "topic": { "S": "test-dynamo-sub" }, "clientid": { "S": "test-dynamo" }},
  5.                {"qos": { "N": "2" }, "topic": { "S": "test-dynamo-sub-1"}, "clientid": { "S": "test-dynamo" }},
  6.                {"qos": { "N": "2" }, "topic": { "S": "test-dynamo-sub-2"}, "clientid": { "S": "test-dynamo" }}],
  7.     "Count": 3,
  8.     "ScannedCount": 3,
  9.     "ConsumedCapacity": null
  10. }

DynamoDB Message Table

mqtt_msg stores MQTT messages:

  1. {
  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:

Query mqtt_msg:

  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. aws dynamodb scan --table-name mqtt_topic_msg_map --region us-west-2  --endpoint-url http://localhost:8000
  1. >   - {
  2. >       - "Items": \[
  3. >           - {
  4. >             "topic": { "S": "test" }, "MsgId": { "SS": \[
  5. >             "Mjg4MTk1MDYwNTk0NjYwNzYzMTg4MDk3OTQ2MDU2Nzg1OTD" \]}
  6. >         }
  7. >     \], "Count": 1, "ScannedCount": 1, "ConsumedCapacity": null
  8. > }

DynamoDB Retained Message Table

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.     }
  15. }

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

  1. {
  2.     "Items": [
  3.         {
  4.             "topic": { "S": "test" },
  5.             "msgid": { "S": "Mjg4MTk1MDYwNTk0NjYwNzYzMTg4MDk3OTQ2MDU2Nzg1OTD" },
  6.             "clientid": { "S": "mqttjs_861e582a70" }
  7.         }
  8.     ],
  9.     "Count": 1,
  10.     "ScannedCount": 1,
  11.     "ConsumedCapacity": null
  12. }

Enable DynamoDB Backend

  1. ./bin/emqx_ctl plugins load emqx_backend_dynamo