如何在Angular中订阅AWS IoT主题？AWS SDK v2问题求助

AWS SDK v2的IoTDataPlaneClient仅支持REST式的IoT操作（比如消息发布），订阅IoT主题需要通过MQTT over WebSocket实现，以下是可行方案：

步骤1：安装依赖 #

使用mqtt.js作为MQTT客户端，它支持WebSocket协议并能与AWS IoT兼容：

npm install mqtt

同时确保已安装AWS相关依赖：

npm install @aws-sdk/credential-provider-cognito-identity @aws-sdk/signature-v4 @aws-crypto/sha256-js @aws-sdk/protocol-http

步骤2：生成签名的WebSocket连接URL #

AWS IoT的WebSocket连接需要基于AWS临时凭证签名，确保请求合法性：

import * as mqtt from 'mqtt';
import { fromCognitoIdentityPool } from '@aws-sdk/credential-provider-cognito-identity';
import { SignatureV4 } from '@aws-sdk/signature-v4';
import { Sha256 } from '@aws-crypto/sha256-js';
import { HttpRequest } from '@aws-sdk/protocol-http';

// 保留你原有的配置
region = "eu-central-1";
identityPoolId = "eu-central-1:83xxxcba-b15f-xxx-bcef-xxxxxxxxx";
iotEndpoint = "wss://axxxxxxdp6-ats.iot.eu-central-1.amazonaws.com";

cognitoCredentials = fromCognitoIdentityPool({
  identityPoolId: this.identityPoolId,
  accountId: "xxxxxxxxxx",
  clientConfig: { region: this.region }
});

mqttClient: mqtt.MqttClient | null = null;

// 生成签名的WebSocket URL
async getSignedWebSocketUrl(): Promise<string> {
  const credentials = await this.cognitoCredentials();
  const signatureV4 = new SignatureV4({
    credentials: {
      accessKeyId: credentials.accessKeyId,
      secretAccessKey: credentials.secretAccessKey,
      sessionToken: credentials.sessionToken
    },
    region: this.region,
    service: 'iotdevicegateway',
    sha256: Sha256
  });

  const url = new URL(this.iotEndpoint.replace('wss://', 'https://'));
  const request = new HttpRequest({
    method: 'GET',
    hostname: url.hostname,
    path: '/mqtt',
    protocol: 'https:',
    query: {
      'X-Amz-Algorithm': 'AWS4-HMAC-SHA256',
      'X-Amz-Credential': `${credentials.accessKeyId}/${signatureV4.getDate()}/${this.region}/iotdevicegateway/aws4_request`,
      'X-Amz-Date': signatureV4.getDate(),
      'X-Amz-SignedHeaders': 'host'
    }
  });

  const signedRequest = await signatureV4.sign(request);
  return `wss://${url.hostname}/mqtt?${new URLSearchParams(signedRequest.query).toString()}`;
}

步骤3：实现订阅逻辑 #

通过mqtt.js连接AWS IoT并订阅主题：

// 订阅指定主题
async subscribeToTopic(topic: string) {
  const wsUrl = await this.getSignedWebSocketUrl();
  this.mqttClient = mqtt.connect(wsUrl, {
    protocol: 'ws',
    clientId: `angular-client-${Math.random().toString(36).substring(2, 10)}`, // 生成唯一客户端ID
    clean: true
  });

  // 连接成功回调
  this.mqttClient.on('connect', () => {
    console.log('已连接到AWS IoT WebSocket');
    this.mqttClient?.subscribe(topic, (err) => {
      err ? console.error('订阅失败:', err) : console.log(`已订阅主题: ${topic}`);
    });
  });

  // 接收消息回调
  this.mqttClient.on('message', (receivedTopic, payload) => {
    console.log(`收到来自${receivedTopic}的消息:`, payload.toString());
    // 在这里处理消息，比如更新组件数据
  });

  // 错误处理
  this.mqttClient.on('error', (err) => {
    console.error('MQTT连接错误:', err);
  });

  // 连接关闭回调
  this.mqttClient.on('close', () => {
    console.log('MQTT连接已关闭');
  });
}

// 组件销毁时断开连接，避免内存泄漏
ngOnDestroy() {
  this.mqttClient?.end();
}

关键权限配置 #

确保Cognito身份池关联的IAM角色拥有以下权限（替换占位符为你的实际信息）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["iot:Connect", "iot:Subscribe", "iot:Receive"],
      "Resource": [
        "arn:aws:iot:eu-central-1:YOUR_ACCOUNT_ID:client/*",
        "arn:aws:iot:eu-central-1:YOUR_ACCOUNT_ID:topicfilter/YOUR_TOPIC"
      ]
    }
  ]
}

内容的提问来源于stack exchange，提问作者Cem Pamir Bana