diff --git a/sdk/eventhubs/azure-messaging-eventhubs/README.md b/sdk/eventhubs/azure-messaging-eventhubs/README.md index 2d18142c964d..8849b6cdb51c 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/README.md @@ -414,7 +414,6 @@ Beyond those discussed, the Azure Event Hubs client library offers support for m advantage of the full feature set of the Azure Event Hubs service. In order to help explore some of the these scenarios, the following set of sample is available [here][samples_readme]. - ## Contributing If you would like to become an active contributor to this project please refer to our [Contribution @@ -439,15 +438,6 @@ Guidelines](./CONTRIBUTING.md) for more information. [qpid_proton_j_apache]: http://qpid.apache.org/proton/ [samples_readme]: ./src/samples/README.md [sample_examples]: ./src/samples/java/com/azure/messaging/eventhubs/ -[sample_consume_event]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java -[sample_consume_sequence_number]: ./src/samples/java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java -[sample_event_processor]: ./src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java -[sample_get_event_hubs_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/GetEventHubMetadata.java -[sample_publish_custom_metadata]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java -[sample_publish_identity]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java -[sample_publish_partitionId]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java -[sample_publish_partitionKey]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java -[sample_publish_size_limited]: ./src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java [source_code]: ./ [AmqpException]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpException.java [AmqpErrorCondition]: ../../core/azure-core-amqp/src/main/java/com/azure/core/amqp/exception/AmqpErrorCondition.java diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java index b4567ca5e821..9bbbc464394a 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java @@ -96,7 +96,7 @@ public static EventPosition fromEnqueuedTime(Instant enqueuedDateTime) { * @return An {@link EventPosition} object. */ public static EventPosition fromOffset(long offset) { - return fromOffset(offset, true); + return fromOffset(offset, false); } /** diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/README.md b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/README.md index e4149dc57766..0ca1fcc678c1 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/README.md +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/README.md @@ -9,6 +9,16 @@ Key concepts are explained in detail [here][sdk_readme_key_concepts]. ## Getting started Please refer to the [Getting Started][sdk_readme_getting_started] section. +### Obtaining an Event Hub instance connection string + +All the samples authorize with an Event Hub using a connection string generated for that Event Hub. The connection +string value can be obtained by: + +1. Going to your Event Hubs namespace in Azure Portal. +1. Creating an Event Hub instance. +1. Creating a "Shared access policy" for your Event Hub instance. +1. Copying the connection string from the policy's properties. + ## Examples - [Inspect Event Hub and partition properties][sample_get_event_hubs_metadata] @@ -16,10 +26,15 @@ Please refer to the [Getting Started][sdk_readme_getting_started] section. - [Publish events to a specific Event Hub partition with partition identifier][sample_publish_partitionId] - [Publish events to a specific Event Hub partition with partition key][sample_publish_partitionKey] - [Publish events to an Event Hub with a size-limited batch][sample_publish_size_limited] +- [Publish events using web sockets and a proxy][sample_publish_web_sockets_proxy] - [Publish events with custom metadata][sample_publish_custom_metadata] +- [Publish stream of events][sample_publish_stream_events] - [Consume events from an Event Hub partition][sample_consume_event] -- [Consume events starting from an event sequence number][sample_consume_sequence_number] - [Consume events from all partitions using EventProcessorClient][sample_event_processor] +- [Consume events from all partitions and manage state of processed events][sample_event_processor_state_management] +- [Consume events from all partitions and manage state of events using + EventProcessorClient][sample_event_processor_aggregate_state_management] +- [Consume events starting from an event sequence number][sample_consume_sequence_number] ## Troubleshooting See [Troubleshooting][sdk_readme_troubleshooting]. @@ -39,12 +54,16 @@ Guidelines](../../CONTRIBUTING.md) for more information. [sdk_readme_next_steps]: ../../README.md#next-steps [sample_consume_event]: ./java/com/azure/messaging/eventhubs/ConsumeEvents.java [sample_consume_sequence_number]: ./java/com/azure/messaging/eventhubs/ConsumeEventsFromKnownSequenceNumberPosition.java -[sample_event_processor]: ./java/com/azure/messaging/eventhubs/EventProcessorSample.java +[sample_event_processor]: ./java/com/azure/messaging/eventhubs/EventProcessorClientSample.java +[sample_event_processor_aggregate_state_management]: ./java/com/azure/messaging/eventhubs/EventProcessorClientAggregateEventsSample.java +[sample_event_processor_state_management]: ./java/com/azure/messaging/eventhubs/EventProcessorClientStateManagement.java [sample_get_event_hubs_metadata]: ./java/com/azure/messaging/eventhubs/GetEventHubMetadata.java [sample_publish_custom_metadata]: ./java/com/azure/messaging/eventhubs/PublishEventsWithCustomMetadata.java [sample_publish_identity]: ./java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java [sample_publish_partitionId]: ./java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java [sample_publish_partitionKey]: ./java/com/azure/messaging/eventhubs/PublishEventsWithPartitionKey.java [sample_publish_size_limited]: ./java/com/azure/messaging/eventhubs/PublishEventsWithSizeLimitedBatches.java +[sample_publish_stream_events]: ./java/com/azure/messaging/eventhubs/PublishStreamOfEvents.java +[sample_publish_web_sockets_proxy]: ./java/com/azure/messaging/eventhubs/PublishEventsWithWebSocketsAndProxy.java ![Impressions](https://azure-sdk-impressions.azurewebsites.net/api/impressions/azure-sdk-for-java%2Fsdk%2Feventhubs%2Fazure-messaging-eventhubs%2Fsrc%2Fsamples%2README.png) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java index f89dcab19dfb..17f2a39efa38 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/ConsumeEvents.java @@ -3,6 +3,7 @@ package com.azure.messaging.eventhubs; import com.azure.messaging.eventhubs.models.EventPosition; +import com.azure.messaging.eventhubs.models.PartitionContext; import com.azure.messaging.eventhubs.models.SendOptions; import reactor.core.Disposable; import reactor.core.publisher.Flux; @@ -24,8 +25,8 @@ public class ConsumeEvents { * Main method to invoke this demo about how to receive events from an Azure Event Hub instance. * * @param args Unused arguments to the program. - * @throws InterruptedException The countdown latch was interrupted while waiting for this sample to - * complete. + * + * @throws InterruptedException The countdown latch was interrupted while waiting for this sample to complete. */ public static void main(String[] args) throws InterruptedException { CountDownLatch countDownLatch = new CountDownLatch(NUMBER_OF_EVENTS); @@ -35,8 +36,8 @@ public static void main(String[] args) throws InterruptedException { // 2. Creating an Event Hub instance. // 3. Creating a "Shared access policy" for your Event Hub instance. // 4. Copying the connection string from the policy's properties. - String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; - + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; // Instantiate a client that will be used to call the service. // Create a consumer. // The "$Default" consumer group is created by default. This value can be found by going to the Event Hub @@ -63,12 +64,25 @@ public static void main(String[] args) throws InterruptedException { Disposable subscription = consumer.receiveFromPartition(firstPartition, EventPosition.latest()) .subscribe(partitionEvent -> { EventData event = partitionEvent.getData(); + PartitionContext partitionContext = partitionEvent.getPartitionContext(); + String contents = new String(event.getBody(), UTF_8); - System.out.println(String.format("[%s] Sequence Number: %s. Contents: %s", countDownLatch.getCount(), - event.getSequenceNumber(), contents)); + System.out.printf("[#%s] Partition id: %s. Sequence Number: %s. Contents: '%s'%n", + countDownLatch.getCount(), partitionContext.getPartitionId(), event.getSequenceNumber(), + contents); countDownLatch.countDown(); - }); + }, + error -> { + System.err.println("Error occurred while consuming events: " + error); + + // Count down until 0, so the main thread does not keep waiting for events. + while (countDownLatch.getCount() > 0) { + countDownLatch.countDown(); + } + }, () -> { + System.out.println("Finished reading events."); + }); EventHubProducerAsyncClient producer = new EventHubClientBuilder() .connectionString(connectionString) diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientAggregateEventsSample.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientAggregateEventsSample.java new file mode 100644 index 000000000000..30579d62796e --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientAggregateEventsSample.java @@ -0,0 +1,407 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.eventhubs; + +import com.azure.messaging.eventhubs.models.CloseContext; +import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import com.azure.messaging.eventhubs.models.ErrorContext; +import com.azure.messaging.eventhubs.models.EventContext; +import com.azure.messaging.eventhubs.models.InitializationContext; +import com.azure.messaging.eventhubs.models.PartitionContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.ConnectableFlux; +import reactor.core.publisher.DirectProcessor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; + +/** + * Demonstrates state management while processing events with {@link EventProcessorClient}. Shows usage of + * {@link EventProcessorClientBuilder#processPartitionInitialization(Consumer)}, + * {@link EventProcessorClientBuilder#processError(Consumer)}, + * {@link EventProcessorClientBuilder#processEvent(Consumer)}, and + * {@link EventProcessorClientBuilder#processPartitionClose(Consumer)}. + * + * An in-memory {@link CheckpointStore} is used to store checkpoint information for the sake of this demo. Production + * environments should leverage a durable store. + * + * A manufacturer has several machines on their assembly lines that emit temperature data. The manufacturer can use + * {@link EventProcessorClient} to aggregate the temperature data to look for anomalies. For example, the temperature + * data can say if a machine is over heating, or if no temperature data for a machine has been collected for a while, it + * may be offline. + * + * The partition key for each produced event is the name of the machine. This ensures that the temperature data for that + * machine always gets routed to the same partition. The contents of each event is the temperature of that machine in + * Celsius. + * + * Every 5 seconds, {@link MachineInformation} reports the current average temperature of the machine. + */ +public class EventProcessorClientAggregateEventsSample { + private static final Duration REPORTING_INTERVAL = Duration.ofSeconds(5); + private static final String EH_CONNECTION_STRING = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + + /** + * Main method to demonstrate starting and stopping a {@link EventProcessorClient}. + * + * @param args The input arguments to this executable. + * @throws Exception If there are any errors while running the {@link EventProcessorClient}. + */ + public static void main(String[] args) throws Exception { + final MachineEventsProcessor aggregator = new MachineEventsProcessor(REPORTING_INTERVAL); + + final EventProcessorClient client = new EventProcessorClientBuilder() + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .connectionString(EH_CONNECTION_STRING) + .processPartitionInitialization(context -> aggregator.onInitialize(context)) + .processPartitionClose(context -> aggregator.onClose(context)) + .processEvent(event -> aggregator.onEvent(event)) + .processError(error -> aggregator.onError(error)) + .checkpointStore(new InMemoryCheckpointStore()) + .buildEventProcessorClient(); + + System.out.println("Starting event processor"); + final AtomicBoolean isRunning = new AtomicBoolean(true); + client.start(); + + // Continue to perform other tasks while the processor is running in the background. In this sample, we are + // randomly generating fake machine events. + generateEvents(isRunning).subscribe(); + + System.out.println("Sleeping..."); + Thread.sleep(TimeUnit.SECONDS.toMillis(30)); + isRunning.set(false); + + System.out.println("Stopping event processor"); + client.stop(); + System.out.println("Exiting process"); + } + + /** + * Helper method that generates events for machines "2A", "9B", and "6C" and sends them to the service. + */ + private static Mono generateEvents(AtomicBoolean isRunning) { + final Logger logger = LoggerFactory.getLogger("Producer"); + final Scheduler scheduler = Schedulers.newElastic("produce"); + final Duration operationTimeout = Duration.ofSeconds(5); + final String[] machineIds = new String[]{"2A", "9B", "6C"}; + final Random random = new Random(); + final EventHubProducerAsyncClient client = new EventHubClientBuilder() + .connectionString(EH_CONNECTION_STRING) + .buildAsyncProducerClient(); + + return Mono.fromRunnable(() -> { + while (isRunning.get()) { + int milliseconds = random.nextInt(1000); + + try { + TimeUnit.MILLISECONDS.sleep(milliseconds); + } catch (InterruptedException ignored) { + } + + final String machineId = machineIds[random.nextInt(machineIds.length)]; + // We want a temperature between 0 - 100. + final int temperature = Math.abs(random.nextInt() % 101); + + logger.info("[{}] Temperature: {}C", machineId, temperature); + + final EventData event = new EventData(String.valueOf(temperature)); + final CreateBatchOptions batchOptions = new CreateBatchOptions().setPartitionKey(machineId); + + client.createBatch(batchOptions).flatMap(batch -> { + batch.tryAdd(event); + return client.send(batch); + }).block(operationTimeout); + } + }).doFinally(signal -> { + logger.info("Disposing of producer."); + client.close(); + }).subscribeOn(scheduler); + } +} + +/** + * Keeps track of machine information by analyzing data in Event Hubs. + */ +class MachineEventsProcessor implements AutoCloseable { + private final Logger logger = LoggerFactory.getLogger(EventProcessorClientSample.class); + private final AtomicBoolean isDisposed = new AtomicBoolean(); + private final Duration reportingInterval; + + /** + * Holds information about what machines this instance is processing temperature data for. + * Key: Machine id. + * Value: Temperature information for that machine. + */ + private final ConcurrentHashMap machineInformation = new ConcurrentHashMap<>(); + /** + * Holds information about what partitions this instance is processing and its associated machines. + * Key: Partition id. + * Value: List of machine ids in that partition. + */ + private final ConcurrentHashMap> partitionsProcessing = new ConcurrentHashMap<>(); + + /** + * Creates an instance of {@link MachineEventsProcessor}. + * + * @param reportingInterval Interval at which to report temperature events. + */ + MachineEventsProcessor(Duration reportingInterval) { + this.reportingInterval = reportingInterval; + } + + /** + * Processes each event by extracting temperature and machine information from the event. + * + * @param eventContext The event received from Event Hubs. + */ + void onEvent(EventContext eventContext) { + final PartitionContext partitionContext = eventContext.getPartitionContext(); + final EventData event = eventContext.getEventData(); + final String contents = event.getBodyAsString(); + final int temperature; + try { + temperature = Integer.parseInt(contents); + } catch (NumberFormatException ex) { + logger.warn("Unable to parse temperature data. Partition: #{}. Seq #{}. Contents: #{}. Error: {}", + partitionContext.getPartitionId(), event.getSequenceNumber(), contents, ex); + return; + } + + final String machineId = event.getPartitionKey(); + if (machineId == null || machineId.isEmpty()) { + logger.warn("PartitionKey is not set on event. Partition: #{}. Seq #{}. Contents: #{}.", + partitionContext.getPartitionId(), event.getSequenceNumber(), contents); + return; + } + + partitionsProcessing.compute(partitionContext.getPartitionId(), (key, value) -> { + if (value == null) { + value = new HashSet<>(); + } + value.add(machineId); + return value; + }); + + final MachineInformation information = machineInformation.computeIfAbsent(machineId, + key -> new MachineInformation(key, reportingInterval)); + + information.onTemperatureEvent(event.getEnqueuedTime(), temperature); + + // Update checkpoint so customers know + eventContext.updateCheckpoint(); + } + + /** + * On initialisation, keeps track of which partitions it is processing. + * + * @param initializationContext Initialisation information. + */ + void onInitialize(InitializationContext initializationContext) { + final PartitionContext partition = initializationContext.getPartitionContext(); + partitionsProcessing.computeIfAbsent(partition.getPartitionId(), key -> new HashSet<>()); + } + + /** + * When an occurs, reports that error to a log. + * + * @param errorContext Error that occurred while processing events. + */ + void onError(ErrorContext errorContext) { + final PartitionContext partition = errorContext.getPartitionContext(); + + logger.error("Error occurred processing partition '{}'? {}", partition.getPartitionId(), + errorContext.getThrowable()); + } + + /** + * When a partition is lost, will dispose of machine information it is processing. In a real service, it might + * persist the averages it currently has, so there is no processed data lost. + * + * @param closeContext Close context for that partition. + */ + void onClose(CloseContext closeContext) { + final PartitionContext partition = closeContext.getPartitionContext(); + final Set machineIds = partitionsProcessing.remove(partition.getPartitionId()); + + logger.info("Stopped processing partition '{}'. Reason: '{}'", partition.getPartitionId(), + closeContext.getCloseReason()); + + for (String id : machineIds) { + final MachineInformation information = machineInformation.remove(id); + if (information == null) { + continue; + } + + // We may want to persist the last calculated average temperature to a durable store, so another process can + // continue from where this instance left off. + logger.info("Stopped processing temperatures for machine: {}", information.getIdentifier()); + information.close(); + } + } + + /** + * Disposes of all the machine information and closes the processor. + */ + @Override + public void close() { + if (isDisposed.getAndSet(true)) { + return; + } + + partitionsProcessing.clear(); + machineInformation.forEach((key, value) -> value.close()); + machineInformation.clear(); + } +} + +/** + * Contains information about a single machine in the factory. + */ +class MachineInformation implements AutoCloseable { + private final String identifier; + private final Logger logger = LoggerFactory.getLogger(MachineInformation.class); + private final AtomicReference> temperatures = new AtomicReference<>(new ArrayList<>()); + private final ConnectableFlux averageTemperatures; + private final DirectProcessor onDispose = DirectProcessor.create(); + private final AtomicBoolean isDisposed = new AtomicBoolean(); + + private volatile Instant lastReported = Instant.EPOCH; + + /** + * Creates a new instance. + * + * @param identifier Identifier for the machine. + * @param reportingInterval Interval at which to emit average temperature information. + */ + MachineInformation(String identifier, Duration reportingInterval) { + this.identifier = identifier; + this.averageTemperatures = Flux.interval(reportingInterval) + .takeUntilOther(onDispose) + .map(unused -> { + final Instant timeCalculated = Instant.now(); + final List temperaturesInInterval = temperatures.getAndSet(new ArrayList<>()); + if (temperaturesInInterval.size() == 0) { + logger.info("[{}]: Average: null", this.identifier); + return new AverageTemperature(timeCalculated, null); + } + + final int sum = temperaturesInInterval.stream().reduce(0, Integer::sum); + double average = sum / (double) temperaturesInInterval.size(); + + logger.info("[{}]: Average: {}", this.identifier, average); + return new AverageTemperature(timeCalculated, average); + }).publish(); + + averageTemperatures.connect(); + } + + /** + * Submits additional temperature data for that machine. + * + * @param dateEnqueued The time the temperature was acquired. + * @param temperature The temperature. + */ + void onTemperatureEvent(Instant dateEnqueued, int temperature) { + lastReported = dateEnqueued; + temperatures.getAndUpdate(list -> { + list.add(temperature); + return list; + }); + } + + /** + * Gets the machine identifier. + * + * @return The machine identifier. + */ + String getIdentifier() { + return identifier; + } + + /** + * A stream of average temperature information for the machine. + * + * @return Stream of average temperature information. + */ + Flux getAverageTemperatures() { + return averageTemperatures; + } + + /** + * Gets the last reported time. + * + * @return The last reported time. + */ + public Instant getLastReported() { + return lastReported; + } + + /** + * Disposes of the information. + */ + @Override + public void close() { + if (isDisposed.getAndSet(true)) { + return; + } + + final FluxSink sink = onDispose.sink(); + sink.next(true); + sink.complete(); + } +} + +/** + * Average temperature calculation. + */ +class AverageTemperature { + private final Instant timeCalculated; + private final Double temperature; + + /** + * Creates an instance. + * + * @param timeCalculated The time that the average temperature was calculated. + * @param temperature The average temperature in Celsius. {@code null} if there was no data during that period. + */ + AverageTemperature(Instant timeCalculated, Double temperature) { + this.timeCalculated = timeCalculated; + this.temperature = temperature; + } + + /** + * Gets the time that the measurement was calculated. + * + * @return The time that the measurement was calculated. + */ + public Instant getTimeCalculated() { + return timeCalculated; + } + + /** + * Gets the average temperature in Celsius. + * + * @return The average temperature in Celsius, or {@code null} if there was no data when the temperature was + * calculated. + */ + public Double getTemperature() { + return temperature; + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorJavaDocCodeSamples.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientJavaDocCodeSamples.java similarity index 96% rename from sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorJavaDocCodeSamples.java rename to sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientJavaDocCodeSamples.java index d04800c5bd3d..6a43081a63be 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorJavaDocCodeSamples.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientJavaDocCodeSamples.java @@ -6,7 +6,7 @@ /** * Code snippets for {@link EventProcessorClient}. */ -public final class EventProcessorJavaDocCodeSamples { +public final class EventProcessorClientJavaDocCodeSamples { /** * Code snippet for showing how to start and stop an {@link EventProcessorClient}. diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientSample.java similarity index 96% rename from sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java rename to sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientSample.java index 8aea74761e00..63ba02c7a007 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorSample.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientSample.java @@ -5,16 +5,16 @@ import com.azure.messaging.eventhubs.models.ErrorContext; import com.azure.messaging.eventhubs.models.EventContext; -import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; /** * Sample code to demonstrate how a customer might use {@link EventProcessorClient}. */ -public class EventProcessorSample { +public class EventProcessorClientSample { private static final String EH_CONNECTION_STRING = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; @@ -26,7 +26,7 @@ public class EventProcessorSample { */ public static void main(String[] args) throws Exception { - Logger logger = LoggerFactory.getLogger(EventProcessorSample.class); + Logger logger = LoggerFactory.getLogger(EventProcessorClientSample.class); Consumer processEvent = eventContext -> { logger.info( "Processing event: Event Hub name = {}; consumer group name = {}; partition id = {}; sequence number = {}", @@ -34,6 +34,7 @@ public static void main(String[] args) throws Exception { eventContext.getPartitionContext().getConsumerGroup(), eventContext.getPartitionContext().getPartitionId(), eventContext.getEventData().getSequenceNumber()); + eventContext.updateCheckpoint(); }; diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientStateManagement.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientStateManagement.java new file mode 100644 index 000000000000..cf1e16cba759 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/EventProcessorClientStateManagement.java @@ -0,0 +1,105 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.eventhubs; + +import com.azure.messaging.eventhubs.models.CloseReason; +import com.azure.messaging.eventhubs.models.PartitionContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +/** + * Sample demonstrating how to maintain processing state. Counts the number of events that were processed for each + * partition. + * + * @see EventProcessorClientAggregateEventsSample Another sample demonstrating state management. + */ +public class EventProcessorClientStateManagement { + private static final String EH_CONNECTION_STRING = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + + private final Logger logger = LoggerFactory.getLogger("Processor"); + /** + * Keeps track of the number of events processed from each partition. + * Key: Partition id + * Value: Number of events processed for each partition. + */ + private final ConcurrentHashMap eventsProcessed = new ConcurrentHashMap<>(); + + /** + * Main method to demonstrate starting and stopping a {@link EventProcessorClient}. + * + * @param args The input arguments to this executable. + * @throws Exception If there are any errors while running the {@link EventProcessorClient}. + */ + public static void main(String[] args) throws Exception { + final EventProcessorClientStateManagement program = new EventProcessorClientStateManagement(); + final EventProcessorClient client = new EventProcessorClientBuilder() + .consumerGroup(EventHubClientBuilder.DEFAULT_CONSUMER_GROUP_NAME) + .connectionString(EH_CONNECTION_STRING) + .processPartitionInitialization(context -> program.onInitialize(context.getPartitionContext())) + .processPartitionClose(context -> program.onClose(context.getPartitionContext(), context.getCloseReason())) + .processEvent(event -> program.onEvent(event.getPartitionContext(), event.getEventData())) + .processError(error -> program.onError(error.getPartitionContext(), error.getThrowable())) + .checkpointStore(new InMemoryCheckpointStore()) + .buildEventProcessorClient(); + + System.out.println("Starting event processor"); + client.start(); + + // Continue to perform other tasks while the processor is running in the background. + Thread.sleep(TimeUnit.SECONDS.toMillis(30)); + + System.out.println("Stopping event processor"); + client.stop(); + System.out.println("Exiting process"); + } + + /** + * When an occurs, reports that error to a log. + * + * @param partitionContext Context information for the partition in which this error occurred. + * @param error Error that occurred. + */ + void onError(PartitionContext partitionContext, Throwable error) { + logger.error("Error occurred processing partition '{}'. Exception: {}", partitionContext.getPartitionId(), + error); + } + + /** + * On initialisation, keeps track of which partitions it is processing. + * + * @param partitionContext Information about partition it is processing. + */ + void onInitialize(PartitionContext partitionContext) { + logger.info("Starting to process partition {}", partitionContext.getPartitionId()); + eventsProcessed.computeIfAbsent(partitionContext.getPartitionId(), unused -> 0); + } + + /** + * Invoked when a partition is no longer being processed. + * + * @param partitionContext Context information for the partition that is no longer being processed. + * @param reason Reason for no longer processing partition. + */ + void onClose(PartitionContext partitionContext, CloseReason reason) { + logger.info("Stopping processing of partition {}. Reason: {}", partitionContext.getPartitionId(), reason); + eventsProcessed.remove(partitionContext.getPartitionId()); + } + + /** + * Processes an event from the partition. Aggregates the number of events that were processed in this partition. + * + * @param partitionContext Information about which partition this event was in. + * @param eventData Event from the partition. + */ + void onEvent(PartitionContext partitionContext, EventData eventData) { + final Integer count = eventsProcessed.compute(partitionContext.getPartitionId(), + (key, value) -> value == null ? 1 : value + 1); + + logger.info("Event {} received for partition: {}. # of events processed: {}", + eventData.getSequenceNumber(), partitionContext.getPartitionId(), count); + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java index 3acc060d8b7a..2ae52053caf5 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsToSpecificPartition.java @@ -2,6 +2,8 @@ // Licensed under the MIT License. package com.azure.messaging.eventhubs; +import com.azure.core.amqp.AmqpRetryMode; +import com.azure.core.amqp.AmqpRetryOptions; import com.azure.messaging.eventhubs.models.CreateBatchOptions; import reactor.core.Exceptions; import reactor.core.publisher.Flux; @@ -14,8 +16,13 @@ import static java.nio.charset.StandardCharsets.UTF_8; /** - * Sample demonstrates how to sent events to specific event hub by defining partition id using {@link - * CreateBatchOptions#setPartitionId(String)}. + * Sample demonstrates how to: + * + *
    + *
  • Send events to specific event hub partition by defining partition id using + * {@link CreateBatchOptions#setPartitionId(String)}.
    • + *
  • Set a custom retry policy for Event Hub operations.
    • + *
*/ public class PublishEventsToSpecificPartition { private static final Duration OPERATION_TIMEOUT = Duration.ofSeconds(30); @@ -33,9 +40,16 @@ public static void main(String[] args) { // 4. Copying the connection string from the policy's properties. String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + // Set some custom retry options other than the default set. + AmqpRetryOptions retryOptions = new AmqpRetryOptions() + .setDelay(Duration.ofSeconds(30)) + .setMaxRetries(2) + .setMode(AmqpRetryMode.EXPONENTIAL); + // Instantiate a client that will be used to call the service. EventHubProducerAsyncClient producer = new EventHubClientBuilder() .connectionString(connectionString) + .retry(retryOptions) .buildAsyncProducerClient(); // To send our events, we need to know what partition to send it to. For the sake of this example, we take the @@ -46,10 +60,10 @@ public static void main(String[] args) { String firstPartition = producer.getPartitionIds().blockFirst(OPERATION_TIMEOUT); // We will publish three events based on simple sentences. - Flux data = Flux.just( - new EventData("EventData Sample 1".getBytes(UTF_8)), - new EventData("EventData Sample 2".getBytes(UTF_8)), - new EventData("EventData Sample 3".getBytes(UTF_8))); + Flux events = Flux.just( + new EventData("This is the first event.".getBytes(UTF_8)), + new EventData("This is the second event.".getBytes(UTF_8)), + new EventData("This is the third event.".getBytes(UTF_8))); // Create a batch to send the events. final CreateBatchOptions options = new CreateBatchOptions() @@ -60,7 +74,7 @@ public static void main(String[] args) { // We try to add as many events as a batch can fit based on the event size and send to Event Hub when // the batch can hold no more events. Create a new batch for next set of events and repeat until all events // are sent. - data.flatMap(event -> { + events.flatMap(event -> { final EventDataBatch batch = currentBatch.get(); if (batch.tryAdd(event)) { return Mono.empty(); diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java index 775b5b92b295..4a74568de2ba 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithAzureIdentity.java @@ -4,13 +4,10 @@ import com.azure.core.credential.TokenCredential; import com.azure.identity.DefaultAzureCredentialBuilder; -import reactor.core.Exceptions; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import java.time.Duration; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; +import java.util.Arrays; +import java.util.List; import static java.nio.charset.StandardCharsets.UTF_8; @@ -26,7 +23,7 @@ public class PublishEventsWithAzureIdentity { * @param args Unused arguments to the program. */ public static void main(String[] args) { - Flux telemetryEvents = Flux.just( + List telemetryEvents = Arrays.asList( new EventData("Roast beef".getBytes(UTF_8)), new EventData("Cheese".getBytes(UTF_8)), new EventData("Tofu".getBytes(UTF_8)), @@ -47,59 +44,34 @@ public static void main(String[] args) { // Create a producer. // "<>" will look similar to "{your-namespace}.servicebus.windows.net" // "<>" will be the name of the Event Hub instance you created inside the Event Hubs namespace. - EventHubProducerAsyncClient producer = new EventHubClientBuilder() + EventHubProducerClient producer = new EventHubClientBuilder() .credential( "<>", "<>", credential) - .buildAsyncProducerClient(); + .buildProducerClient(); - final AtomicReference currentBatch = new AtomicReference<>( - producer.createBatch().block()); + // Creates an EventDataBatch where the Event Hubs service will automatically load balance the events between all + // available partitions. + EventDataBatch currentBatch = producer.createBatch(); - // The sample Flux contains three events, but it could be an infinite stream of telemetry events. - telemetryEvents.flatMap(event -> { - final EventDataBatch batch = currentBatch.get(); - if (batch.tryAdd(event)) { - return Mono.empty(); + // We try to add as many events as a batch can fit based on the event size and send to Event Hub when + // the batch can hold no more events. Create a new batch for next set of events and repeat until all events + // are sent. + for (EventData event : telemetryEvents) { + if (currentBatch.tryAdd(event)) { + continue; } - // The batch is full, so we create a new batch and send the batch. Mono.when completes when both operations - // have completed. - return Mono.when( - producer.send(batch), - producer.createBatch().map(newBatch -> { - currentBatch.set(newBatch); + // The batch is full, so we create a new batch and send the batch. + producer.send(currentBatch); + currentBatch = producer.createBatch(); - // Add that event that we couldn't before. - if (!newBatch.tryAdd(event)) { - throw Exceptions.propagate(new IllegalArgumentException(String.format( - "Event is too large for an empty batch. Max size: %s. Event: %s", - newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); - } - - return newBatch; - })); - }).then() - .doFinally(signal -> { - final EventDataBatch batch = currentBatch.getAndSet(null); - if (batch != null) { - producer.send(batch).block(OPERATION_TIMEOUT); - } - }) - .subscribe(unused -> System.out.println("Complete"), - error -> System.out.println("Error sending events: " + error), - () -> System.out.println("Completed sending events.")); - - // The .subscribe() creation and assignment is not a blocking call. For the purpose of this example, we sleep - // the thread so the program does not end before the send operation is complete. Using .block() instead of - // .subscribe() will turn this into a synchronous call. - try { - TimeUnit.SECONDS.sleep(5); - } catch (InterruptedException ignored) { - } finally { - // Disposing of our producer. - producer.close(); + // Add that event that we couldn't before. + if (!currentBatch.tryAdd(event)) { + System.err.printf("Event is too large for an empty batch. Skipping. Max size: %s. Event: %s%n", + currentBatch.getMaxSizeInBytes(), event.getBodyAsString()); + } } } } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithWebSocketsAndProxy.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithWebSocketsAndProxy.java new file mode 100644 index 000000000000..60b9cf5000df --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishEventsWithWebSocketsAndProxy.java @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.eventhubs; + + +import com.azure.core.amqp.AmqpTransportType; +import com.azure.core.amqp.ProxyAuthenticationType; +import com.azure.core.amqp.ProxyOptions; + +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.util.Arrays; +import java.util.List; + +import static java.nio.charset.StandardCharsets.UTF_8; + +/** + * Demonstrates how to publish events when using web sockets and a proxy and synchronous producer. + */ +public class PublishEventsWithWebSocketsAndProxy { + /** + * Main method to invoke this sample. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + List telemetryEvents = Arrays.asList( + new EventData("Roast beef".getBytes(UTF_8)), + new EventData("Cheese".getBytes(UTF_8)), + new EventData("Tofu".getBytes(UTF_8)), + new EventData("Turkey".getBytes(UTF_8))); + + // The connection string value can be obtained by: + // 1. Going to your Event Hubs namespace in Azure Portal. + // 2. Creating an Event Hub instance. + // 3. Creating a "Shared access policy" for your Event Hub instance. + // 4. Copying the connection string from the policy's properties. + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + + // By default, the AMQP port 5671 is used, but clients can use web sockets, port 443. + // When using web sockets, developers can specify proxy options. + // ProxyOptions.SYSTEM_DEFAULTS can be used if developers want to use the JVM configured proxy. + ProxyOptions proxyOptions = new ProxyOptions(ProxyAuthenticationType.DIGEST, + new Proxy(Proxy.Type.HTTP, InetSocketAddress.createUnresolved("10.13.1.3", 9992)), + "digest-user", "digest-user-password"); + + // Instantiate a client that will be used to call the service. + EventHubProducerClient producer = new EventHubClientBuilder() + .transportType(AmqpTransportType.AMQP_WEB_SOCKETS) + .proxyOptions(proxyOptions) + .connectionString(connectionString) + .buildProducerClient(); + + // Creates an EventDataBatch where the Event Hubs service will automatically load balance the events between all + // available partitions. + EventDataBatch currentBatch = producer.createBatch(); + + // We try to add as many events as a batch can fit based on the event size and send to Event Hub when + // the batch can hold no more events. Create a new batch for next set of events and repeat until all events + // are sent. + for (EventData event : telemetryEvents) { + if (currentBatch.tryAdd(event)) { + continue; + } + + // The batch is full, so we create a new batch and send the batch. + producer.send(currentBatch); + currentBatch = producer.createBatch(); + + // Add that event that we couldn't before. + if (!currentBatch.tryAdd(event)) { + System.err.printf("Event is too large for an empty batch. Skipping. Max size: %s. Event: %s%n", + currentBatch.getMaxSizeInBytes(), event.getBodyAsString()); + } + } + + producer.close(); + } +} diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishStreamOfEvents.java b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishStreamOfEvents.java new file mode 100644 index 000000000000..58f71414b3b6 --- /dev/null +++ b/sdk/eventhubs/azure-messaging-eventhubs/src/samples/java/com/azure/messaging/eventhubs/PublishStreamOfEvents.java @@ -0,0 +1,234 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.messaging.eventhubs; + +import com.azure.messaging.eventhubs.models.CreateBatchOptions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.DirectProcessor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.util.Objects; +import java.util.Random; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +/** + * Demonstrates how to continuously publish events using {@link EventHubProducerAsyncClient}. + * {@link CustomPublisher} is publishes size-limited batches when they are full. In addition, it checks for periodically + * for a partial batch that needs to be sent. + */ +public class PublishStreamOfEvents { + private static final String EVENT_NUMBER = "EVENT_NUMBER"; + + /** + * Main method to invoke this demo about how continuously publish a stream of events to Event Hubs. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + // The connection string value can be obtained by: + // 1. Going to your Event Hubs namespace in Azure Portal. + // 2. Creating an Event Hub instance. + // 3. Creating a "Shared access policy" for your Event Hub instance. + // 4. Copying the connection string from the policy's properties. + String connectionString = "Endpoint={endpoint};SharedAccessKeyName={sharedAccessKeyName};" + + "SharedAccessKey={sharedAccessKey};EntityPath={eventHubName}"; + + // Limit the size of the batches to 1024 bytes instead of using the default max size. + final CreateBatchOptions batchOptions = new CreateBatchOptions().setMaximumSizeInBytes(1024); + final CustomPublisher publisher = new CustomPublisher(connectionString, Duration.ofSeconds(1), batchOptions); + + // This represents a stream of events that we want to publish. + final DirectProcessor events = DirectProcessor.create(); + + System.out.println("Publishing events..."); + publisher.publish(events).subscribe(unused -> System.out.println("Completed."), + error -> System.err.println("Error sending events: " + error), + () -> System.out.println("Completed sending events.")); + + emitEvents(events.sink()); + + // The .subscribe() creation and assignment is not a blocking call. For the purpose of this example, we sleep + // the thread so the program does not end before the send operation is complete. + try { + TimeUnit.SECONDS.sleep(10); + } catch (InterruptedException ignored) { + } finally { + System.out.println("Closing custom publisher."); + publisher.close(); + } + } + + /** + * Helper function that emits 50 events. The interval between each event is randomly selected between 0 - 250ms and + * is a random substring of the lorem ipsum text. + * + * @param sink Sink for generated events. + */ + private static void emitEvents(FluxSink sink) { + final String contents = "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do " + + "eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis " + + "nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure " + + "dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur" + + " sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est " + + "laborum."; + final Random random = new Random(); + + for (int i = 0; i < 50; i++) { + int milliseconds = random.nextInt(250); + + try { + TimeUnit.MILLISECONDS.sleep(milliseconds); + } catch (InterruptedException ignored) { + } + + final int size = random.nextInt(contents.length()); + final int endIndex = size < 1 ? contents.length() : size; + final EventData event = new EventData(contents.substring(0, endIndex)); + event.getProperties().put(EVENT_NUMBER, String.valueOf(i)); + + sink.next(event); + } + + sink.complete(); + } + + /** + * Aggregates events into batches based on the given {@link CreateBatchOptions} and sends the batches to the Event + * Hubs service when either: + * + *
    + *
  • The batch has reached its max size and no more events can be added to it.
    • + *
  • The timeout window has elapsed.
    • + *
+ */ + private static class CustomPublisher implements AutoCloseable { + private final Logger logger = LoggerFactory.getLogger(CustomPublisher.class); + private final AtomicBoolean isDisposed = new AtomicBoolean(); + private final AtomicReference currentBatch = new AtomicReference<>(); + private final EventHubProducerAsyncClient producer; + private final CreateBatchOptions batchOptions; + private final Duration windowDuration; + + /** + * Creates a new instance of {@link CustomPublisher}. + * + * @param connectionString Connection string for the specific Event Hub. + * @param windowDuration Intervals to check for an available {@link EventDataBatch} to send. + * @param batchOptions Options to use when creating the {@link EventDataBatch}. If {@code null}, the default + * batch options are used. + */ + CustomPublisher(String connectionString, Duration windowDuration, CreateBatchOptions batchOptions) { + producer = new EventHubClientBuilder() + .connectionString(connectionString) + .buildAsyncProducerClient(); + + this.batchOptions = batchOptions != null ? batchOptions : new CreateBatchOptions(); + this.windowDuration = Objects.requireNonNull(windowDuration, "'windowDuration' cannot be null."); + } + + /** + * Subscribes to a stream of events and publishes event batches when: + * + *
    + *
  • The {@link EventDataBatch} is full.
    • + *
  • Timeout window has elapsed and there is a batch.
    • + *
+ * + * @param events Events to publish to the service. + * + * @return Mono that completes when all the events have been published. + */ + Mono publish(Flux events) { + final Flux fullBatchFlux = Flux.create(sink -> { + events.subscribe(event -> { + // For each event, tries to add it to the current batch. If the current batch is full, then we emit + // that. Otherwise an empty Mono is emitted. + // Blocking at the very end of the method to ensure that the event is added to a batch before + // requesting another event. + getOrCreateBatch() + .flatMap(batch -> batch.tryAdd(event) ? Mono.empty() : Mono.just(batch)) + .flatMap(fullBatch -> { + sink.next(fullBatch); + + return createBatch().map(newBatch -> { + if (!newBatch.tryAdd(event)) { + sink.error(new IllegalArgumentException(String.format( + "Event is too large for an empty batch. Max size: %s. Event: %s", + newBatch.getMaxSizeInBytes(), event.getBodyAsString()))); + } + return newBatch; + }); + }).block(); + }, error -> { + sink.error(new RuntimeException("Error fetching next event.", error)); + }, () -> { + final EventDataBatch lastBatch = currentBatch.getAndSet(null); + if (lastBatch != null) { + sink.next(lastBatch); + } + + sink.complete(); + }); + }).publish().autoConnect(); + + // Periodically checks to see if there is an available, not full, batch to send. If there is, it sends that + // batch. It keeps checking for an available batch until the last event in `events` has been published. + final Flux emitAtIntervals = Flux.interval(windowDuration) + .takeUntilOther(events.then()) + .flatMap(v -> { + final EventDataBatch batch = currentBatch.getAndSet(null); + logger.info("Interval check. Has items? {}", batch != null && batch.getCount() > 0); + return batch != null ? Mono.just(batch) : Mono.empty(); + }); + + // Merge the two fluxes together so the results + return Flux.merge(fullBatchFlux, emitAtIntervals) + .flatMap(batchToSend -> { + final String eventNumbers = batchToSend.getEvents() + .stream() + .map(e -> e.getProperties().getOrDefault(EVENT_NUMBER, "n/a").toString()) + .collect(Collectors.joining(", ")); + + logger.info("Sending batch with {} events. Size: {} bytes. Event numbers in batch [{}]", + batchToSend.getCount(), batchToSend.getSizeInBytes(), eventNumbers); + + return producer.send(batchToSend); + }) + .then(); + } + + private Mono getOrCreateBatch() { + final EventDataBatch current = currentBatch.get(); + return current != null + ? Mono.just(current) + : createBatch(); + } + + private Mono createBatch() { + return producer.createBatch(batchOptions).map(batch -> { + currentBatch.set(batch); + return batch; + }); + } + + public void close() { + if (isDisposed.getAndSet(true)) { + return; + } + + final EventDataBatch batch = currentBatch.getAndSet(null); + if (batch != null) { + producer.send(batch).block(Duration.ofSeconds(30)); + } + + producer.close(); + } + } +}