Azure Event Hubs のクイックスタート・サンプルを軽く試す（Node.js でのお試し）

以下の「Azure Event Hubs」が気になったので、軽く試してみたという話です。

●Azure クイック スタート - Azure portal を使用したイベント ハブの作成 - Azure Event Hubs | Microsoft Docs
　https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-create

内容を理解していく過程というか準備のためにサンプルを触った、という感じになるため、細かなところはまだ分かっていない状況だったりします。

Azure portal での下準備

最初に、上で掲載していた公式ドキュメント「Azure クイック スタート - Azure portal を使用したイベント ハブの作成 - Azure Event Hubs」の中の、以下を進めていきます。

• リソース グループの作成
• Event Hubs 名前空間の作成
• イベント ハブの作成

ページに書いてあるとおりで、特に問題なく進められました。

接続文字列の取得

この後、プログラムを準備したりしていくのですが、その際に必要なキーとなる情報を取得していきます。

これも先ほどと同じように、公式ドキュメント通りで問題なく進められました。
参照先は、以下のページの「ポータルから接続文字列を取得する」という部分になります。

●接続文字列を取得する - Azure Event Hubs - Azure Event Hubs | Microsoft Docs
　https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-get-connection-string#get-connection-string-from-the-portal

手順の中で、上記のドキュメントの記載通り、デフォルトのポリシーが設定されているのが確認できました（以下の画像のとおり）。
ここでは、共有アクセスポリシーを今回用に新規に追加し、その後の手順を進めていくようにしました。

新規追加したものは、デフォルトの全権限を持ったものから、管理権限を取り除いたものになります。

プログラムを書いて動かす

ここからプログラムを準備していきます。

イベントの送信

以下の公式ドキュメントを見つつ、まずはイベントの送信を試します。

●JavaScript を使用して Azure Event Hubs との間でイベントを送受信する (最新) - Azure Event Hubs | Microsoft Docs
　https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-node-get-started-send

ちなみに、上記のページに掲載されていたサンプルをそのまま使いました。

const { EventHubProducerClient } = require("@azure/event-hubs");

const connectionString = "EVENT HUBS NAMESPACE CONNECTION STRING";
const eventHubName = "EVENT HUB NAME";

async function main() {

  // Create a producer client to send messages to the event hub.
  const producer = new EventHubProducerClient(connectionString, eventHubName);

  // Prepare a batch of three events.
  const batch = await producer.createBatch();
  batch.tryAdd({ body: "First event" });
  batch.tryAdd({ body: "Second event" });
  batch.tryAdd({ body: "Third event" });    

  // Send the batch to the event hub.
  await producer.sendBatch(batch);

  // Close the producer client.
  await producer.close();

  console.log("A batch of three events have been sent to the event hub");
}

main().catch((err) => {
  console.log("Error occurred: ", err);
});

ドキュメントに書かれているとおり、以下の 2箇所は書きかえが必要です。

• EVENT HUBS NAMESPACE CONNECTION STRING
• EVENT HUB NAME

EVENT HUBS NAMESPACE CONNECTION STRING のほうは、上の手順の「接続文字列の取得」で取得できる内容です。
EVENT HUB NAME のほうは、上の手順の「Azure portal での下準備」の中で作成した名前を用いることになります。

プログラムを実行後、公式ドキュメントに書かれた以下のような画面が確認できれば OK です。

イベントの受信

イベントの送信を進める際に参照した公式ドキュメントでは、イベントの受信の処理は「Azure Blob Storage」を組み合わせる方法が書いてありました。

その他に、もっとシンプルなサンプルがないかと情報を見ていると、以下の公式サンプルが GitHub上にあったので、これを試していきました。

●azure-sdk-for-js/receiveEvents.js at main · Azure/azure-sdk-for-js
　https://github.com/Azure/azure-sdk-for-js/blob/main/sdk/eventhub/event-hubs/samples/v5/javascript/receiveEvents.js

なお、GitHub のリポジトリにあるサンプルは、以下の公式ドキュメントに書かれた「前回終了したところからメッセージを継続して受信」というのが考慮されていない内容になるようですので、その点はご注意ください。

プログラムの内容は、以下のとおりです。
環境変数を利用していた部分のみ、少し変更を加えています。

const { EventHubConsumerClient, earliestEventPosition } = require("@azure/event-hubs");
  
 const connectionString = `EVENT HUBS NAMESPACE CONNECTION STRING`;
 const eventHubName = `EVENT HUB NAME`;
 const consumerGroup = "";

 async function main() {
   console.log(`Running receiveEvents sample`);
 
   const consumerClient = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName);
 
   const subscription = consumerClient.subscribe(
     {
       processEvents: async (events, context) => {
         for (const event of events) {
           console.log(
             `Received event: '${event.body}' from partition: '${context.partitionId}' and consumer group: '${context.consumerGroup}'`
           );
         }
       },
       processError: async (err, context) => {
         console.log(`Error on partition "${context.partitionId}": ${err}`);
       }
     },
     { startPosition: earliestEventPosition }
   );
 
   // Wait for a bit before cleaning up the sample
   setTimeout(async () => {
     await subscription.close();
     await consumerClient.close();
     console.log(`Exiting receiveEvents sample`);
   }, 30 * 1000);
 }
 
 main().catch((error) => {
   console.error("Error running sample:", error);
 });

送信の処理を行った時と同様、以下の部分は書きかえが必要です。

• EVENT HUBS NAMESPACE CONNECTION STRING
• EVENT HUB NAME

consumerGroup も設定が必要だったかもしれないですが、とりあえず空欄で動作をさせてみました。
その結果、以下のように「イベントの送信」のステップで送っていた内容を、受信して出力することができたようでした。

用語

ひとまず空のままで動かしてみた consumerGroup という内容について、用語を検索して確認してみました。

以下の公式ドキュメントに「イベント コンシューマー > コンシューマー グループ」という項目があり、Event Hubs ストリーム処理のアーキテクチャの図なども載っていました。

●機能の概要 - Azure Event Hubs - Azure Event Hubs | Microsoft Docs
　https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-features#consumer-groups

このあたりの全体像を理解するのには、もう少し調べてみたり等が必要そうです。

おわりに

とりあえず、Azure Event Hubs が気になったので、クイックスタート・サンプルを試していき、メッセージの送受信を行うことができました。

この後は、理解を深めていくために公式ドキュメントをさらに見ていったり、他のサンプルも触ってみたりしようと思います。