Package com.microsoft.azure.eventhubs.impl

Class EventHubClientImpl

java.lang.Object

com.microsoft.azure.eventhubs.impl.EventHubClientImpl

All Implemented Interfaces:

EventHubClient

public final class EventHubClientImpl
extends Object
implements EventHubClient

Field Summary

Fields

Modifier and Type	Field	Description
protected ScheduledExecutorService	executor
static String	USER_AGENT	It will be truncated to 128 characters

Fields inherited from interface com.microsoft.azure.eventhubs.EventHubClient

DEFAULT_CONSUMER_GROUP_NAME

Method Summary

Modifier and Type	Method	Description
CompletableFuture<Void>	close()
void	closeSync()
static CompletableFuture<EventHubClient>	create(String connectionString, RetryPolicy retryPolicy, ScheduledExecutorService executor, ProxyConfiguration proxyConfiguration)
static CompletableFuture<EventHubClient>	create(String connectionString, RetryPolicy retryPolicy, ScheduledExecutorService executor, ProxyConfiguration proxyConfiguration, Duration watchdogTriggerTime)
static CompletableFuture<EventHubClient>	create(URI endpoint, String eventHubName, ITokenProvider tokenProvider, ScheduledExecutorService executor, EventHubClientOptions options)
EventDataBatch	createBatch(BatchOptions options)	Creates an Empty Collection of EventData.
CompletableFuture<PartitionReceiver>	createEpochReceiver(String consumerGroupName, String partitionId, EventPosition eventPosition, long epoch)	Create a Epoch based EventHub receiver with given partition id and start receiving from the beginning of the partition stream.
CompletableFuture<PartitionReceiver>	createEpochReceiver(String consumerGroupName, String partitionId, EventPosition eventPosition, long epoch, ReceiverOptions receiverOptions)	Create a Epoch based EventHub receiver with given partition id and start receiving from the beginning of the partition stream.
CompletableFuture<PartitionSender>	createPartitionSender(String partitionId)	Create a PartitionSender which can publish EventData's directly to a specific EventHub partition (sender type iii.
CompletableFuture<PartitionReceiver>	createReceiver(String consumerGroupName, String partitionId, EventPosition eventPosition)	Create the EventHub receiver with given partition id and start receiving from the specified starting offset.
CompletableFuture<PartitionReceiver>	createReceiver(String consumerGroupName, String partitionId, EventPosition eventPosition, ReceiverOptions receiverOptions)	Create the EventHub receiver with given partition id and start receiving from the specified starting offset.
String	getClientId()
String	getEventHubName()
protected Exception	getLastKnownError()
CompletableFuture<PartitionRuntimeInformation>	getPartitionRuntimeInformation(String partitionId)	Retrieves dynamic information about a partition of an event hub (see PartitionRuntimeInformation for details.
CompletableFuture<EventHubRuntimeInformation>	getRuntimeInformation()	Retrieves general information about an event hub (see EventHubRuntimeInformation for details).
CompletableFuture<Void>	onClose()
CompletableFuture<Void>	send(EventData data)	Send EventData to EventHub.
CompletableFuture<Void>	send(EventDataBatch eventDatas)	Send EventDataBatch to EventHub.
CompletableFuture<Void>	send(EventData eventData, String partitionKey)	Send an 'EventData with a partitionKey' to EventHub.
CompletableFuture<Void>	send(Iterable<EventData> eventDatas)	Send a batch of EventData to EventHub.
CompletableFuture<Void>	send(Iterable<EventData> eventDatas, String partitionKey)	Send a 'batch of EventData with the same partitionKey' to EventHub.
protected void	setClosed()
protected void	throwIfClosed()

Methods inherited from class java.lang.Object

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

Methods inherited from interface com.microsoft.azure.eventhubs.EventHubClient

close, closeSync, createBatch, createEpochReceiverSync, createEpochReceiverSync, createPartitionSenderSync, createReceiverSync, createReceiverSync, sendSync, sendSync, sendSync, sendSync, sendSync

Field Detail

USER_AGENT

public static String USER_AGENT

It will be truncated to 128 characters

executor

protected final ScheduledExecutorService executor

Method Detail

create

public static CompletableFuture<EventHubClient> create(String connectionString,
                                                       RetryPolicy retryPolicy,
                                                       ScheduledExecutorService executor,
                                                       ProxyConfiguration proxyConfiguration)
                                                throws IOException

Throws:

IOException

create

public static CompletableFuture<EventHubClient> create(String connectionString,
                                                       RetryPolicy retryPolicy,
                                                       ScheduledExecutorService executor,
                                                       ProxyConfiguration proxyConfiguration,
                                                       Duration watchdogTriggerTime)
                                                throws IOException

Throws:

IOException

create

public static CompletableFuture<EventHubClient> create(URI endpoint,
                                                       String eventHubName,
                                                       ITokenProvider tokenProvider,
                                                       ScheduledExecutorService executor,
                                                       EventHubClientOptions options)
                                                throws IOException

Throws:

IOException

getEventHubName

public String getEventHubName()

Specified by:

getEventHubName in interface EventHubClient

Returns:

the name of the Event Hub this client is connected to.

createBatch

public EventDataBatch createBatch(BatchOptions options)
                           throws EventHubException

Description copied from interface: EventHubClient

Creates an Empty Collection of EventData. The same partitionKey must be used while sending these events using EventHubClient.send(EventDataBatch).

Specified by:

createBatch in interface EventHubClient

Parameters:

options - see BatchOptions for more details

Returns:

the empty EventDataBatch, after negotiating maximum message size with EventHubs service

Throws:

EventHubException - if the Microsoft Azure Event Hubs service encountered problems during the operation.

send

public CompletableFuture<Void> send(EventData data)

Description copied from interface: EventHubClient

Send EventData to EventHub. The sent EventData will land on any arbitrarily chosen EventHubs partition.

There are 3 ways to send to EventHubs, each exposed as a method (along with its sendBatch overload):

• EventHubClient.send(EventData), EventHubClient.send(Iterable), or EventHubClient.send(EventDataBatch)
• EventHubClient.send(EventData, String) or EventHubClient.send(Iterable, String)
• PartitionSender.send(EventData), PartitionSender.send(Iterable), or PartitionSender.send(EventDataBatch)

Use this method to Send, if:

 a)  the send(EventData) operation should be highly available and
 b)  the data needs to be evenly distributed among all partitions; exception being, when a subset of partitions are unavailable
 

EventHubClient.send(EventData) send's the EventData to a Service Gateway, which in-turn will forward the EventData to one of the EventHubs' partitions. Here's the message forwarding algorithm:

 i.  Forward the EventData's to EventHub partitions, by equally distributing the data among all partitions (ex: Round-robin the EventData's to all EventHubs' partitions)
 ii. If one of the EventHub partitions is unavailable for a moment, the Service Gateway will automatically detect it and forward the message to another available partition - making the Send operation highly-available.
 

Specified by:

send in interface EventHubClient

Parameters:

data - the EventData to be sent.

Returns:

a CompletableFuture that can be completed when the send operations is done..

See Also:

EventHubClient.send(EventData, String), PartitionSender.send(EventData)

send

public CompletableFuture<Void> send(Iterable<EventData> eventDatas)

Description copied from interface: EventHubClient

Send a batch of EventData to EventHub. The sent EventData will land on any arbitrarily chosen EventHubs partition. This is the most recommended way to Send to EventHubs.

There are 3 ways to send to EventHubs, to understand this particular type of Send refer to the overload EventHubClient.send(EventData), which is used to send single EventData. Use this overload versus EventHubClient.send(EventData), if you need to send a batch of EventData.

Sending a batch of EventData's is useful in the following cases:

 i.   Efficient send - sending a batch of EventData maximizes the overall throughput by optimally using the number of sessions created to EventHubs' service.
 ii.  Send multiple EventData's in a Transaction. To achieve ACID properties, the Gateway Service will forward all EventData's in the batch to a single EventHubs' partition.
 

Sample code (sample uses sync version of the api but concept are identical):

 Gson gson = new GsonBuilder().create();
 EventHubClient client = EventHubClient.createSync("__connection__");

 while (true)
 {
     LinkedList<EventData> events = new LinkedList<EventData>();}
     for (int count = 1; count < 11; count++)
     {
         PayloadEvent payload = new PayloadEvent(count);
         byte[] payloadBytes = gson.toJson(payload).getBytes(Charset.defaultCharset());
         EventData sendEvent = new EventData(payloadBytes);
         sendEvent.getProperties().put("from", "javaClient");
         events.add(sendEvent);
     }

     client.sendSync(events);
     System.out.println(String.format("Sent Batch... Size: %s", events.size()));
 }
 

for Exceptions refer to EventHubClient.sendSync(Iterable)

Specified by:

send in interface EventHubClient

Parameters:

eventDatas - batch of events to send to EventHub

Returns:

a CompletableFuture that can be completed when the send operations is done..

See Also:

EventHubClient.send(EventData, String), PartitionSender.send(EventData)

send

public CompletableFuture<Void> send(EventDataBatch eventDatas)

Description copied from interface: EventHubClient

Send EventDataBatch to EventHub. The sent EventDataBatch will land according the partition key set in the EventDataBatch. If a partition key is not set, then we will Round-robin the EventData's to all EventHubs' partitions.

Specified by:

send in interface EventHubClient

Parameters:

eventDatas - EventDataBatch to send to EventHub

Returns:

a CompleteableFuture that can be completed when the send operations are done

See Also:

EventHubClient.send(Iterable), EventDataBatch

send

public CompletableFuture<Void> send(EventData eventData,
                                    String partitionKey)

Description copied from interface: EventHubClient

Send an 'EventData with a partitionKey' to EventHub. All EventData's with a partitionKey are guaranteed to land on the same partition. This send pattern emphasize data correlation over general availability and latency.

There are 3 ways to send to EventHubs, each exposed as a method (along with its sendBatch overload):

 i.   EventHubClient.send(EventData) or EventHubClient.send(Iterable)
 ii.  EventHubClient.send(EventData, String) or EventHubClient.send(Iterable, String)
 iii. PartitionSender.send(EventData) or PartitionSender.send(Iterable)
 

Use this type of Send, if:

 i.  There is a need for correlation of events based on Sender instance; The sender can generate a UniqueId and set it as partitionKey - which on the received Message can be used for correlation
 ii. The client wants to take control of distribution of data across partitions.
 

Multiple PartitionKey's could be mapped to one Partition. EventHubs service uses a proprietary Hash algorithm to map the PartitionKey to a PartitionId. Using this type of Send (Sending using a specific partitionKey), could sometimes result in partitions which are not evenly distributed.

Specified by:

send in interface EventHubClient

Parameters:

eventData - the EventData to be sent.

partitionKey - the partitionKey will be hash'ed to determine the partitionId to send the eventData to. On the Received message this can be accessed at EventData.SystemProperties.getPartitionKey()

Returns:

a CompletableFuture that can be completed when the send operations is done..

See Also:

EventHubClient.send(EventData), PartitionSender.send(EventData)

send

public CompletableFuture<Void> send(Iterable<EventData> eventDatas,
                                    String partitionKey)

Description copied from interface: EventHubClient

Send a 'batch of EventData with the same partitionKey' to EventHub. All EventData's with a partitionKey are guaranteed to land on the same partition. Multiple PartitionKey's will be mapped to one Partition.

There are 3 ways to send to EventHubs, to understand this particular type of Send refer to the overload EventHubClient.send(EventData, String), which is the same type of Send and is used to send single EventData.

Sending a batch of EventData's is useful in the following cases:

 i.   Efficient send - sending a batch of EventData maximizes the overall throughput by optimally using the number of sessions created to EventHubs service.
 ii.  Send multiple events in One Transaction. This is the reason why all events sent in a batch needs to have same partitionKey (so that they are sent to one partition only).
 

Specified by:

send in interface EventHubClient

Parameters:

eventDatas - the batch of events to send to EventHub

partitionKey - the partitionKey will be hash'ed to determine the partitionId to send the eventData to. On the Received message this can be accessed at EventData.SystemProperties.getPartitionKey()

Returns:

a CompletableFuture that can be completed when the send operations is done..

See Also:

EventHubClient.send(EventData), PartitionSender.send(EventData)

createPartitionSender

public CompletableFuture<PartitionSender> createPartitionSender(String partitionId)
                                                         throws EventHubException

Description copied from interface: EventHubClient

Create a PartitionSender which can publish EventData's directly to a specific EventHub partition (sender type iii. in the below list).

There are 3 patterns/ways to send to EventHubs:

 i.   EventHubClient.send(EventData) or EventHubClient.send(Iterable)
 ii.  EventHubClient.send(EventData, String) or EventHubClient.send(Iterable, String)
 iii. PartitionSender.send(EventData) or PartitionSender.send(Iterable)
 

Specified by:

createPartitionSender in interface EventHubClient

Parameters:

partitionId - partitionId of EventHub to send the EventData's to

Returns:

a CompletableFuture that would result in a PartitionSenderImpl when it is completed.

Throws:

EventHubException - if Service Bus service encountered problems during connection creation.

See Also:

PartitionSender

createReceiver

public CompletableFuture<PartitionReceiver> createReceiver(String consumerGroupName,
                                                           String partitionId,
                                                           EventPosition eventPosition)
                                                    throws EventHubException

Description copied from interface: EventHubClient

Create the EventHub receiver with given partition id and start receiving from the specified starting offset. The receiver is created for a specific EventHub Partition from the specific consumer group.

Specified by:

createReceiver in interface EventHubClient

Parameters:

consumerGroupName - the consumer group name that this receiver should be grouped under.

partitionId - the partition Id that the receiver belongs to. All data received will be from this partition only.

eventPosition - the position to start receiving the events from. See EventPosition

Returns:

a CompletableFuture that would result in a PartitionReceiver instance when it is completed.

Throws:

EventHubException - if Service Bus service encountered problems during the operation.

See Also:

PartitionReceiver

createReceiver

public CompletableFuture<PartitionReceiver> createReceiver(String consumerGroupName,
                                                           String partitionId,
                                                           EventPosition eventPosition,
                                                           ReceiverOptions receiverOptions)
                                                    throws EventHubException

Description copied from interface: EventHubClient

Create the EventHub receiver with given partition id and start receiving from the specified starting offset. The receiver is created for a specific EventHub Partition from the specific consumer group.

Specified by:

createReceiver in interface EventHubClient

Parameters:

consumerGroupName - the consumer group name that this receiver should be grouped under.

partitionId - the partition Id that the receiver belongs to. All data received will be from this partition only.

eventPosition - the position to start receiving the events from. See EventPosition

receiverOptions - the set of options to enable on the event hubs receiver

Returns:

a CompletableFuture that would result in a PartitionReceiver instance when it is completed.

Throws:

EventHubException - if Service Bus service encountered problems during the operation.

See Also:

PartitionReceiver

createEpochReceiver

public CompletableFuture<PartitionReceiver> createEpochReceiver(String consumerGroupName,
                                                                String partitionId,
                                                                EventPosition eventPosition,
                                                                long epoch)
                                                         throws EventHubException

Description copied from interface: EventHubClient

Create a Epoch based EventHub receiver with given partition id and start receiving from the beginning of the partition stream. The receiver is created for a specific EventHub Partition from the specific consumer group.

It is important to pay attention to the following when creating epoch based receiver:

• Ownership enforcement - Once you created an epoch based receiver, you cannot create a non-epoch receiver to the same consumerGroup-Partition combo until all receivers to the combo are closed.
• Ownership stealing - If a receiver with higher epoch value is created for a consumerGroup-Partition combo, any older epoch receiver to that combo will be force closed.
• Any receiver closed due to lost of ownership to a consumerGroup-Partition combo will get ReceiverDisconnectedException for all operations from that receiver.

Specified by:

createEpochReceiver in interface EventHubClient

Parameters:

consumerGroupName - the consumer group name that this receiver should be grouped under.

partitionId - the partition Id that the receiver belongs to. All data received will be from this partition only.

eventPosition - the position to start receiving the events from. See EventPosition

epoch - an unique identifier (epoch value) that the service uses, to enforce partition/lease ownership.

Returns:

a CompletableFuture that would result in a PartitionReceiver when it is completed.

Throws:

EventHubException - if Service Bus service encountered problems during the operation.

See Also:

PartitionReceiver, ReceiverDisconnectedException

createEpochReceiver

public CompletableFuture<PartitionReceiver> createEpochReceiver(String consumerGroupName,
                                                                String partitionId,
                                                                EventPosition eventPosition,
                                                                long epoch,
                                                                ReceiverOptions receiverOptions)
                                                         throws EventHubException

Description copied from interface: EventHubClient

Create a Epoch based EventHub receiver with given partition id and start receiving from the beginning of the partition stream. The receiver is created for a specific EventHub Partition from the specific consumer group.

It is important to pay attention to the following when creating epoch based receiver:

• Ownership enforcement - Once you created an epoch based receiver, you cannot create a non-epoch receiver to the same consumerGroup-Partition combo until all receivers to the combo are closed.
• Ownership stealing - If a receiver with higher epoch value is created for a consumerGroup-Partition combo, any older epoch receiver to that combo will be force closed.
• Any receiver closed due to lost of ownership to a consumerGroup-Partition combo will get ReceiverDisconnectedException for all operations from that receiver.

Specified by:

createEpochReceiver in interface EventHubClient

Parameters:

consumerGroupName - the consumer group name that this receiver should be grouped under.

partitionId - the partition Id that the receiver belongs to. All data received will be from this partition only.

eventPosition - the position to start receiving the events from. See EventPosition

epoch - an unique identifier (epoch value) that the service uses, to enforce partition/lease ownership.

receiverOptions - the set of options to enable on the event hubs receiver

Returns:

a CompletableFuture that would result in a PartitionReceiver when it is completed.

Throws:

EventHubException - if Service Bus service encountered problems during the operation.

See Also:

PartitionReceiver, ReceiverDisconnectedException

onClose

public CompletableFuture<Void> onClose()

getRuntimeInformation

public CompletableFuture<EventHubRuntimeInformation> getRuntimeInformation()

Description copied from interface: EventHubClient

Retrieves general information about an event hub (see EventHubRuntimeInformation for details). Retries until it reaches the operation timeout, then either rethrows the last error if available or returns null to indicate timeout.

Specified by:

getRuntimeInformation in interface EventHubClient

Returns:

CompletableFuture which returns an EventHubRuntimeInformation on success, or null on timeout.

getPartitionRuntimeInformation

public CompletableFuture<PartitionRuntimeInformation> getPartitionRuntimeInformation(String partitionId)

Description copied from interface: EventHubClient

Retrieves dynamic information about a partition of an event hub (see PartitionRuntimeInformation for details. Retries until it reaches the operation timeout, then either rethrows the last error if available or returns null to indicate timeout.

Specified by:

getPartitionRuntimeInformation in interface EventHubClient

Parameters:

partitionId - Partition to get information about. Must be one of the partition ids returned by EventHubClient.getRuntimeInformation().

Returns:

CompletableFuture which returns an PartitionRuntimeInformation on success, or null on timeout.

getClientId

public String getClientId()

setClosed

protected final void setClosed()

close

public final CompletableFuture<Void> close()

closeSync

public final void closeSync()
                     throws EventHubException

Throws:

EventHubException

throwIfClosed

protected final void throwIfClosed()

getLastKnownError

protected Exception getLastKnownError()