Java でプル配信を使用してイベントを受信する
この記事では、Event Grid のプル配信を使用して CloudEvents を受け取る簡単な手順のガイドを提供します。 受信、確認 (Event Grid からのイベント削除) を行うためのサンプル コードを提供します。
前提条件
続行する前に必要な前提条件は次のとおりです。
名前空間、トピック、およびイベント サブスクリプション。
- 名前空間を作成して管理する
- 名前空間トピックを作成して管理する
- イベント サブスクリプションを作成して管理する
最新の ベータ版 SDK パッケージ。 maven を使用している場合は、maven の中央リポジトリを参照できます。
重要
プル配信データ プレーン SDK のサポートは、ベータ パッケージで利用できます。 プロジェクトでは最新のベータ パッケージを使用する必要があります。
Java をサポートする IDE (IntelliJ IDEA、Eclipse IDE、Visual Studio Code など)。
Java 8 言語レベルを実行している Java JRE。
トピックで使用可能なイベントが必要です。 名前空間へのイベントの発行に関するトピックを参照してください。
プル配信を使用してイベントを受信する
Event Grid からイベントを読み取る場合は、受信操作で名前空間トピックとキュー イベント サブスクリプションを指定します。 イベント サブスクリプションは、コンシューマー クライアントが読み取ることができる CloudEvents のコレクションを効果的に定義するリソースです。 このサンプル コードでは、キーベースの認証を使用します。これにより、すばやく簡単に認証する方法が得られます。 運用環境のシナリオでは、Microsoft Entry ID 認証を使用する必要があります。これは、より堅牢な認証メカニズムを提供するからです。
package com.azure.messaging.eventgrid.samples;
import com.azure.core.credential.AzureKeyCredential;
import com.azure.core.http.HttpClient;
import com.azure.core.models.CloudEvent;
import com.azure.messaging.eventgrid.EventGridClient;
import com.azure.messaging.eventgrid.EventGridClientBuilder;
import com.azure.messaging.eventgrid.EventGridMessagingServiceVersion;
import com.azure.messaging.eventgrid.models.*;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
/**
* <p>Simple demo consumer app of CloudEvents from queue event subscriptions created for namespace topics.
* This code samples should use Java 1.8 level or above to avoid compilation errors.
* You should consult the resources below to use the client SDK and set up your project using maven.
* @see <a href="https://github.com/Azure/azure-sdk-for-java/tree/main/sdk/eventgrid/azure-messaging-eventgrid">Event Grid data plane client SDK documentation</a>
* @see <a href="https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/boms/azure-sdk-bom/README.md">Azure BOM for client libraries</a>
* @see <a href="https://aka.ms/spring/versions">Spring Version Mapping</a> if you are using Spring.
* @see <a href="https://aka.ms/azsdk">Tool with links to control plane and data plane SDKs across all languages supported</a>.
*</p>
*/
public class NamespaceTopicConsumer {
private static final String TOPIC_NAME = "<yourNamespaceTopicName>";
public static final String EVENT_SUBSCRIPTION_NAME = "<yourEventSusbcriptionName>";
public static final String ENDPOINT = "<yourFullHttpsUrlToTheNamespaceEndpoint>";
public static final int MAX_NUMBER_OF_EVENTS_TO_RECEIVE = 10;
public static final Duration MAX_WAIT_TIME_FOR_EVENTS = Duration.ofSeconds(10);
private static EventGridClient eventGridClient;
private static List<String> receivedCloudEventLockTokens = new ArrayList<>();
private static List<CloudEvent> receivedCloudEvents = new ArrayList<>();
//TODO Do NOT include keys in source code. This code's objective is to give you a succinct sample about using Event Grid, not to provide an authoritative example for handling secrets in applications.
/**
* For security concerns, you should not have keys or any other secret in any part of the application code.
* You should use services like Azure Key Vault for managing your keys.
*/
public static final AzureKeyCredential CREDENTIAL = new AzureKeyCredential("<namespace key>");
public static void main(String[] args) {
//TODO Update Event Grid version number to your desired version. You can find more information on data plane APIs here:
//https://learn.microsoft.com/en-us/rest/api/eventgrid/.
eventGridClient = new EventGridClientBuilder()
.httpClient(HttpClient.createDefault()) // Requires Java 1.8 level
.endpoint(ENDPOINT)
.serviceVersion(EventGridMessagingServiceVersion.V2023_06_01_PREVIEW)
.credential(CREDENTIAL).buildClient(); // you may want to use .buildAsyncClient() for an asynchronous (project reactor) client.
System.out.println("Waiting " + MAX_WAIT_TIME_FOR_EVENTS.toSecondsPart() + " seconds for events to be read...");
List<ReceiveDetails> receiveDetails = eventGridClient.receiveCloudEvents(TOPIC_NAME, EVENT_SUBSCRIPTION_NAME,
MAX_NUMBER_OF_EVENTS_TO_RECEIVE, MAX_WAIT_TIME_FOR_EVENTS).getValue();
for (ReceiveDetails detail : receiveDetails) {
// Add order message received to a tracking list
CloudEvent orderCloudEvent = detail.getEvent();
receivedCloudEvents.add(orderCloudEvent);
// Add lock token to a tracking list. Lock token functions like an identifier to a cloudEvent
BrokerProperties metadataForCloudEventReceived = detail.getBrokerProperties();
String lockToken = metadataForCloudEventReceived.getLockToken();
receivedCloudEventLockTokens.add(lockToken);
}
System.out.println("<-- Number of events received: " + receivedCloudEvents.size());
イベントを確認する
イベントを確認するには、イベントの受信に使用されるものと同じコードを使用し、次の行を追加して確認プライベート メソッドを呼び出します。
// Acknowledge (i.e. delete from Event Grid the) events
acknowledge(receivedCloudEventLockTokens);
失敗したロック トークンに関する情報を出力する確認メソッド (およびユーティリティ メソッド) の実装例を次に示します。
private static void acknowledge(List<String> lockTokens) {
AcknowledgeResult acknowledgeResult = eventGridClient.acknowledgeCloudEvents(TOPIC_NAME, EVENT_SUBSCRIPTION_NAME, new AcknowledgeOptions(lockTokens));
List<String> succeededLockTokens = acknowledgeResult.getSucceededLockTokens();
if (succeededLockTokens != null && lockTokens.size() >= 1)
System.out.println("@@@ " + succeededLockTokens.size() + " events were successfully acknowledged:");
for (String lockToken : succeededLockTokens) {
System.out.println(" Acknowledged event lock token: " + lockToken);
}
// Print the information about failed lock tokens
if (succeededLockTokens.size() < lockTokens.size()) {
System.out.println(" At least one event was not acknowledged (deleted from Event Grid)");
writeFailedLockTokens(acknowledgeResult.getFailedLockTokens());
}
}
private static void writeFailedLockTokens(List<FailedLockToken> failedLockTokens) {
for (FailedLockToken failedLockToken : failedLockTokens) {
System.out.println(" Failed lock token: " + failedLockToken.getLockToken());
System.out.println(" Error code: " + failedLockToken.getErrorCode());
System.out.println(" Error description: " + failedLockToken.getErrorDescription());
}
}
イベントを開放する
イベントを解放して、再配信できるようにします。 イベントを確認するために行ったのと同様に、次の静的メソッドとそれを呼び出す行を追加して、引数として渡されたロック トークンによって識別されるイベントを解放できます。 このメソッドをコンパイルするには、writeFailedLockTokens メソッドが必要です。
private static void release(List<String> lockTokens) {
ReleaseResult releaseResult = eventGridClient.releaseCloudEvents(TOPIC_NAME, EVENT_SUBSCRIPTION_NAME, new ReleaseOptions(lockTokens));
List<String> succeededLockTokens = releaseResult.getSucceededLockTokens();
if (succeededLockTokens != null && lockTokens.size() >= 1)
System.out.println("^^^ " + succeededLockTokens.size() + " events were successfully released:");
for (String lockToken : succeededLockTokens) {
System.out.println(" Released event lock token: " + lockToken);
}
// Print the information about failed lock tokens
if (succeededLockTokens.size() < lockTokens.size()) {
System.out.println(" At least one event was not released back to Event Grid.");
writeFailedLockTokens(releaseResult.getFailedLockTokens());
}
}
イベントを拒否する
コンシューマー アプリケーションで処理できないイベントを拒否します。 イベントを拒否する条件には、形式に誤りがあり解析できないイベントや、イベントを処理するアプリケーションに関する問題が含まれます。
private static void reject(List<String> lockTokens) {
RejectResult rejectResult = eventGridClient.rejectCloudEvents(TOPIC_NAME, EVENT_SUBSCRIPTION_NAME, new RejectOptions(lockTokens));
List<String> succeededLockTokens = rejectResult.getSucceededLockTokens();
if (succeededLockTokens != null && lockTokens.size() >= 1)
System.out.println("--- " + succeededLockTokens.size() + " events were successfully rejected:");
for (String lockToken : succeededLockTokens) {
System.out.println(" Rejected event lock token: " + lockToken);
}
// Print the information about failed lock tokens
if (succeededLockTokens.size() < lockTokens.size()) {
System.out.println(" At least one event was not rejected.");
writeFailedLockTokens(rejectResult.getFailedLockTokens());
}
}
次のステップ
- Java API リファレンスを参照してください。
- プル配信モデルの詳細については、プル配信の概要に関するページを参照してください。