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](https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-create)」の中の、以下を進めていきます。

- リソース グループの作成
- 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

手順の中で、上記のドキュメントの記載通り、デフォルトのポリシーが設定されているのが確認できました（以下の画像のとおり）。
ここでは、共有アクセスポリシーを今回用に新規に追加し、その後の手順を進めていくようにしました。
![共有アクセスポリシー.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/c1a999bb-6306-af2c-00ef-f41f302702e4.jpeg)
新規追加したものは、デフォルトの全権限を持ったものから、管理権限を取り除いたものになります。

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

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

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

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

```javascript
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 です。
![結果の確認.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/a9c45b55-31f5-851a-f217-737a354e2c68.jpeg)

### イベントの受信
[イベントの送信を進める際に参照した公式ドキュメント](https://docs.microsoft.com/ja-jp/azure/event-hubs/event-hubs-node-get-started-send)では、イベントの受信の処理は「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 のリポジトリにあるサンプルは、以下の公式ドキュメントに書かれた「前回終了したところからメッセージを継続して受信」というのが考慮されていない内容になるようですので、その点はご注意ください。
![イベントの受信について](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/3fefe30d-ef45-06d4-92b1-851f6a082ea1.jpeg)

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

```javascript
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` も設定が必要だったかもしれないですが、とりあえず空欄で動作をさせてみました。
その結果、以下のように「イベントの送信」のステップで送っていた内容を、受信して出力することができたようでした。
![受信の動作確認.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/958ee553-ff23-61dc-4172-acd580005cc6.jpeg)

### 用語
ひとまず空のままで動かしてみた `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
![Event Hubs ストリーム処理のアーキテクチャ](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/50868/e9b545fc-cb55-3804-68d7-c71e7f93068d.jpeg)

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

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

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