diff --git a/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md b/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md index d0a55f1456a3..a810b0506f3c 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/README.md @@ -69,6 +69,7 @@ sequence number and the timestamp of when it was enqueued. ### Create an instance of Storage container with SAS token + ```java BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() .connectionString("") @@ -91,32 +92,35 @@ In our example, we will focus on building the [`EventProcessor`][source_eventpro [`BlobCheckpointStore`][source_blobcheckpointstore], and a simple callback function to process the events received from the Event Hubs, writes to console and updates the checkpoint in Blob storage after each event. + ```java -class Program { - public static void main(String[] args) { - EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() - .consumerGroup("<< CONSUMER GROUP NAME >>") - .connectionString("<< EVENT HUB CONNECTION STRING >>") - .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient)) - .processEvent(eventContext -> { - System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " - + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); - }) - .processError(errorContext -> { - System.out.println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); - }) - .buildEventProcessorClient(); - - // This will start the processor. It will start processing events from all partitions. - eventProcessorClient.start(); - - // (for demo purposes only - adding sleep to wait for receiving events) - TimeUnit.SECONDS.sleep(2); - - // When the user wishes to stop processing events, they can call `stop()`. - eventProcessorClient.stop(); - } -} +BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() + .connectionString("") + .containerName("") + .sasToken("") + .buildAsyncClient(); + +EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() + .consumerGroup("<< CONSUMER GROUP NAME >>") + .connectionString("<< EVENT HUB CONNECTION STRING >>") + .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient)) + .processEvent(eventContext -> { + System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " + + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); + }) + .processError(errorContext -> { + System.out.println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); + }) + .buildEventProcessorClient(); + +// This will start the processor. It will start processing events from all partitions. +eventProcessorClient.start(); + +// (for demo purposes only - adding sleep to wait for receiving events) +TimeUnit.SECONDS.sleep(2); + +// When the user wishes to stop processing events, they can call `stop()`. +eventProcessorClient.stop(); ``` ## Troubleshooting diff --git a/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/blob/ReadmeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/blob/ReadmeSamples.java new file mode 100644 index 000000000000..8f5d15d5ce63 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs-checkpointstore-blob/src/samples/java/com/azure/messaging/eventhubs/checkpointstore/blob/ReadmeSamples.java @@ -0,0 +1,65 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.eventhubs.checkpointstore.blob; + +import com.azure.messaging.eventhubs.EventProcessorClient; +import com.azure.messaging.eventhubs.EventProcessorClientBuilder; +import com.azure.storage.blob.BlobContainerAsyncClient; +import com.azure.storage.blob.BlobContainerClientBuilder; +import java.util.concurrent.TimeUnit; + +/** + * WARNING: MODIFYING THIS FILE WILL REQUIRE CORRESPONDING UPDATES TO README.md FILE. LINE NUMBERS ARE USED TO EXTRACT + * APPROPRIATE CODE SEGMENTS FROM THIS FILE. ADD NEW CODE AT THE BOTTOM TO AVOID CHANGING LINE NUMBERS OF EXISTING CODE + * SAMPLES. + * + * Class containing code snippets that will be injected to README.md. + */ +public class ReadmeSamples { + + /** + * Code sample for creating an async blob container client. + */ + public void createBlobContainerClient() { + BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() + .connectionString("") + .containerName("") + .sasToken("") + .buildAsyncClient(); + } + + /** + * Code sample for consuming events from event processor with blob checkpoint store. + * @throws InterruptedException If the thread is interrupted. + */ + public void consumeEventsUsingEventProcessor() throws InterruptedException { + BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder() + .connectionString("") + .containerName("") + .sasToken("") + .buildAsyncClient(); + + EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() + .consumerGroup("<< CONSUMER GROUP NAME >>") + .connectionString("<< EVENT HUB CONNECTION STRING >>") + .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient)) + .processEvent(eventContext -> { + System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " + + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); + }) + .processError(errorContext -> { + System.out.println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); + }) + .buildEventProcessorClient(); + + // This will start the processor. It will start processing events from all partitions. + eventProcessorClient.start(); + + // (for demo purposes only - adding sleep to wait for receiving events) + TimeUnit.SECONDS.sleep(2); + + // When the user wishes to stop processing events, they can call `stop()`. + eventProcessorClient.stop(); + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 8849b6cdb51c..1c87bb054a52 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -90,6 +90,7 @@ consumers. The snippet below creates a synchronous Event Hub producer. + ```java String connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>"; String eventHubName = "<< NAME OF THE EVENT HUB >>"; @@ -123,6 +124,7 @@ Follow the instructions in [Creating a service principal using Azure Portal][app service principal and a client secret. The corresponding `clientId` and `tenantId` for the service principal can be obtained from the [App registration page][app_registration_page]. + ```java ClientSecretCredential credential = new ClientSecretCredentialBuilder() .clientId("<< APPLICATION (CLIENT) ID >>") @@ -188,6 +190,7 @@ Event Hubs service to hash the events and send them to the same partition. The snippet below creates a synchronous producer and sends events to any partition, allowing Event Hubs service to route the event to an available partition. + ```java EventHubProducerClient producer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") @@ -208,7 +211,6 @@ for (EventData eventData : allEvents) { } } } - // send the last batch of remaining events if (eventDataBatch.getCount() > 0) { producer.send(eventDataBatch); @@ -226,6 +228,7 @@ Hub, their names are assigned at the time of creation. To understand what partit using `EventHubsClientBuilder` can query for metadata about the Event Hub using `getPartitionIds()` or `getEventHubProperties()`. + ```java EventHubProducerClient producer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") @@ -244,7 +247,12 @@ When an Event Hub producer is not associated with any specific partition, it may Hubs service keep different events or batches of events together on the same partition. This can be accomplished by setting a `partition key` when publishing the events. + ```java +EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + CreateBatchOptions batchOptions = new CreateBatchOptions().setPartitionKey("grouping-key"); EventDataBatch eventDataBatch = producer.createBatch(batchOptions); @@ -266,6 +274,7 @@ to newest events that get pushed to the partition by invoking `receiveFromPartit can begin receiving events from multiple partitions using the same EventHubConsumerAsyncClient by calling `receiveFromPartition(String, EventPosition)` with another partition id, and subscribing to that Flux. + ```java EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") @@ -283,6 +292,7 @@ consumer.receiveFromPartition("0", EventPosition.latest()).subscribe(event -> { Developers can create a synchronous consumer that returns events in batches using an `EventHubConsumerClient`. In the snippet below, a consumer is created that starts reading events from the beginning of the partition's event stream. + ```java EventHubConsumerClient consumer = new EventHubClientBuilder() .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") @@ -313,32 +323,30 @@ In our example, we will focus on building the [`EventProcessorClient`][EventProc [`InMemoryCheckpointStore`][InMemoryCheckpointStore] available in samples, and a callback function that processes events received from the Event Hub and writes to console. + ```java -class Program { - public static void main(String[] args) { - EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() - .consumerGroup("<< CONSUMER GROUP NAME >>") - .connectionString("<< EVENT HUB CONNECTION STRING >>") - .checkpointStore(new InMemoryCheckpointStore()) - .processEvent(eventContext -> { - System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " - + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); - }) - .processError(errorContext -> { - System.out.println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); - }) - .buildEventProcessorClient(); - - // This will start the processor. It will start processing events from all partitions. - eventProcessorClient.start(); - - // (for demo purposes only - adding sleep to wait for receiving events) - TimeUnit.SECONDS.sleep(2); - - // When the user wishes to stop processing events, they can call `stop()`. - eventProcessorClient.stop(); - } -} +EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() + .consumerGroup("<< CONSUMER GROUP NAME >>") + .connectionString("<< EVENT HUB CONNECTION STRING >>") + .checkpointStore(new InMemoryCheckpointStore()) + .processEvent(eventContext -> { + System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " + + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); + }) + .processError(errorContext -> { + System.out + .println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); + }) + .buildEventProcessorClient(); + +// This will start the processor. It will start processing events from all partitions. +eventProcessorClient.start(); + +// (for demo purposes only - adding sleep to wait for receiving events) +TimeUnit.SECONDS.sleep(2); + +// This will stop processing events. +eventProcessorClient.stop(); ``` ## Troubleshooting diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ReadmeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ReadmeSamples.java new file mode 100644 index 000000000000..be2d68950331 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ReadmeSamples.java @@ -0,0 +1,179 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.eventhubs; + +import com.azure.core.util.IterableStream; +import com.azure.identity.ClientSecretCredential; +import com.azure.identity.ClientSecretCredentialBuilder; +import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import com.azure.messaging.eventhubs.models.EventPosition; +import com.azure.messaging.eventhubs.models.PartitionEvent; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * WARNING: MODIFYING THIS FILE WILL REQUIRE CORRESPONDING UPDATES TO README.md FILE. LINE NUMBERS ARE USED TO EXTRACT + * APPROPRIATE CODE SEGMENTS FROM THIS FILE. ADD NEW CODE AT THE BOTTOM TO AVOID CHANGING LINE NUMBERS OF EXISTING CODE + * SAMPLES. + * + * Class containing code snippets that will be injected to README.md. + */ +public class ReadmeSamples { + + /** + * Code sample for creating a synchronous Event Hub producer. + */ + public void createSynchronousEventHubProducer() { + String connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>"; + String eventHubName = "<< NAME OF THE EVENT HUB >>"; + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString(connectionString, eventHubName) + .buildProducerClient(); + } + + /** + * Code sample for using AAD authorization to create a client. + */ + public void useAadAuthorization() { + ClientSecretCredential credential = new ClientSecretCredentialBuilder() + .clientId("<< APPLICATION (CLIENT) ID >>") + .clientSecret("<< APPLICATION SECRET >>") + .tenantId("<< DIRECTORY (TENANT) ID >>") + .build(); + + // The fully qualified namespace for the Event Hubs instance. This is likely to be similar to: + // {your-namespace}.servicebus.windows.net + String fullyQualifiedNamespace = "my-test-eventhubs.servicebus.windows.net"; + String eventHubName = "<< NAME OF THE EVENT HUB >>"; + EventHubProducerClient client = new EventHubClientBuilder() + .credential(fullyQualifiedNamespace, eventHubName, credential) + .buildProducerClient(); + } + + /** + * Code sample for publishing events. + * @throws IllegalArgumentException if the event data is bigger than max batch size. + */ + public void publishEvents() { + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + + List allEvents = Arrays.asList(new EventData("Foo"), new EventData("Bar")); + EventDataBatch eventDataBatch = producer.createBatch(); + + for (EventData eventData : allEvents) { + if (!eventDataBatch.tryAdd(eventData)) { + producer.send(eventDataBatch); + eventDataBatch = producer.createBatch(); + + // Try to add that event that couldn't fit before. + if (!eventDataBatch.tryAdd(eventData)) { + throw new IllegalArgumentException("Event is too large for an empty batch. Max size: " + + eventDataBatch.getMaxSizeInBytes()); + } + } + } + // send the last batch of remaining events + if (eventDataBatch.getCount() > 0) { + producer.send(eventDataBatch); + } + } + + /** + * Code sample for publishing events to a specific partition. + */ + public void publishEventsToPartition() { + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + + CreateBatchOptions options = new CreateBatchOptions().setPartitionId("0"); + EventDataBatch batch = producer.createBatch(options); + + // Add events to batch and when you want to send the batch, send it using the producer. + producer.send(batch); + } + + /** + * Code sample for publishing events with a partition key. + */ + public void publishEventsWithPartitionKey() { + EventHubProducerClient producer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .buildProducerClient(); + + CreateBatchOptions batchOptions = new CreateBatchOptions().setPartitionKey("grouping-key"); + EventDataBatch eventDataBatch = producer.createBatch(batchOptions); + + // Add events to batch and when you want to send the batch, send it using the producer. + producer.send(eventDataBatch); + } + + /** + * Code sample for consuming events from a specific partition. + */ + public void consumeEventsFromPartition() { + EventHubConsumerAsyncClient consumer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildAsyncConsumerClient(); + + // Receive events from partition with id "0", only getting events that are newly added to the partition. + consumer.receiveFromPartition("0", EventPosition.latest()).subscribe(event -> { + // Process each event as it arrives. + }); + } + + /** + * Code sample for consuming events from synchronous client. + */ + public void consumeEventsFromPartitionUsingSyncClient() { + EventHubConsumerClient consumer = new EventHubClientBuilder() + .connectionString("<< CONNECTION STRING FOR SPECIFIC EVENT HUB INSTANCE >>") + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .buildConsumerClient(); + + String partitionId = "<< EVENT HUB PARTITION ID >>"; + + // Get the first 15 events in the stream, or as many events as can be received within 40 seconds. + IterableStream events = consumer.receiveFromPartition(partitionId, 15, + EventPosition.earliest(), Duration.ofSeconds(40)); + for (PartitionEvent event : events) { + System.out.println("Event: " + event.getData().getBodyAsString()); + } + } + + /** + * Code sample for using Event Processor to consume events. + * @throws InterruptedException if the thread is interrupted. + */ + public void consumeEventsUsingEventProcessor() throws InterruptedException { + EventProcessorClient eventProcessorClient = new EventProcessorClientBuilder() + .consumerGroup("<< CONSUMER GROUP NAME >>") + .connectionString("<< EVENT HUB CONNECTION STRING >>") + .checkpointStore(new InMemoryCheckpointStore()) + .processEvent(eventContext -> { + System.out.println("Partition id = " + eventContext.getPartitionContext().getPartitionId() + " and " + + "sequence number of event = " + eventContext.getEventData().getSequenceNumber()); + }) + .processError(errorContext -> { + System.out + .println("Error occurred while processing events " + errorContext.getThrowable().getMessage()); + }) + .buildEventProcessorClient(); + + // This will start the processor. It will start processing events from all partitions. + eventProcessorClient.start(); + + // (for demo purposes only - adding sleep to wait for receiving events) + TimeUnit.SECONDS.sleep(2); + + // This will stop processing events. + eventProcessorClient.stop(); + } +} +