com.azure.messaging.eventhubs

Class EventHubConsumer

java.lang.Object

com.azure.messaging.eventhubs.EventHubConsumer

All Implemented Interfaces:

Closeable, AutoCloseable

public class EventHubConsumer
extends Object
implements Closeable

A consumer responsible for reading EventData from a specific Event Hub partition in the context of a specific consumer group.

• If EventHubConsumer is created where EventHubConsumerOptions.getOwnerLevel() has a value, then Event Hubs service will guarantee only one active consumer exists per partitionId and consumer group combination. This consumer is sometimes referred to as an "Epoch Consumer."
• Multiple consumers per partitionId and consumer group combination can be created by not setting EventHubConsumerOptions.getOwnerLevel() when creating consumers. This non-exclusive consumer is sometimes referred to as a "Non-Epoch Consumer."

Creating a synchronous consumer

Create an EventHubConsumer using EventHubClient.

EventHubClient client = new EventHubClientBuilder()
    .connectionString("event-hub-instance-connection-string")
    .buildClient();

String partitionId = "0";
String consumerGroup = "$DEFAULT";
EventHubConsumer consumer = client.createConsumer(consumerGroup, partitionId, EventPosition.latest());

Consuming events from an Event Hub

Events can be consumed using EventHubConsumer.receive(int) or EventHubConsumer.receive(int, Duration). The call to `receive` completes and returns an IterableStream when either the number of events is reached, or the timeout duration is reached.

// Obtain partitionId from EventHubClient.getPartitionIds().
String partitionId = "0";
Instant twelveHoursAgo = Instant.now().minus(Duration.ofHours(12));
EventHubConsumer consumer = client.createConsumer(EventHubAsyncClient.DEFAULT_CONSUMER_GROUP_NAME, partitionId,
    EventPosition.fromEnqueuedTime(twelveHoursAgo));

IterableStream<EventData> events = consumer.receive(100, Duration.ofSeconds(30));

for (EventData event : events) {
    // For each event, perform some sort of processing.
    System.out.print("Event received: " + event.getSequenceNumber());
}

// Gets the next set of events to consume and process.
IterableStream<EventData> nextEvents = consumer.receive(100, Duration.ofSeconds(30));

See Also:

EventHubClient.createConsumer(String, String, EventPosition), EventHubClient.createConsumer(String, String, EventPosition, EventHubConsumerOptions)

Method Summary

Modifier and Type	Method and Description
void	close()
LastEnqueuedEventProperties	getLastEnqueuedEventProperties() A set of information about the last enqueued event of a partition, as observed by the consumer as events are received from the Event Hubs service.
IterableStream<EventData>	receive(int maximumMessageCount) Receives a batch of EventData from the Event Hub partition.
IterableStream<EventData>	receive(int maximumMessageCount, Duration maximumWaitTime) Receives a batch of EventData from the Event Hub partition

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

receive

public IterableStream<EventData> receive(int maximumMessageCount)

Receives a batch of EventData from the Event Hub partition.

Parameters:

maximumMessageCount - The maximum number of messages to receive in this batch.

Returns:

A set of EventData that was received. The iterable contains up to maximumMessageCount events.

Throws:

IllegalArgumentException - if maximumMessageCount is less than 1.

receive

public IterableStream<EventData> receive(int maximumMessageCount,
                                         Duration maximumWaitTime)

Receives a batch of EventData from the Event Hub partition

Parameters:

maximumMessageCount - The maximum number of messages to receive in this batch.

maximumWaitTime - The maximum amount of time to wait to build up the requested message count for the batch; if not specified, the default wait time specified when the consumer was created will be used.

Returns:

A set of EventData that was received. The iterable contains up to maximumMessageCount events.

Throws:

NullPointerException - if maximumWaitTime is null.

IllegalArgumentException - if maximumMessageCount is less than 1 or maximumWaitTime is zero or a negative duration.

getLastEnqueuedEventProperties

public LastEnqueuedEventProperties getLastEnqueuedEventProperties()

A set of information about the last enqueued event of a partition, as observed by the consumer as events are received from the Event Hubs service.

Returns:

null if EventHubConsumerOptions.getTrackLastEnqueuedEventProperties() was not set when creating the consumer. Otherwise, the properties describing the most recently enqueued event in the partition.

close

public void close()
           throws IOException

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

Throws:

IOException