[AWS IoT Core] AWS IoT Core Message Queuing を試してみました

# Thing作成
aws iot create-thing --thing-name message-queuing-test-device

# 証明書作成
cd certs
aws iot create-keys-and-certificate --set-as-active \
  --certificate-pem-outfile device.pem.crt \
  --private-key-outfile private.pem.key \
  --public-key-outfile public.pem.key

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["iot:Connect"],
      "Resource": "arn:aws:iot:*:*:client/message-queuing-test-*",
      "Condition": {
        "StringLike": {"iot:ClientId": "message-queuing-test-*"}
      }
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish"],
      "Resource": "arn:aws:iot:*:*:topic/test/shared/*"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Subscribe"],
      "Resource": [
        "arn:aws:iot:*:*:topicfilter/test/shared/*",
        "arn:aws:iot:*:*:topicfilter/$share/test-group/test/shared/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Receive"],
      "Resource": "arn:aws:iot:*:*:topic/test/shared/*"
    }
  ]
}

IoTCode/
├── src/
│   ├── certs/               # AWS IoT証明書
│   ├── config.py           # 設定管理
│   ├── publisher.py        # メッセージ送信
│   ├── subscriber.py       # シェアサブスクライバー
│   ├── test_message_queuing.py  # 統合テスト
│   └── .env.example        # 環境変数テンプレート
└── blog/
    └── aws-iot-core-message-queuing-tryout.md

# subscriber.py
client = mqtt.Client(client_id=self.client_id, clean_session=False, protocol=mqtt.MQTTv311)

# publisher.py  
result = self.client.publish(topic, json.dumps(message), qos=1)

# subscriber.py
shared_topic = f"$share/{self.shared_subscription_group}/{self.topic_prefix}/messages"
client.subscribe(shared_topic, qos=1)

# ターミナル1でサブスクライバーを起動
python -m src.subscriber

[Main] サブスクライバー開始中...
[Manager] 3個のサブスクライバーが接続成功
[Subscriber-01] 接続成功: message-queuing-test-subscriber-01, セッション保持: True
[Subscriber-02] 接続成功: message-queuing-test-subscriber-02, セッション保持: True
[Subscriber-03] 接続成功: message-queuing-test-subscriber-03, セッション保持: True

=== メッセージキューイングテスト実行中 ===
Ctrl+C で終了

# ターミナル2でパブリッシャーを起動
python -m src.publisher

=== サブスクライバー統計 ===
  01: 接続中, メッセージ数: 7
  02: 切断中, メッセージ数: 4 
  03: 接続中, メッセージ数: 9
📊 合計受信メッセージ数: 20
⏳ 1個のサブスクライバーが切断中
💡 切断中のサブスクライバー再接続時にキューイングメッセージが配信されます
🔄 Message Queuing機能 動作中！

# ❌ キューイングされない
client.publish(topic, message, qos=0)  

# ✅ キューイング対象
client.publish(topic, message, qos=1)

# ❌ キューイング無効
client = mqtt.Client(clean_session=True)

# ✅ キューイング有効  
client = mqtt.Client(clean_session=False)