EventBridge Pipes のソースとしての Amazon MQ メッセージブローカー。

EventBridge Pipes を使用して、Amazon MQ メッセージブローカーからレコードを受け取ることができます。その後、オプションでこれらのレコードをフィルタリングまたは拡張してから、処理可能な送信先のいずれかに送信できます。Amazon MQ に固有の設定があり、パイプをセットアップするときに選択できます。EventBridge Pipes は、データを送信先に送信するときに、メッセージブローカーのレコードの順序を維持します。

Amazon MQ は、Apache ActiveMQ および RabbitMQ 用のマネージドメッセージブローカーサービスです。メッセージブローカーを使用すると、ソフトウェアアプリケーションおよびコンポーネントは、さまざまなプログラミング言語、オペレーティングシステム、および、トピックまたはキューをイベント送信先とする正式なメッセージングプロトコルを使って、通信できるようになります。

また Amazon MQ は、ActiveMQ か RabbitMQ ブローカーをインストールすることにより、ユーザーに代わって Amazon Elastic Compute Cloud (Amazon EC2)インスタンスを管理することもできます。ブローカーをインストールすると、さまざまなネットワークトポロジやその他のインフラストラクチャのニーズがインスタンスに提供されます。

Amazon MQ ソースには、次の設定制限があります。

EventBridge は、Amazon MQ がサポートする ActiveMQ および RabbitMQ の最新バージョンを自動的にサポートします。サポートされている最新バージョンについては、「Amazon MQ デベロッパーガイド」の「Amazon MQ リリースノート」を参照してください。

イベントの例

次のサンプルイベントは、パイプが受信した情報を示しています。このイベントを使用して、イベントパターンを作成およびフィルタリングしたり、入力変換を定義したりできます。すべてのフィールドをフィルタリングできるわけではありません。フィルターできるフィールドの詳細については、「Amazon EventBridge Pipes フィルタリング」を参照してください。

ActiveMQ

[
  {
    "eventSource": "aws:amq",
    "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",    
    "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
    "messageType": "jms/text-message",
    "data": "QUJDOkFBQUE=",
    "connectionId": "myJMSCoID",
    "redelivered": false,
    "destination": {
      "physicalname": "testQueue"
    },
    "timestamp": 1598827811958,
    "brokerInTime": 1598827811958,
    "brokerOutTime": 1598827811959
  },
  {
    "eventSource": "aws:amq",
    "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",        
    "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
    "messageType": "jms/bytes-message",
    "data": "3DTOOW7crj51prgVLQaGQ82S48k=",
    "connectionId": "myJMSCoID1",
    "persistent": false,
    "destination": {
      "physicalname": "testQueue"
    },
    "timestamp": 1598827811958,
    "brokerInTime": 1598827811958,
    "brokerOutTime": 1598827811959
  }
]

RabbitMQ

[
  {
    "eventSource": "aws:rmq",
    "eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
    "eventSourceKey": "pizzaQueue::/",
    "basicProperties": {
      "contentType": "text/plain",
      "contentEncoding": null,
      "headers": {
        "header1": {
          "bytes": [
            118,
            97,
            108,
            117,
            101,
            49
          ]
        },
        "header2": {
          "bytes": [
            118,
            97,
            108,
            117,
            101,
            50
          ]
        },
        "numberInHeader": 10
      },
      "deliveryMode": 1,
      "priority": 34,
      "correlationId": null,
      "replyTo": null,
      "expiration": "60000",
      "messageId": null,
      "timestamp": "Jan 1, 1970, 12:33:41 AM",
      "type": null,
      "userId": "AIDACKCEVSQ6C2EXAMPLE",
      "appId": null,
      "clusterId": null,
      "bodySize": 80
    },
    "redelivered": false,
    "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
  }
]

コンシューマーグループ

Amazon MQ と対話するため、EventBridge は、Amazon MQ ブローカーから読み取ることができるコンシューマーグループを作成します。コンシューマーグループは、パイプ UUID と同じ ID で作成されます。

Amazon MQ トソースの場合、EventBridge はレコードをまとめてバッチ処理し、それらを単一のペイロードで関数に送信します。動作を制御するには、バッチ処理ウィンドウとバッチサイズを設定できます。EventBridge は、以下のいずれかが発生すると、メッセージをプルします。

EventBridge はバッチを単一のペイロードに変換してから、関数を呼び出します。メッセージは永続化も逆シリアル化もされません。その代わりに、コンシューマーグループはそれらをバイトの BLOB として取得します。次に、base64 でエンコードして JSON ペイロードにします。パイプがバッチ内のいずれかのメッセージに対してエラーを返すとEventBridge は、処理が成功するかメッセージが期限切れになるまでメッセージのバッチ全体を再試行します。

ネットワーク構成

デフォルトではAmazon MQ、 ブローカーは PubliclyAccessible フラグを false に設定して作成されます。ブローカーにパブリック IP アドレスが与えられるのは、PubliclyAccessible が true に設定されている場合のみです。パイプでフルアクセスする場合、ブローカーはパブリックエンドポイントを使用するか、VPC へアクセスを提供する必要があります。

Amazon MQ ブローカーがパブリックにアクセス可能ではない場合、EventBridge は、Amazon MSK クラスターに関連付けられている Amazon Virtual Private Cloud (Amazon VPC) リソースにアクセスできる必要があります。

Amazon VPC セキュリティグループは、少なくとも以下のルールを使用して設定してください。