createEventHubClient() throws EventHubException, IOException {
+ return EventHubClient.createWithTokenProvider(this.endpoint, this.eventHubPath, this.tokenProvider, this.executor, this.options);
+ }
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHost.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHost.java
index c9eaf61e266d..5fc9acbd1e71 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHost.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/EventProcessorHost.java
@@ -3,15 +3,26 @@
package com.microsoft.azure.eventprocessorhost;
+import com.microsoft.azure.eventhubs.AzureActiveDirectoryTokenProvider;
+import com.microsoft.azure.eventhubs.AzureActiveDirectoryTokenProvider.AuthenticationCallback;
import com.microsoft.azure.eventhubs.ConnectionStringBuilder;
+import com.microsoft.azure.eventhubs.EventHubClientOptions;
+import com.microsoft.azure.eventhubs.ITokenProvider;
import com.microsoft.azure.eventhubs.RetryPolicy;
+import com.microsoft.azure.eventhubs.TransportType;
+import com.microsoft.azure.eventhubs.impl.StringUtil;
+import com.microsoft.azure.storage.StorageCredentials;
import com.microsoft.azure.storage.StorageException;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.net.URI;
import java.net.URISyntaxException;
import java.security.InvalidKeyException;
+import java.time.Duration;
import java.util.Locale;
+import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -39,244 +50,24 @@ public final class EventProcessorHost {
private PartitionManager partitionManager;
private PartitionManagerOptions partitionManagerOptions = null;
- /**
- * Create a new host instance to process events from an Event Hub.
- *
- * Since Event Hubs are generally used for scale-out, high-traffic scenarios, in most scenarios there will
- * be only one host instances per process, and the processes will be run on separate machines. Besides scale, this also
- * provides isolation: one process or machine crashing will not take out multiple host instances. However, it is
- * supported to run multiple host instances on one machine, or even within one process, for development and testing.
- *
- * The hostName parameter is a name for this event processor host, which must be unique among all event processor host instances
- * receiving from this event hub+consumer group combination: the unique name is used to distinguish which event processor host
- * instance owns the lease for a given partition. An easy way to generate a unique hostName which also includes
- * other information is to call EventProcessorHost.createHostName("mystring").
- *
- * This overload of the constructor uses the built-in lease and checkpoint managers. The
- * Azure Storage account specified by the storageConnectionString parameter is used by the built-in
- * managers to record leases and checkpoints, in the specified container.
- *
- * The Event Hub connection string may be conveniently constructed using the ConnectionStringBuilder class
- * from the Java Event Hub client.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param storageConnectionString Connection string for the Azure Storage account to use for persisting leases and checkpoints.
- * @param storageContainerName Azure Storage container name for use by built-in lease and checkpoint manager.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
- final String storageConnectionString,
- final String storageContainerName) {
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString, storageConnectionString, storageContainerName, (ScheduledExecutorService) null);
- }
-
- /**
- * Create a new host to process events from an Event Hub.
- *
- * This overload adds an argument to specify a user-provided thread pool. The number of partitions in the
- * target event hub and the number of host instances should be considered when choosing the size of the thread pool:
- * how many partitions is one instance expected to own under normal circumstances? One thread per partition should
- * provide good performance, while being able to support more partitions adequately if a host instance fails and its
- * partitions must be redistributed.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param storageConnectionString Connection string for the Azure Storage account to use for persisting leases and checkpoints.
- * @param storageContainerName Azure Storage container name for use by built-in lease and checkpoint manager.
- * @param executorService User-supplied thread executor, or null to use EventProcessorHost-internal executor.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
- final String storageConnectionString,
- final String storageContainerName,
- final ScheduledExecutorService executorService) {
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString, storageConnectionString, storageContainerName, (String) null, executorService);
- }
-
- /**
- * Create a new host to process events from an Event Hub.
- *
- * This overload adds an argument to specify a prefix used by the built-in lease manager when naming blobs in Azure Storage.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param storageConnectionString Connection string for the Azure Storage account to use for persisting leases and checkpoints.
- * @param storageContainerName Azure Storage container name for use by built-in lease and checkpoint manager.
- * @param storageBlobPrefix Prefix used when naming blobs within the storage container.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
- final String storageConnectionString,
- final String storageContainerName,
- final String storageBlobPrefix) {
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString, storageConnectionString, storageContainerName, storageBlobPrefix,
- (ScheduledExecutorService) null);
- }
-
- /**
- * Create a new host to process events from an Event Hub.
- *
- * This overload allows the caller to specify both a user-supplied thread pool and
- * a prefix used by the built-in lease manager when naming blobs in Azure Storage.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param storageConnectionString Connection string for the Azure Storage account to use for persisting leases and checkpoints.
- * @param storageContainerName Azure Storage container name for use by built-in lease and checkpoint manager.
- * @param storageBlobPrefix Prefix used when naming blobs within the storage container.
- * @param executorService User-supplied thread executor, or null to use EventProcessorHost-internal executor.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
- final String storageConnectionString,
- final String storageContainerName,
- final String storageBlobPrefix,
- final ScheduledExecutorService executorService) {
- // Would like to check storageConnectionString and storageContainerName here but can't, because Java doesn't allow statements before
- // calling another constructor. storageBlobPrefix is allowed to be null or empty, doesn't need checking.
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString,
- new AzureStorageCheckpointLeaseManager(storageConnectionString, storageContainerName, storageBlobPrefix), executorService);
- this.initializeLeaseManager = true;
- this.partitionManagerOptions = new AzureStoragePartitionManagerOptions();
- }
-
- // Because Java won't let you do ANYTHING before calling another constructor. In particular, you can't
- // new up an object and pass it as TWO parameters of the other constructor.
private EventProcessorHost(
final String hostName,
final String eventHubPath,
final String consumerGroupName,
- final String eventHubConnectionString,
- final AzureStorageCheckpointLeaseManager combinedManager,
- final ScheduledExecutorService executorService) {
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString, combinedManager, combinedManager, executorService, null);
- }
-
- /**
- * Create a new host to process events from an Event Hub.
- *
- * This overload allows the caller to provide their own lease and checkpoint managers to replace the built-in
- * ones based on Azure Storage.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param checkpointManager Implementation of ICheckpointManager, to be replacement checkpoint manager.
- * @param leaseManager Implementation of ILeaseManager, to be replacement lease manager.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
- ICheckpointManager checkpointManager,
- ILeaseManager leaseManager) {
- this(hostName, eventHubPath, consumerGroupName, eventHubConnectionString, checkpointManager, leaseManager, null, null);
- }
-
- /**
- * Create a new host to process events from an Event Hub.
- *
- * This overload allows the caller to provide their own lease and checkpoint managers to replace the built-in
- * ones based on Azure Storage, and to provide an executor service and a retry policy for communications with the event hub.
- *
- * @param hostName A name for this event processor host. See method notes.
- * @param eventHubPath Specifies the Event Hub to receive events from.
- * @param consumerGroupName The name of the consumer group to use when receiving from the Event Hub.
- * @param eventHubConnectionString Connection string for the Event Hub to receive from.
- * @param checkpointManager Implementation of ICheckpointManager, to be replacement checkpoint manager.
- * @param leaseManager Implementation of ILeaseManager, to be replacement lease manager.
- * @param executorService User-supplied thread executor, or null to use EventProcessorHost-internal executor.
- * @param retryPolicy Retry policy governing communications with the event hub.
- */
- public EventProcessorHost(
- final String hostName,
- final String eventHubPath,
- final String consumerGroupName,
- final String eventHubConnectionString,
+ final EventHubClientFactory eventHubClientFactory,
ICheckpointManager checkpointManager,
ILeaseManager leaseManager,
- ScheduledExecutorService executorService,
- RetryPolicy retryPolicy) {
- if ((hostName == null) || hostName.isEmpty()) {
- throw new IllegalArgumentException("hostName argument must not be null or empty string");
- }
-
- // eventHubPath is allowed to be null or empty if it is provided in the connection string. That will be checked later.
- if ((consumerGroupName == null) || consumerGroupName.isEmpty()) {
- throw new IllegalArgumentException("consumerGroupName argument must not be null or empty");
- }
-
- if ((eventHubConnectionString == null) || eventHubConnectionString.isEmpty()) {
- throw new IllegalArgumentException("eventHubConnectionString argument must not be null or empty");
- }
-
- // The event hub path must appear in at least one of the eventHubPath argument or the connection string.
- // If it appears in both, then it must be the same in both. If it appears in only one, populate the other.
- ConnectionStringBuilder providedCSB = new ConnectionStringBuilder(eventHubConnectionString);
- String extractedEntityPath = providedCSB.getEventHubName();
- String effectiveEventHubPath = eventHubPath;
- String effectiveEventHubConnectionString = eventHubConnectionString;
- if ((effectiveEventHubPath != null) && !effectiveEventHubPath.isEmpty()) {
- if (extractedEntityPath != null) {
- if (effectiveEventHubPath.compareTo(extractedEntityPath) != 0) {
- throw new IllegalArgumentException("Provided EventHub path in eventHubPath parameter conflicts with the path in provided EventHub connection string");
- }
- // else they are the same and that's fine
- } else {
- // There is no entity path in the connection string, so put it there.
- ConnectionStringBuilder rebuildCSB = new ConnectionStringBuilder()
- .setEndpoint(providedCSB.getEndpoint())
- .setEventHubName(effectiveEventHubPath)
- .setSasKeyName(providedCSB.getSasKeyName())
- .setSasKey(providedCSB.getSasKey());
- rebuildCSB.setOperationTimeout(providedCSB.getOperationTimeout());
- effectiveEventHubConnectionString = rebuildCSB.toString();
- }
+ final boolean initializeLeaseManager,
+ ScheduledExecutorService executorService) {
+ this.initializeLeaseManager = initializeLeaseManager;
+ if (this.initializeLeaseManager) {
+ this.partitionManagerOptions = new AzureStoragePartitionManagerOptions();
} else {
- if ((extractedEntityPath != null) && !extractedEntityPath.isEmpty()) {
- effectiveEventHubPath = extractedEntityPath;
- } else {
- throw new IllegalArgumentException("Provide EventHub entity path in either eventHubPath argument or in eventHubConnectionString");
- }
+ // Using user-supplied implementation.
+ // Establish generic defaults in case the user doesn't provide an options object.
+ this.partitionManagerOptions = new PartitionManagerOptions();
}
- if (checkpointManager == null) {
- throw new IllegalArgumentException("Must provide an object which implements ICheckpointManager");
- }
- if (leaseManager == null) {
- throw new IllegalArgumentException("Must provide an object which implements ILeaseManager");
- }
-
- // executorService argument is allowed to be null, that is the indication to use an internal threadpool.
-
- // Normally will not be null because we're using the AzureStorage implementation.
- // If it is null, we're using user-supplied implementation. Establish generic defaults
- // in case the user doesn't provide an options object.
- this.partitionManagerOptions = new PartitionManagerOptions();
-
if (executorService != null) {
// User has supplied an ExecutorService, so use that.
this.weOwnExecutor = false;
@@ -285,12 +76,13 @@ public EventProcessorHost(
this.weOwnExecutor = true;
this.executorService = Executors.newScheduledThreadPool(
this.executorServicePoolSize,
- new EventProcessorHostThreadPoolFactory(hostName, effectiveEventHubPath, consumerGroupName));
+ new EventProcessorHostThreadPoolFactory(hostName, eventHubPath, consumerGroupName));
}
+ eventHubClientFactory.setExecutor(this.executorService);
this.hostContext = new HostContext(this.executorService,
this, hostName,
- effectiveEventHubPath, consumerGroupName, effectiveEventHubConnectionString, retryPolicy,
+ eventHubPath, consumerGroupName, eventHubClientFactory,
leaseManager, checkpointManager);
this.partitionManager = new PartitionManager(hostContext);
@@ -571,4 +363,362 @@ public void uncaughtException(Thread t, Throwable e) {
}
}
}
+
+ /**
+ * Builder class to create EventProcessorHost instances.
+ *
+ * To use, start with: EventProcessorHost.EventProcessorHostBuilder.newBuilder(...)
+ * Then either use the built-in Azure Storage-based lease and checkpoint managers, or user implementations.
+ * Then either supply an Event Hub connection string or use Azure Active Directory (AAD) authentication.
+ * If using AAD auth, either provide a callback or an ITokenProvider
+ * Finally, set various optional values as desired, then call build() to get an EventProcessorHost instance.
+ */
+ public static class EventProcessorHostBuilder {
+ /**
+ * The process of building starts here, with arguments that are always required.
+ *
+ * The hostName parameter is a name for this EventProcessorHost instance, which must be unique among
+ * all instances consuming from the same Event Hub and consumer group. The name must be unique because
+ * it is used to distinguish which instance owns the lease for a given partition of the event hub. An
+ * easy way to generate a unique host name is to call EventProcessorHost.createHostName("mystring").
+ *
+ * @param hostName a name for this host instance. See method notes.
+ * @param consumerGroupName the consumer group on the Event Hub
+ * @return interface for setting the lease and checkpoint managers
+ */
+ public static ManagerStep newBuilder(final String hostName, final String consumerGroupName) {
+ return new Steps(hostName, consumerGroupName);
+ }
+
+ private EventProcessorHostBuilder() {
+ }
+
+ public static interface ManagerStep {
+ /**
+ * Use the built-in Azure Storage-based lease and checkpoint managers.
+ *
+ * @param storageConnectionString connection string for an Azure Storage account
+ * @param storageContainerName name for the blob container within the Storage account
+ * @param storageBlobPrefix prefix for the names of the blobs within the blob container, can be empty or null
+ * @return interface for setting the Event Hub connection info and auth
+ */
+ AuthStep useAzureStorageCheckpointLeaseManager(String storageConnectionString, String storageContainerName, String storageBlobPrefix);
+
+ /**
+ * Use the built-in Azure Storage-based lease and checkpoint managers.
+ *
+ * @param storageCredentials credentials for an Azure Storage account, such as an AAD token
+ * @param storageContainerName name for the blob container within the Storage account
+ * @param storageBlobPrefix prefix for the names of the blobs within the blob container, can be empty or null
+ * @return interface for setting the Event Hub connection info and auth
+ */
+ AuthStep useAzureStorageCheckpointLeaseManager(StorageCredentials storageCredentials, String storageContainerName, String storageBlobPrefix);
+
+ /**
+ * Use user-implemented lease and checkpoint managers.
+ *
+ * @param checkpointManager user-supplied implementation of {@link ICheckpointManager}
+ * @param leaseManager user-supplied implementation of {@link ILeaseManager}
+ * @return interface for setting the Event Hub connection info and auth
+ */
+ AuthStep useUserCheckpointAndLeaseManagers(ICheckpointManager checkpointManager, ILeaseManager leaseManager);
+ }
+
+ public static interface AuthStep {
+ /**
+ * Azure Portal can provide a connection string with auth information that applies only to one
+ * individual Event Hub. In that case, the connection string contains the name of the Event Hub.
+ *
+ * @param eventHubConnectionString Event Hub connection string (which contains the name of the Event Hub)
+ * @return interface for setting optional values
+ */
+ OptionalStep useEventHubConnectionString(String eventHubConnectionString);
+
+ /**
+ * Azure Portal can provide a connection string with auth information that applies to the entire
+ * namespace instead of an individual Event Hub. Use this overload with such a connection string,
+ * which requires you to specify the name of the Event Hub separately.
+ *
+ * @param eventHubConnectionString Event Hub connection string (which does not contain the name of the Event Hub)
+ * @param eventHubPath name of the Event Hub
+ * @return interface for setting optional values
+ */
+ OptionalStep useEventHubConnectionString(String eventHubConnectionString, String eventHubPath);
+
+ /**
+ * When using AAD auth, call this method to specify the Event Hub, then add AAD-based auth information in the next step.
+ *
+ * @param endpoint URI of the Event Hub namespace
+ * @param eventHubPath name of the Event Hub
+ * @return interface for setting AAD auth info
+ */
+ AADAuthStep useAADAuthentication(URI endpoint, String eventHubPath);
+ }
+
+ public static interface AADAuthStep {
+ /**
+ * Provide a callback which will be called when a token is needed. See {@link AzureActiveDirectoryTokenProvider}
+ *
+ * @param authCallback the callback
+ * @param authority AAD authority string which will be passed to the callback. Used for national cloud support.
+ * @return interface for setting optional values
+ */
+ OptionalStep useAuthenticationCallback(AuthenticationCallback authCallback, String authority);
+
+ /**
+ * Provide a user-implemented token provider which will be called when a token is needed.
+ *
+ * @param tokenProvider user implementation of ITokenProvider
+ * @return interface for setting optional values
+ */
+ OptionalStep useTokenProvider(ITokenProvider tokenProvider);
+ }
+
+ public static interface OptionalStep {
+ /**
+ * Event Processor Host runs tasks on the supplied threadpool, or creates an internal one.
+ * @param executor threadpool, or null to use an internal one
+ * @return interface for setting optional values
+ */
+ OptionalStep setExecutor(ScheduledExecutorService executor);
+
+ /**
+ * {@link RetryPolicy} for Event Hubs operations. Event Processor Host uses RetryPolicy.getDefault()
+ * if none is supplied.
+ *
+ * @param retryPolicy desired retry policy
+ * @return interface for setting optional values
+ */
+ OptionalStep setRetryPolicy(RetryPolicy retryPolicy);
+
+ /**
+ * {@link TransportType} for connections to the Event Hubs service. Defaults to TransportType.AMQP.
+ * The transport type can also be set in the Event Hub connection string. The value set here will
+ * override the value in the connection string, if any.
+ *
+ * @param transportType desired transport type
+ * @return interface for setting optional values
+ */
+ OptionalStep setTransportType(TransportType transportType);
+
+ /**
+ * The timeout for Event Hubs operations. Defaults to MessagingFactory.DefaultOperationTimeout.
+ * The timeout can also be set in the Event Hub connection string. The value set here will override
+ * the value in the connection string, if any.
+ *
+ * @param operationTimeout desired timeout
+ * @return interface for setting optional values
+ */
+ OptionalStep setOperationTimeout(Duration operationTimeout);
+
+ /**
+ * After setting all desired optional values, call this method to build an EventProcessorHost instance.
+ *
+ * @return new EventProcessorHost instance
+ */
+ EventProcessorHost build();
+ }
+
+ private static class Steps implements ManagerStep, AuthStep, AADAuthStep, OptionalStep {
+ private final String hostName;
+ private final String consumerGroupName;
+
+ // OptionalStep
+ private ScheduledExecutorService executor = null;
+ private RetryPolicy retryPolicy = null;
+ private TransportType transportType = null;
+ private Duration operationTimeout = null;
+
+ // Auth steps
+ private String eventHubConnectionString = null; // group 1
+ private String eventHubPath = null; // optional for group 1, required for groups 2-3
+ private URI endpoint = null; // groups 2-3
+ private AuthenticationCallback authCallback = null; // group 2
+ private String authority = null; // group 2
+ private ITokenProvider tokenProvider = null; // group 3
+
+ // ManagerStep
+ private ICheckpointManager checkpointManager;
+ private ILeaseManager leaseManager;
+ private boolean initializeManagers = false;
+
+
+ public Steps(final String hostName, final String consumerGroupName) {
+ if (StringUtil.isNullOrWhiteSpace(hostName) || StringUtil.isNullOrWhiteSpace(consumerGroupName)) {
+ throw new IllegalArgumentException("hostName and consumerGroupName cannot be null or empty");
+ }
+
+ this.hostName = hostName;
+ this.consumerGroupName = consumerGroupName;
+ }
+
+ @Override
+ public OptionalStep setExecutor(final ScheduledExecutorService executor) {
+ // executor is allowed to be null, causes EPH to create and use an internal one
+ this.executor = executor;
+ return this;
+ }
+
+ @Override
+ public OptionalStep setRetryPolicy(final RetryPolicy retryPolicy) {
+ this.retryPolicy = retryPolicy;
+ return this;
+ }
+
+ @Override
+ public OptionalStep setTransportType(final TransportType transportType) {
+ Objects.requireNonNull(transportType);
+
+ this.transportType = transportType;
+ return this;
+ }
+
+ @Override
+ public OptionalStep setOperationTimeout(final Duration operationTimeout) {
+ Objects.requireNonNull(operationTimeout);
+
+ this.operationTimeout = operationTimeout;
+ return this;
+ }
+
+ @Override
+ public OptionalStep useAuthenticationCallback(final AuthenticationCallback authCallback, final String authority) {
+ Objects.requireNonNull(authCallback);
+ if (StringUtil.isNullOrWhiteSpace(authority)) {
+ throw new IllegalArgumentException("authority cannot be null or empty");
+ }
+
+ this.authCallback = authCallback;
+ this.authority = authority;
+ return this;
+ }
+
+ @Override
+ public OptionalStep useTokenProvider(final ITokenProvider tokenProvider) {
+ Objects.requireNonNull(tokenProvider);
+
+ this.tokenProvider = tokenProvider;
+ return this;
+ }
+
+ @Override
+ public OptionalStep useEventHubConnectionString(final String eventHubConnectionString) {
+ return useEventHubConnectionString(eventHubConnectionString, null);
+ }
+
+ @Override
+ public OptionalStep useEventHubConnectionString(final String eventHubConnectionString, final String eventHubPath) {
+ if (StringUtil.isNullOrWhiteSpace(eventHubConnectionString)) {
+ throw new IllegalArgumentException("eventHubConnectionString cannot be null or empty");
+ }
+ if ((eventHubPath != null) && StringUtil.isNullOrWhiteSpace(eventHubPath)) {
+ throw new IllegalArgumentException("eventHubPath cannot be empty. Use null if the connection string already contains the path.");
+ }
+
+ this.eventHubConnectionString = eventHubConnectionString;
+ this.eventHubPath = eventHubPath;
+ return this;
+ }
+
+ @Override
+ public AADAuthStep useAADAuthentication(final URI endpoint, final String eventHubPath) {
+ Objects.requireNonNull(endpoint);
+ if (StringUtil.isNullOrWhiteSpace(eventHubPath)) {
+ throw new IllegalArgumentException("eventHubPath cannot be null or empty");
+ }
+
+ this.endpoint = endpoint;
+ this.eventHubPath = eventHubPath;
+ return this;
+ }
+
+ @Override
+ public AuthStep useAzureStorageCheckpointLeaseManager(final String storageConnectionString,
+ final String storageContainerName, final String storageBlobPrefix) {
+ AzureStorageCheckpointLeaseManager mgr = new AzureStorageCheckpointLeaseManager(storageConnectionString, storageContainerName, storageBlobPrefix);
+ this.initializeManagers = true;
+ return useUserCheckpointAndLeaseManagers(mgr, mgr);
+ }
+
+ @Override
+ public AuthStep useAzureStorageCheckpointLeaseManager(final StorageCredentials storageCredentials,
+ final String storageContainerName, final String storageBlobPrefix) {
+ AzureStorageCheckpointLeaseManager mgr = new AzureStorageCheckpointLeaseManager(storageCredentials, storageContainerName, storageBlobPrefix);
+ this.initializeManagers = true;
+ return useUserCheckpointAndLeaseManagers(mgr, mgr);
+ }
+
+ @Override
+ public AuthStep useUserCheckpointAndLeaseManagers(final ICheckpointManager checkpointManager,
+ final ILeaseManager leaseManager) {
+ Objects.requireNonNull(checkpointManager);
+ Objects.requireNonNull(leaseManager);
+
+ this.checkpointManager = checkpointManager;
+ this.leaseManager = leaseManager;
+ return this;
+ }
+
+ @Override
+ public EventProcessorHost build() {
+ // One of these conditions MUST be true. Can't get to the OptionalStep interface where build() is available
+ // without setting one of the auth options.
+ EventHubClientFactory ehcFactory = null;
+ if (this.eventHubConnectionString != null) {
+ normalizeConnectionStringAndEventHubPath();
+ ehcFactory = new EventHubClientFactory.EHCFWithConnectionString(this.eventHubConnectionString, this.retryPolicy);
+ } else if (this.authCallback != null) {
+ ehcFactory = new EventHubClientFactory.EHCFWithAuthCallback(this.endpoint, this.eventHubPath,
+ this.authCallback, this.authority, packOptions());
+ } else if (this.tokenProvider != null) {
+ ehcFactory = new EventHubClientFactory.EHCFWithTokenProvider(this.endpoint, this.eventHubPath, this.tokenProvider, packOptions());
+ }
+ return new EventProcessorHost(this.hostName,
+ this.eventHubPath,
+ this.consumerGroupName,
+ ehcFactory,
+ this.checkpointManager,
+ this.leaseManager,
+ this.initializeManagers,
+ this.executor);
+ }
+
+ private EventHubClientOptions packOptions() {
+ return (new EventHubClientOptions()).setOperationTimeout(this.operationTimeout).setRetryPolicy(this.retryPolicy).setTransportType(this.transportType);
+ }
+
+ private void normalizeConnectionStringAndEventHubPath() {
+ // The event hub path must appear in at least one of the eventHubPath argument or the connection string.
+ // If it appears in both, then it must be the same in both. If it appears in only one, populate the other.
+ ConnectionStringBuilder csb = new ConnectionStringBuilder(this.eventHubConnectionString);
+ String extractedEntityPath = csb.getEventHubName();
+ if ((this.eventHubPath != null) && !this.eventHubPath.isEmpty()) {
+ if (extractedEntityPath != null) {
+ if (this.eventHubPath.compareTo(extractedEntityPath) != 0) {
+ throw new IllegalArgumentException("Provided EventHub path in eventHubPath parameter conflicts with the path in provided EventHub connection string");
+ }
+ // else they are the same and that's fine
+ } else {
+ // There is no entity path in the connection string, so put it there.
+ csb.setEventHubName(this.eventHubPath);
+ }
+ } else {
+ if ((extractedEntityPath != null) && !extractedEntityPath.isEmpty()) {
+ this.eventHubPath = extractedEntityPath;
+ } else {
+ throw new IllegalArgumentException("Provide EventHub entity path in either eventHubPath argument or in eventHubConnectionString");
+ }
+ }
+
+ if (this.transportType != null) {
+ csb.setTransportType(this.transportType);
+ }
+ if (this.operationTimeout != null) {
+ csb.setOperationTimeout(this.operationTimeout);
+ }
+
+ this.eventHubConnectionString = csb.toString();
+ }
+ }
+ }
}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/HostContext.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/HostContext.java
index 1998a782e97a..26b7cc0285ed 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/HostContext.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/HostContext.java
@@ -3,8 +3,6 @@
package com.microsoft.azure.eventprocessorhost;
-import com.microsoft.azure.eventhubs.RetryPolicy;
-
import java.util.concurrent.ScheduledExecutorService;
final class HostContext {
@@ -18,8 +16,7 @@ final class HostContext {
private final String eventHubPath;
private final String consumerGroupName;
- private final String eventHubConnectionString;
- private final RetryPolicy retryPolicy;
+ private final EventHubClientFactory eventHubClientFactory;
private final ILeaseManager leaseManager;
private final ICheckpointManager checkpointManager;
@@ -33,7 +30,7 @@ final class HostContext {
HostContext(ScheduledExecutorService executor,
EventProcessorHost host, String hostName,
- String eventHubPath, String consumerGroupName, String eventHubConnectionString, RetryPolicy retryPolicy,
+ String eventHubPath, String consumerGroupName, EventHubClientFactory eventHubClientFactory,
ILeaseManager leaseManager, ICheckpointManager checkpointManager) {
this.executor = executor;
@@ -42,8 +39,7 @@ final class HostContext {
this.eventHubPath = eventHubPath;
this.consumerGroupName = consumerGroupName;
- this.eventHubConnectionString = eventHubConnectionString;
- this.retryPolicy = retryPolicy;
+ this.eventHubClientFactory = eventHubClientFactory;
this.leaseManager = leaseManager;
this.checkpointManager = checkpointManager;
@@ -65,12 +61,8 @@ String getConsumerGroupName() {
return this.consumerGroupName;
}
- String getEventHubConnectionString() {
- return this.eventHubConnectionString;
- }
-
- RetryPolicy getRetryPolicy() {
- return this.retryPolicy;
+ EventHubClientFactory getEventHubClientFactory() {
+ return this.eventHubClientFactory;
}
ILeaseManager getLeaseManager() {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionManager.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionManager.java
index 8f419a35e6e2..96d6ede65281 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionManager.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionManager.java
@@ -47,7 +47,7 @@ CompletableFuture cachePartitionIds() {
final CompletableFuture cleanupFuture = new CompletableFuture();
// Stage 0A: get EventHubClient for the event hub
- retval = EventHubClient.create(this.hostContext.getEventHubConnectionString(), this.hostContext.getRetryPolicy(), this.hostContext.getExecutor())
+ retval = this.hostContext.getEventHubClientFactory().createEventHubClient()
// Stage 0B: set up a way to close the EventHubClient when we're done
.thenApplyAsync((ehClient) -> {
final EventHubClient saveForCleanupClient = ehClient;
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionPump.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionPump.java
index d2e16d1e3c09..ec049c63d511 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionPump.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/main/java/com/microsoft/azure/eventprocessorhost/PartitionPump.java
@@ -175,8 +175,7 @@ private CompletableFuture openClients() {
CompletableFuture startOpeningFuture = null;
try {
- startOpeningFuture = EventHubClient.create(this.hostContext.getEventHubConnectionString(),
- this.hostContext.getRetryPolicy(), this.hostContext.getExecutor());
+ startOpeningFuture = this.hostContext.getEventHubClientFactory().createEventHubClient();
} catch (EventHubException | IOException e2) {
// Marking startOpeningFuture as completed exceptionally will cause all the
// following stages to fall through except stage 1 which will report the error.
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/CheckpointManagerTest.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/CheckpointManagerTest.java
index 838a17c5344a..669aebb50c04 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/CheckpointManagerTest.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/CheckpointManagerTest.java
@@ -232,15 +232,17 @@ private void setupOneManager(boolean useAzureStorage, int index, String suffix,
} else {
TestBase.logInfo("Container name: " + containerName);
String azureStorageConnectionString = TestUtilities.getStorageConnectionString();
- AzureStorageCheckpointLeaseManager azMgr = new AzureStorageCheckpointLeaseManager(azureStorageConnectionString, containerName);
+ AzureStorageCheckpointLeaseManager azMgr = new AzureStorageCheckpointLeaseManager(azureStorageConnectionString, containerName, null);
leaseMgr = azMgr;
checkpointMgr = azMgr;
}
// Host name needs to be unique per host so use index. Event hub should be the same for all hosts in a test, so use the supplied suffix.
- EventProcessorHost host = new EventProcessorHost("dummyHost" + String.valueOf(index), RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_EVENT_HUB_PATH + suffix,
- EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING + suffix, checkpointMgr, leaseMgr);
-
+ EventProcessorHost host = EventProcessorHost.EventProcessorHostBuilder.newBuilder("dummyHost" + String.valueOf(index), EventHubClient.DEFAULT_CONSUMER_GROUP_NAME)
+ .useUserCheckpointAndLeaseManagers(checkpointMgr, leaseMgr)
+ .useEventHubConnectionString(RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING + suffix,
+ RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_EVENT_HUB_PATH + suffix)
+ .build();
try {
if (!useAzureStorage) {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/EPHConstructorTests.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/EPHConstructorTests.java
index 3ae2c7b3f6e8..51070c03f3ef 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/EPHConstructorTests.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/EPHConstructorTests.java
@@ -38,7 +38,8 @@ public void missingEventHubPathTest() throws Exception {
settings.inEventHubDoesNotExist = true;
settings.inoutEPHConstructorArgs.dummyStorageConnection();
- settings.inoutEPHConstructorArgs.setEHPath("", PerTestSettings.EPHConstructorArgs.EH_PATH_OVERRIDE_AND_REPLACE);
+ settings.inoutEPHConstructorArgs.removePathFromEHConnection();
+ settings.inoutEPHConstructorArgs.setEHPath(null, PerTestSettings.EPHConstructorArgs.EH_PATH_OVERRIDE);
try {
settings = testSetup(settings);
@@ -191,7 +192,7 @@ public void ehPathOnlyInConnStringTest() throws Exception {
settings.inEventHubDoesNotExist = true;
settings.inoutEPHConstructorArgs.dummyStorageConnection();
- settings.inoutEPHConstructorArgs.setEHPath("", PerTestSettings.EPHConstructorArgs.EH_PATH_OVERRIDE);
+ settings.inoutEPHConstructorArgs.setEHPath(null, PerTestSettings.EPHConstructorArgs.EH_PATH_OVERRIDE);
try {
settings = testSetup(settings);
@@ -212,7 +213,7 @@ public void nullCheckpointManagerTest() throws Exception {
try {
settings = testSetup(settings);
fail("No exception occurred");
- } catch (IllegalArgumentException e) {
+ } catch (NullPointerException e) {
TestBase.logInfo("Got expected exception");
} finally {
testFinish(settings, NO_CHECKS);
@@ -231,7 +232,7 @@ public void nullLeaseManagerTest() throws Exception {
try {
settings = testSetup(settings);
fail("No exception occurred");
- } catch (IllegalArgumentException e) {
+ } catch (NullPointerException e) {
TestBase.logInfo("Got expected exception");
} finally {
testFinish(settings, NO_CHECKS);
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/LeaseManagerTest.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/LeaseManagerTest.java
index d424606afe7b..c819b0fee2cb 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/LeaseManagerTest.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/LeaseManagerTest.java
@@ -281,14 +281,17 @@ private void setupOneManager(boolean useAzureStorage, int index, String suffix,
} else {
TestBase.logInfo("Container name: " + containerName);
String azureStorageConnectionString = TestUtilities.getStorageConnectionString();
- AzureStorageCheckpointLeaseManager azMgr = new AzureStorageCheckpointLeaseManager(azureStorageConnectionString, containerName);
+ AzureStorageCheckpointLeaseManager azMgr = new AzureStorageCheckpointLeaseManager(azureStorageConnectionString, containerName, null);
leaseMgr = azMgr;
checkpointMgr = azMgr;
}
// Host name needs to be unique per host so use index. Event hub should be the same for all hosts in a test, so use the supplied suffix.
- EventProcessorHost host = new EventProcessorHost("dummyHost" + String.valueOf(index), RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_EVENT_HUB_PATH + suffix,
- EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING + suffix, checkpointMgr, leaseMgr);
+ EventProcessorHost host = EventProcessorHost.EventProcessorHostBuilder.newBuilder("dummyHost" + String.valueOf(index), EventHubClient.DEFAULT_CONSUMER_GROUP_NAME)
+ .useUserCheckpointAndLeaseManagers(checkpointMgr, leaseMgr)
+ .useEventHubConnectionString(RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING + suffix,
+ RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_EVENT_HUB_PATH + suffix)
+ .build();
try {
if (!useAzureStorage) {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PartitionManagerTest.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PartitionManagerTest.java
index fcce41773e7b..42fdc30bffa1 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PartitionManagerTest.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PartitionManagerTest.java
@@ -327,8 +327,11 @@ private void setup(int hostCount, int partitionCount, long latency, int threads)
if (threads > 0) {
threadpool = Executors.newScheduledThreadPool(threads);
}
- this.hosts[i] = new EventProcessorHost("dummyHost" + String.valueOf(i), "NOTREAL", EventHubClient.DEFAULT_CONSUMER_GROUP_NAME,
- RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING, cm, lm, threadpool, null);
+ this.hosts[i] = EventProcessorHost.EventProcessorHostBuilder.newBuilder("dummyHost" + String.valueOf(i), EventHubClient.DEFAULT_CONSUMER_GROUP_NAME)
+ .useUserCheckpointAndLeaseManagers(cm, lm)
+ .useEventHubConnectionString(RealEventHubUtilities.SYNTACTICALLY_CORRECT_DUMMY_CONNECTION_STRING, "NOTREAL")
+ .setExecutor(threadpool)
+ .build();
lm.initialize(this.hosts[i].getHostContext());
lm.setLatency(latency);
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PerTestSettings.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PerTestSettings.java
index 1ad68b3950f5..7172e8a8e4a5 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PerTestSettings.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/PerTestSettings.java
@@ -6,6 +6,8 @@
import java.util.ArrayList;
import java.util.concurrent.ScheduledExecutorService;
+import com.microsoft.azure.eventhubs.AzureActiveDirectoryTokenProvider;
+
public class PerTestSettings {
// In-out properties: may be set before test setup and then changed by setup.
final EPHConstructorArgs inoutEPHConstructorArgs;
@@ -60,6 +62,7 @@ class EPHConstructorArgs {
static final int LEASE_MANAGER_OVERRIDE = 0x0800;
static final int EXPLICIT_MANAGER = CHECKPOINT_MANAGER_OVERRIDE | LEASE_MANAGER_OVERRIDE;
static final int TELLTALE_ON_TIMEOUT = 0x1000;
+ static final int AUTH_CALLBACK = 0x2000;
private int flags;
@@ -67,6 +70,8 @@ class EPHConstructorArgs {
private String ehPath;
private String consumerGroupName;
private String ehConnection;
+ private AzureActiveDirectoryTokenProvider.AuthenticationCallback authCallback;
+ private String authAuthority;
private String storageConnection;
private String storageContainerName;
private String storageBlobPrefix;
@@ -81,6 +86,8 @@ class EPHConstructorArgs {
this.ehPath = null;
this.consumerGroupName = null;
this.ehConnection = null;
+ this.authCallback = null;
+ this.authAuthority = null;
this.storageConnection = null;
this.storageContainerName = null;
this.storageBlobPrefix = null;
@@ -136,6 +143,21 @@ void setEHConnection(String ehConnection) {
this.ehConnection = ehConnection;
this.flags |= EH_CONNECTION_OVERRIDE;
}
+
+ AzureActiveDirectoryTokenProvider.AuthenticationCallback getAuthCallback() {
+ return this.authCallback;
+ }
+
+ String getAuthAuthority() {
+ return this.authAuthority;
+ }
+
+ void setAuthCallback(AzureActiveDirectoryTokenProvider.AuthenticationCallback authCallback, String authAuthority)
+ {
+ this.authCallback = authCallback;
+ this.authAuthority = authAuthority;
+ this.flags |= AUTH_CALLBACK;
+ }
String getStorageConnection() {
return this.storageConnection;
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/RealEventHubUtilities.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/RealEventHubUtilities.java
index b3174c613ea7..492b9080fdb8 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/RealEventHubUtilities.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/RealEventHubUtilities.java
@@ -37,7 +37,7 @@ ArrayList setup(boolean skipIfFakeEH, int fakePartitions) throws EventHu
ArrayList partitionIds = setupWithoutSenders(skipIfFakeEH, fakePartitions);
// EventHubClient is source of all senders
- this.client = EventHubClient.createSync(this.hubConnectionString.toString(), TestUtilities.EXECUTOR_SERVICE);
+ this.client = EventHubClient.createFromConnectionStringSync(this.hubConnectionString.toString(), TestUtilities.EXECUTOR_SERVICE);
return partitionIds;
}
@@ -132,7 +132,7 @@ ArrayList getPartitionIdsForTest() throws EventHubException, IOException
this.cachedPartitionIds = new ArrayList();
ehCacheCheck(true);
- EventHubClient idClient = EventHubClient.createSync(this.hubConnectionString.toString(), TestUtilities.EXECUTOR_SERVICE);
+ EventHubClient idClient = EventHubClient.createFromConnectionStringSync(this.hubConnectionString.toString(), TestUtilities.EXECUTOR_SERVICE);
try {
EventHubRuntimeInformation info = idClient.getRuntimeInformation().get();
String[] ids = info.getPartitionIds();
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/Repros.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/Repros.java
index 8d0c6dcf9b40..9c072cd18d4d 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/Repros.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/Repros.java
@@ -42,17 +42,19 @@ public void conflictingHosts() throws Exception {
PrefabGeneralErrorHandler general1 = new PrefabGeneralErrorHandler();
PrefabProcessorFactory factory1 = new PrefabProcessorFactory(telltale, doCheckpointing, doMarker);
- EventProcessorHost host1 = new EventProcessorHost(conflictingName, utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- TestUtilities.getStorageConnectionString(), storageName);
+ EventProcessorHost host1 = EventProcessorHost.EventProcessorHostBuilder.newBuilder(conflictingName, utils.getConsumerGroup())
+ .useAzureStorageCheckpointLeaseManager(TestUtilities.getStorageConnectionString(), storageName, null)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .build();
EventProcessorOptions options1 = EventProcessorOptions.getDefaultOptions();
options1.setExceptionNotification(general1);
PrefabGeneralErrorHandler general2 = new PrefabGeneralErrorHandler();
PrefabProcessorFactory factory2 = new PrefabProcessorFactory(telltale, doCheckpointing, doMarker);
- EventProcessorHost host2 = new EventProcessorHost(conflictingName, utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- TestUtilities.getStorageConnectionString(), storageName);
+ EventProcessorHost host2 = EventProcessorHost.EventProcessorHostBuilder.newBuilder(conflictingName, utils.getConsumerGroup())
+ .useAzureStorageCheckpointLeaseManager(TestUtilities.getStorageConnectionString(), storageName, null)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .build();
EventProcessorOptions options2 = EventProcessorOptions.getDefaultOptions();
options2.setExceptionNotification(general2);
@@ -79,9 +81,11 @@ public void infiniteReceive() throws Exception {
PrefabProcessorFactory factory = new PrefabProcessorFactory("never match", PrefabEventProcessor.CheckpointChoices.CKP_NONE, true, false);
InMemoryCheckpointManager checkpointer = new InMemoryCheckpointManager();
InMemoryLeaseManager leaser = new InMemoryLeaseManager();
- EventProcessorHost host = new EventProcessorHost("infiniteReceive-1", utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- checkpointer, leaser, Executors.newScheduledThreadPool(16), null);
+ EventProcessorHost host = EventProcessorHost.EventProcessorHostBuilder.newBuilder("infiniteReceive-1", utils.getConsumerGroup())
+ .useUserCheckpointAndLeaseManagers(checkpointer, leaser)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .setExecutor(Executors.newScheduledThreadPool(16))
+ .build();
checkpointer.initialize(host.getHostContext());
leaser.initialize(host.getHostContext());
@@ -113,17 +117,19 @@ public void infiniteReceive2Hosts() throws Exception {
PrefabGeneralErrorHandler general1 = new PrefabGeneralErrorHandler();
PrefabProcessorFactory factory1 = new PrefabProcessorFactory("never match", PrefabEventProcessor.CheckpointChoices.CKP_NONE, true, false);
- EventProcessorHost host1 = new EventProcessorHost("infiniteReceive2Hosts-1", utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- TestUtilities.getStorageConnectionString(), storageName);
+ EventProcessorHost host1 = EventProcessorHost.EventProcessorHostBuilder.newBuilder("infiniteReceive2Hosts-1", utils.getConsumerGroup())
+ .useAzureStorageCheckpointLeaseManager(TestUtilities.getStorageConnectionString(), storageName, null)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .build();
EventProcessorOptions options1 = EventProcessorOptions.getDefaultOptions();
options1.setExceptionNotification(general1);
PrefabGeneralErrorHandler general2 = new PrefabGeneralErrorHandler();
PrefabProcessorFactory factory2 = new PrefabProcessorFactory("never match", PrefabEventProcessor.CheckpointChoices.CKP_NONE, true, false);
- EventProcessorHost host2 = new EventProcessorHost("infiniteReceive2Hosts-2", utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- TestUtilities.getStorageConnectionString(), storageName);
+ EventProcessorHost host2 = EventProcessorHost.EventProcessorHostBuilder.newBuilder("infiniteReceive2Hosts-1", utils.getConsumerGroup())
+ .useAzureStorageCheckpointLeaseManager(TestUtilities.getStorageConnectionString(), storageName, null)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .build();
EventProcessorOptions options2 = EventProcessorOptions.getDefaultOptions();
options2.setExceptionNotification(general2);
@@ -173,9 +179,10 @@ public void infiniteReceive2Hosts() throws Exception {
host2 = null;
} else {
factory2 = new PrefabProcessorFactory("never match", PrefabEventProcessor.CheckpointChoices.CKP_NONE, true, false);
- host2 = new EventProcessorHost("infiniteReceive2Hosts-2", utils.getConnectionString(true).getEventHubName(),
- utils.getConsumerGroup(), utils.getConnectionString(true).toString(),
- TestUtilities.getStorageConnectionString(), storageName);
+ host2 = EventProcessorHost.EventProcessorHostBuilder.newBuilder("infiniteReceive2Hosts-1", utils.getConsumerGroup())
+ .useAzureStorageCheckpointLeaseManager(TestUtilities.getStorageConnectionString(), storageName, null)
+ .useEventHubConnectionString(utils.getConnectionString(true).toString(), utils.getConnectionString(true).getEventHubName())
+ .build();
options2 = EventProcessorOptions.getDefaultOptions();
options2.setExceptionNotification(general2);
@@ -229,7 +236,7 @@ public void rawEpochStealing() throws Exception {
System.out.println("\nParked: " + parkedCount + " SELECTING: " + selectingList);
System.out.println("Client " + clientSerialNumber + " starting");
- EventHubClient client = EventHubClient.createSync(utils.getConnectionString(true).toString(), TestUtilities.EXECUTOR_SERVICE);
+ EventHubClient client = EventHubClient.createFromConnectionStringSync(utils.getConnectionString(true).toString(), TestUtilities.EXECUTOR_SERVICE);
PartitionReceiver receiver = client.createReceiver(utils.getConsumerGroup(), "0", EventPosition.fromStartOfStream()).get();
//client.createEpochReceiver(utils.getConsumerGroup(), "0", PartitionReceiver.START_OF_STREAM, 1).get();
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/SmokeTest.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/SmokeTest.java
index 1fc9b86b2eed..704b920b51ca 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/SmokeTest.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/SmokeTest.java
@@ -3,11 +3,20 @@
package com.microsoft.azure.eventprocessorhost;
+import com.microsoft.aad.msal4j.ClientCredentialParameters;
+import com.microsoft.aad.msal4j.ClientSecret;
+import com.microsoft.aad.msal4j.ConfidentialClientApplication;
+import com.microsoft.aad.msal4j.IAuthenticationResult;
+import com.microsoft.azure.eventhubs.AzureActiveDirectoryTokenProvider;
import com.microsoft.azure.eventhubs.EventPosition;
+
import org.junit.Assert;
import org.junit.Test;
import java.time.Instant;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
@@ -22,6 +31,51 @@ public void sendRecv1MsgTest() throws Exception {
testFinish(settings, SmokeTest.ANY_NONZERO_COUNT);
}
+
+ /**
+ * This JUnit test case is all commented out by default because it can only be run with special setup.
+ * It extracts the namespace (endpoint) and event hub name from the connection string in the environment variable
+ * which all test cases use, but it assumes that the namespace (or event hub) has been set up with special permissions.
+ * Within the AAD directory indicated by "authority", there is a registered application with id "clientId" and a secret
+ * "clientSecret". This application has been granted the "Azure Event Hubs Data Owner" role on the namespace or
+ * event hub.
+ */
+ //@Test
+ public void sendRecv1MsgAADTest() throws Exception {
+ PerTestSettings settings = new PerTestSettings("SendRecv1MsgAAD");
+ AzureActiveDirectoryTokenProvider.AuthenticationCallback authCallback = new MsalAuthCallback();
+ String authAuthority = "https://login.windows.net/replaceWithTenantIdGuid";
+ settings.inoutEPHConstructorArgs.setAuthCallback(authCallback, authAuthority);
+ settings = testSetup(settings);
+
+ settings.outUtils.sendToAny(settings.outTelltale);
+ waitForTelltale(settings);
+
+ testFinish(settings, SmokeTest.ANY_NONZERO_COUNT);
+ }
+
+ private class MsalAuthCallback implements AzureActiveDirectoryTokenProvider.AuthenticationCallback {
+ final private String clientId = "replaceWithClientIdGuid";
+ final private String clientSecret = "replaceWithClientSecret";
+
+ @Override
+ public CompletableFuture acquireToken(String audience, String authority, Object state) {
+ try {
+ ConfidentialClientApplication app = ConfidentialClientApplication.builder(this.clientId, new ClientSecret(this.clientSecret))
+ .authority(authority)
+ .build();
+
+ ClientCredentialParameters parameters = ClientCredentialParameters.builder(Collections.singleton(audience + ".default")).build();
+
+ IAuthenticationResult result = app.acquireToken(parameters).get();
+
+ return CompletableFuture.completedFuture(result.accessToken());
+ }
+ catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ }
+ }
@Test
public void receiverRuntimeMetricsTest() throws Exception {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/TestBase.java b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/TestBase.java
index a7447e5392e8..69451871e266 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/TestBase.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-eph/src/test/java/com/microsoft/azure/eventprocessorhost/TestBase.java
@@ -6,6 +6,8 @@
import com.microsoft.azure.eventhubs.ConnectionStringBuilder;
import com.microsoft.azure.eventhubs.EventHubClient;
import com.microsoft.azure.eventhubs.EventHubException;
+import com.microsoft.azure.eventprocessorhost.EventProcessorHost.EventProcessorHostBuilder.AuthStep;
+import com.microsoft.azure.eventprocessorhost.EventProcessorHost.EventProcessorHostBuilder.OptionalStep;
import org.junit.After;
import org.junit.AfterClass;
@@ -151,15 +153,17 @@ PerTestSettings testSetup(PerTestSettings settings) throws Exception {
settings.inOptions.setExceptionNotification(settings.outGeneralErrorHandler);
if (settings.inoutEPHConstructorArgs.useExplicitManagers()) {
- ICheckpointManager effectiveCheckpointMananger = settings.inoutEPHConstructorArgs.isFlagSet(PerTestSettings.EPHConstructorArgs.CHECKPOINT_MANAGER_OVERRIDE)
+ ICheckpointManager effectiveCheckpointManager = settings.inoutEPHConstructorArgs.isFlagSet(PerTestSettings.EPHConstructorArgs.CHECKPOINT_MANAGER_OVERRIDE)
? settings.inoutEPHConstructorArgs.getCheckpointMananger()
: new BogusCheckpointMananger();
ILeaseManager effectiveLeaseManager = settings.inoutEPHConstructorArgs.isFlagSet(PerTestSettings.EPHConstructorArgs.LEASE_MANAGER_OVERRIDE)
? settings.inoutEPHConstructorArgs.getLeaseManager()
: new BogusLeaseManager();
- settings.outHost = new EventProcessorHost(effectiveHostName, effectiveEntityPath, effectiveConsumerGroup, effectiveConnectionString,
- effectiveCheckpointMananger, effectiveLeaseManager, effectiveExecutor, null);
+ settings.outHost = EventProcessorHost.EventProcessorHostBuilder.newBuilder(effectiveHostName, effectiveConsumerGroup)
+ .useUserCheckpointAndLeaseManagers(effectiveCheckpointManager, effectiveLeaseManager)
+ .useEventHubConnectionString(effectiveConnectionString, effectiveEntityPath)
+ .setExecutor(effectiveExecutor).build();
} else {
String effectiveStorageConnectionString = settings.inoutEPHConstructorArgs.isFlagSet(PerTestSettings.EPHConstructorArgs.STORAGE_CONNECTION_OVERRIDE)
? settings.inoutEPHConstructorArgs.getStorageConnection()
@@ -179,8 +183,17 @@ PerTestSettings testSetup(PerTestSettings settings) throws Exception {
? settings.inoutEPHConstructorArgs.getStorageBlobPrefix()
: null;
- settings.outHost = new EventProcessorHost(effectiveHostName, effectiveEntityPath, effectiveConsumerGroup, effectiveConnectionString,
- effectiveStorageConnectionString, effectiveStorageContainerName, effectiveBlobPrefix, effectiveExecutor);
+ AuthStep intermediate = EventProcessorHost.EventProcessorHostBuilder.newBuilder(effectiveHostName, effectiveConsumerGroup)
+ .useAzureStorageCheckpointLeaseManager(effectiveStorageConnectionString, effectiveStorageContainerName, effectiveBlobPrefix);
+ OptionalStep almostDone = null;
+ if (settings.inoutEPHConstructorArgs.isFlagSet(PerTestSettings.EPHConstructorArgs.AUTH_CALLBACK)) {
+ ConnectionStringBuilder csb = new ConnectionStringBuilder(effectiveConnectionString);
+ almostDone = intermediate.useAADAuthentication(csb.getEndpoint(), effectiveEntityPath)
+ .useAuthenticationCallback(settings.inoutEPHConstructorArgs.getAuthCallback(), settings.inoutEPHConstructorArgs.getAuthAuthority());
+ } else {
+ almostDone = intermediate.useEventHubConnectionString(effectiveConnectionString, effectiveEntityPath);
+ }
+ settings.outHost = almostDone.setExecutor(effectiveExecutor).build();
}
if (!settings.inEventHubDoesNotExist) {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-extensions/pom.xml b/sdk/eventhubs/microsoft-azure-eventhubs-extensions/pom.xml
index 5faee62b351f..9dcf397d18dd 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-extensions/pom.xml
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-extensions/pom.xml
@@ -7,7 +7,7 @@
com.microsoft.azure
azure-eventhubs-clients
- 2.3.1
+ 3.0.0
../pom.data.xml
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs-extensions/src/main/java/com/microsoft/azure/eventhubs/extensions/appender/EventHubsManager.java b/sdk/eventhubs/microsoft-azure-eventhubs-extensions/src/main/java/com/microsoft/azure/eventhubs/extensions/appender/EventHubsManager.java
index c4c9a9e6ad3e..aa22b28b1200 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs-extensions/src/main/java/com/microsoft/azure/eventhubs/extensions/appender/EventHubsManager.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs-extensions/src/main/java/com/microsoft/azure/eventhubs/extensions/appender/EventHubsManager.java
@@ -43,6 +43,6 @@ public void send(final Iterable messages) throws EventHubException {
}
public void startup() throws EventHubException, IOException {
- this.eventHubSender = EventHubClient.createSync(this.eventHubConnectionString, EXECUTOR_SERVICE);
+ this.eventHubSender = EventHubClient.createFromConnectionStringSync(this.eventHubConnectionString, EXECUTOR_SERVICE);
}
}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/pom.xml b/sdk/eventhubs/microsoft-azure-eventhubs/pom.xml
index dc90cc557ca5..0be6b248879b 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/pom.xml
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/pom.xml
@@ -7,7 +7,7 @@
com.microsoft.azure
azure-eventhubs-clients
- 2.3.1
+ 3.0.0
../pom.data.xml
@@ -30,4 +30,30 @@
scm:git:https://github.com/Azure/azure-sdk-for-java
+
+
+ com.microsoft.azure
+ azure-client-authentication
+ ${client-runtime.version}
+ compile
+
+
+ com.microsoft.azure
+ msal4j
+ 0.4.0-preview
+ test
+
+
+ com.microsoft.azure
+ adal4j
+ ${adal4j.version}
+ test
+
+
+ com.nimbusds
+ nimbus-jose-jwt
+ 6.0.1
+
+
+
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AuthorizationFailedException.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AuthorizationFailedException.java
index 3905f043321c..436f5fd9b69c 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AuthorizationFailedException.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AuthorizationFailedException.java
@@ -8,7 +8,7 @@
/**
* Authorization failed exception is thrown when error is encountered during authorizing user's permission to run the intended operations.
* When encountered this exception user should check whether the token/key provided in the connection string (e.g. one passed to
- * {@link EventHubClient#create(String, ScheduledExecutorService)}) is valid, and has correct execution right for the intended operations (e.g.
+ * {@link EventHubClient#createFromConnectionString(String, ScheduledExecutorService)}) is valid, and has correct execution right for the intended operations (e.g.
* Receive call will need Listen claim associated with the key/token).
*
* @see http://go.microsoft.com/fwlink/?LinkId=761101
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AzureActiveDirectoryTokenProvider.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AzureActiveDirectoryTokenProvider.java
new file mode 100644
index 000000000000..5247aebf7cf1
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/AzureActiveDirectoryTokenProvider.java
@@ -0,0 +1,42 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.azure.eventhubs;
+
+import java.text.ParseException;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+import com.microsoft.azure.eventhubs.impl.ClientConstants;
+
+public final class AzureActiveDirectoryTokenProvider implements ITokenProvider {
+ private final AuthenticationCallback authCallback;
+ private final String authority;
+ private final Object authCallbackState;
+
+ public AzureActiveDirectoryTokenProvider(
+ final AuthenticationCallback authenticationCallback,
+ final String authority,
+ final Object state) {
+ this.authCallbackState = state;
+ this.authority = authority;
+ this.authCallback = authenticationCallback;
+ }
+
+ @Override
+ public CompletableFuture getToken(String resource, Duration timeout) {
+ return this.authCallback.acquireToken(ClientConstants.EVENTHUBS_AUDIENCE, this.authority, this.authCallbackState)
+ .thenApply((rawToken) -> {
+ try {
+ return new JsonSecurityToken(rawToken, resource);
+ } catch (ParseException e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+
+ public interface AuthenticationCallback {
+ CompletableFuture acquireToken(final String audience, final String authority, final Object state);
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ConnectionStringBuilder.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ConnectionStringBuilder.java
index d203026ddec2..0118a4058f8f 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ConnectionStringBuilder.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ConnectionStringBuilder.java
@@ -61,10 +61,13 @@ public final class ConnectionStringBuilder {
static final String SHARED_ACCESS_KEY_CONFIG_NAME = "SharedAccessKey";
static final String SHARED_ACCESS_SIGNATURE_CONFIG_NAME = "SharedAccessSignature";
static final String TRANSPORT_TYPE_CONFIG_NAME = "TransportType";
+ static final String AUTHENTICATION_CONFIG_NAME = "Authentication";
+
+ static public final String MANAGED_IDENTITY_AUTHENTICATION = "Managed Identity";
private static final String ALL_KEY_ENUMERATE_REGEX = "(" + HOST_NAME_CONFIG_NAME + "|" + ENDPOINT_CONFIG_NAME + "|" + SHARED_ACCESS_KEY_NANE_CONFIG_NAME
+ "|" + SHARED_ACCESS_KEY_CONFIG_NAME + "|" + SHARED_ACCESS_SIGNATURE_CONFIG_NAME + "|" + ENTITY_PATH_CONFIG_NAME + "|" + OPERATION_TIMEOUT_CONFIG_NAME
- + "|" + TRANSPORT_TYPE_CONFIG_NAME + ")";
+ + "|" + TRANSPORT_TYPE_CONFIG_NAME + "|" + AUTHENTICATION_CONFIG_NAME + ")";
private static final String KEYS_WITH_DELIMITERS_REGEX = KEY_VALUE_PAIR_DELIMITER + ALL_KEY_ENUMERATE_REGEX
+ KEY_VALUE_SEPARATOR;
@@ -76,6 +79,7 @@ public final class ConnectionStringBuilder {
private String sharedAccessSignature;
private Duration operationTimeout;
private TransportType transportType;
+ private String authentication;
/**
* Creates an empty {@link ConnectionStringBuilder}. At minimum, a namespace name, an entity path, SAS key name, and SAS key
@@ -278,6 +282,26 @@ public ConnectionStringBuilder setTransportType(final TransportType transportTyp
this.transportType = transportType;
return this;
}
+
+ /**
+ * Get the authentication type in the Connection String.
+ *
+ * @return authentication
+ */
+ public String getAuthentication() {
+ return this.authentication;
+ }
+
+ /**
+ * Set the authentication type in the Connection String. The only valid values are "Managed Identity" or null.
+ *
+ * @param authentication
+ * @return the {@link ConnectionStringBuilder} instance being set.
+ */
+ public ConnectionStringBuilder setAuthentication(final String authentication) {
+ this.authentication = authentication;
+ return this;
+ }
/**
* Returns an inter-operable connection string that can be used to connect to EventHubs instances.
@@ -321,6 +345,11 @@ public String toString() {
connectionStringBuilder.append(String.format(Locale.US, "%s%s%s%s", TRANSPORT_TYPE_CONFIG_NAME,
KEY_VALUE_SEPARATOR, this.transportType.toString(), KEY_VALUE_PAIR_DELIMITER));
}
+
+ if (!StringUtil.isNullOrWhiteSpace(this.authentication)) {
+ connectionStringBuilder.append(String.format(Locale.US, "%s%s%s%s", AUTHENTICATION_CONFIG_NAME,
+ KEY_VALUE_SEPARATOR, this.authentication, KEY_VALUE_PAIR_DELIMITER));
+ }
connectionStringBuilder.deleteCharAt(connectionStringBuilder.length() - 1);
return connectionStringBuilder.toString();
@@ -409,7 +438,9 @@ private void parseConnectionString(final String connectionString) {
String.format(Locale.US, "Invalid value specified for property '%s' in the ConnectionString.", TRANSPORT_TYPE_CONFIG_NAME),
exception);
}
- } else {
+ } else if (key.equalsIgnoreCase(AUTHENTICATION_CONFIG_NAME)) {
+ this.authentication = values[valueIndex];
+ } else {
throw new IllegalConnectionStringFormatException(
String.format(Locale.US, "Illegal connection string parameter name: %s", key));
}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ErrorContext.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ErrorContext.java
index ebf4ecaf3037..ef55c00f1f3a 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ErrorContext.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ErrorContext.java
@@ -9,7 +9,9 @@
import java.util.Locale;
public abstract class ErrorContext implements Serializable {
- private final String namespaceName;
+ private static final long serialVersionUID = -841174412304936908L;
+
+ private final String namespaceName;
protected ErrorContext(final String namespaceName) {
this.namespaceName = namespaceName;
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventData.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventData.java
index e2091f2977e6..128eccf75f7c 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventData.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventData.java
@@ -50,7 +50,7 @@ public interface EventData extends Serializable, Comparable {
*
* @param data the actual payload of data in bytes to be Sent to EventHubs.
* @return EventData the created {@link EventData} to send to EventHubs.
- * @see EventHubClient#create(String, ScheduledExecutorService)
+ * @see EventHubClient#createFromConnectionString(String, ScheduledExecutorService)
*/
static EventData create(final byte[] data) {
return new EventDataImpl(data);
@@ -74,7 +74,7 @@ static EventData create(final byte[] data) {
* @param offset Offset in the byte[] to read from ; inclusive index
* @param length length of the byte[] to be read, starting from offset
* @return EventData the created {@link EventData} to send to EventHubs.
- * @see EventHubClient#create(String, ScheduledExecutorService)
+ * @see EventHubClient#createFromConnectionString(String, ScheduledExecutorService)
*/
static EventData create(final byte[] data, final int offset, final int length) {
return new EventDataImpl(data, offset, length);
@@ -96,7 +96,7 @@ static EventData create(final byte[] data, final int offset, final int length) {
*
* @param buffer ByteBuffer which references the payload of the Event to be sent to EventHubs
* @return EventData the created {@link EventData} to send to EventHubs.
- * @see EventHubClient#create(String, ScheduledExecutorService)
+ * @see EventHubClient#createFromConnectionString(String, ScheduledExecutorService)
*/
static EventData create(final ByteBuffer buffer) {
return new EventDataImpl(buffer);
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClient.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClient.java
index ea49239ed07e..3d11e7ee95d3 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClient.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClient.java
@@ -7,6 +7,7 @@
import com.microsoft.azure.eventhubs.impl.ExceptionUtil;
import java.io.IOException;
+import java.net.URI;
import java.nio.channels.UnresolvedAddressException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
@@ -14,14 +15,14 @@
/**
* Anchor class - all EventHub client operations STARTS here.
*
- * @see EventHubClient#create(String, ScheduledExecutorService)
+ * @see EventHubClient#createFromConnectionString(String, ScheduledExecutorService)
*/
public interface EventHubClient {
String DEFAULT_CONSUMER_GROUP_NAME = "$Default";
/**
- * Synchronous version of {@link #create(String, ScheduledExecutorService)}.
+ * Synchronous version of {@link #createFromConnectionString(String, ScheduledExecutorService)}.
*
* @param connectionString The connection string to be used. See {@link ConnectionStringBuilder} to construct a connectionString.
* @param executor An {@link ScheduledExecutorService} to run all tasks performed by {@link EventHubClient}.
@@ -29,13 +30,13 @@ public interface EventHubClient {
* @throws EventHubException If Service Bus service encountered problems during connection creation.
* @throws IOException If the underlying Proton-J layer encounter network errors.
*/
- static EventHubClient createSync(final String connectionString, final ScheduledExecutorService executor)
+ static EventHubClient createFromConnectionStringSync(final String connectionString, final ScheduledExecutorService executor)
throws EventHubException, IOException {
- return EventHubClient.createSync(connectionString, null, executor);
+ return createFromConnectionStringSync(connectionString, null, executor);
}
/**
- * Synchronous version of {@link #create(String, ScheduledExecutorService)}.
+ * Synchronous version of {@link #createFromConnectionString(String, ScheduledExecutorService)}.
*
* @param connectionString The connection string to be used. See {@link ConnectionStringBuilder} to construct a connectionString.
* @param retryPolicy A custom {@link RetryPolicy} to be used when communicating with EventHub.
@@ -44,9 +45,9 @@ static EventHubClient createSync(final String connectionString, final ScheduledE
* @throws EventHubException If Service Bus service encountered problems during connection creation.
* @throws IOException If the underlying Proton-J layer encounter network errors.
*/
- static EventHubClient createSync(final String connectionString, final RetryPolicy retryPolicy, final ScheduledExecutorService executor)
+ static EventHubClient createFromConnectionStringSync(final String connectionString, final RetryPolicy retryPolicy, final ScheduledExecutorService executor)
throws EventHubException, IOException {
- return ExceptionUtil.syncWithIOException(() -> create(connectionString, retryPolicy, executor).get());
+ return ExceptionUtil.syncWithIOException(() -> createFromConnectionString(connectionString, retryPolicy, executor).get());
}
/**
@@ -60,9 +61,9 @@ static EventHubClient createSync(final String connectionString, final RetryPolic
* @throws EventHubException If Service Bus service encountered problems during connection creation.
* @throws IOException If the underlying Proton-J layer encounter network errors.
*/
- static CompletableFuture create(final String connectionString, final ScheduledExecutorService executor)
+ static CompletableFuture createFromConnectionString(final String connectionString, final ScheduledExecutorService executor)
throws EventHubException, IOException {
- return EventHubClient.create(connectionString, null, executor);
+ return createFromConnectionString(connectionString, null, executor);
}
/**
@@ -77,12 +78,61 @@ static CompletableFuture create(final String connectionString, f
* @throws EventHubException If Service Bus service encountered problems during connection creation.
* @throws IOException If the underlying Proton-J layer encounter network errors.
*/
- static CompletableFuture create(
+ static CompletableFuture createFromConnectionString(
final String connectionString, final RetryPolicy retryPolicy, final ScheduledExecutorService executor)
throws EventHubException, IOException {
return EventHubClientImpl.create(connectionString, retryPolicy, executor);
}
+ /**
+ * Factory method to create an instance of {@link EventHubClient} using the supplied namespace endpoint address, eventhub name and authentication mechanism.
+ * In a normal scenario (when re-direct is not enabled) - one EventHubClient instance maps to one Connection to the Azure ServiceBus EventHubs service.
+ * The {@link EventHubClient} created from this method creates a Sender instance internally, which is used by the {@link #send(EventData)} methods.
+ *
+ * @param endpointAddress namespace level endpoint. This needs to be in the format of scheme://fullyQualifiedServiceBusNamespaceEndpointName
+ * @param eventHubName EventHub name
+ * @param authCallback A callback which returns a JSON Web Token obtained from AAD.
+ * @param authority Address of the AAD authority to issue the token.
+ * @param executor An {@link ScheduledExecutorService} to run all tasks performed by {@link EventHubClient}.
+ * @param options Options {@link EventHubClientOptions} for creating the client. Uses all defaults if null.
+ * @return EventHubClient which can be used to create Senders and Receivers to EventHub
+ * @throws EventHubException If the EventHubs service encountered problems during connection creation.
+ * @throws IOException If the underlying Proton-J layer encounter network errors.
+ */
+ public static CompletableFuture createWithAzureActiveDirectory(
+ final URI endpointAddress,
+ final String eventHubName,
+ final AzureActiveDirectoryTokenProvider.AuthenticationCallback authCallback,
+ final String authority,
+ final ScheduledExecutorService executor,
+ final EventHubClientOptions options) throws EventHubException, IOException {
+ ITokenProvider tokenProvider = new AzureActiveDirectoryTokenProvider(authCallback, authority, null);
+ return createWithTokenProvider(endpointAddress, eventHubName, tokenProvider, executor, options);
+ }
+
+ /**
+ * Factory method to create an instance of {@link EventHubClient} using the supplied namespace endpoint address, eventhub name and authentication mechanism.
+ * In a normal scenario (when re-direct is not enabled) - one EventHubClient instance maps to one Connection to the Azure ServiceBus EventHubs service.
+ * The {@link EventHubClient} created from this method creates a Sender instance internally, which is used by the {@link #send(EventData)} methods.
+ *
+ * @param endpointAddress namespace level endpoint. This needs to be in the format of scheme://fullyQualifiedServiceBusNamespaceEndpointName
+ * @param eventHubName EventHub name
+ * @param tokenProvider The {@link ITokenProvider} implementation to be used to authenticate
+ * @param executor An {@link ScheduledExecutorService} to run all tasks performed by {@link EventHubClient}.
+ * @param options Options {@link EventHubClientOptions} for creating the client. Uses all defaults if null.
+ * @return EventHubClient which can be used to create Senders and Receivers to EventHub
+ * @throws EventHubException If the EventHubs service encountered problems during connection creation.
+ * @throws IOException If the underlying Proton-J layer encounter network errors.
+ */
+ public static CompletableFuture createWithTokenProvider(
+ final URI endpointAddress,
+ final String eventHubName,
+ final ITokenProvider tokenProvider,
+ final ScheduledExecutorService executor,
+ final EventHubClientOptions options) throws EventHubException, IOException {
+ return EventHubClientImpl.create(endpointAddress, eventHubName, tokenProvider, executor, options);
+ }
+
/**
* @return the name of the Event Hub this client is connected to.
*/
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClientOptions.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClientOptions.java
new file mode 100644
index 000000000000..562a0ae40cba
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/EventHubClientOptions.java
@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.azure.eventhubs;
+
+import java.time.Duration;
+
+/**
+ * Convenient container for options for creating an {@link EventHubClient}
+ * All options default to not specified (null)
+ */
+public class EventHubClientOptions {
+ private Duration operationTimeout = null;
+ private TransportType transportType = null;
+ private RetryPolicy retryPolicy = null;
+
+ /**
+ * Create with all defaults
+ */
+ public EventHubClientOptions() {
+ }
+
+ /**
+ * Set the operation timeout.
+ * @param operationTimeout new operation timeout, null to unset any previous value
+ * @return this options object
+ */
+ public EventHubClientOptions setOperationTimeout(Duration operationTimeout) {
+ this.operationTimeout = operationTimeout;
+ return this;
+ }
+
+ /**
+ * Get the operation timeout.
+ * @return operation timeout or null if not set
+ */
+ public Duration getOperationTimeout() {
+ return this.operationTimeout;
+ }
+
+ /**
+ * Set the {@link TransportType} for the connection to the Event Hubs service
+ * @param transportType new transport type, null to unset any previous value
+ * @return this options object
+ */
+ public EventHubClientOptions setTransportType(TransportType transportType) {
+ this.transportType = transportType;
+ return this;
+ }
+
+ /**
+ * Get the transport type
+ * @return {@link TransportType} or null if not set
+ */
+ public TransportType getTransportType() {
+ return this.transportType;
+ }
+
+ /**
+ * Set the {@link RetryPolicy} for operations
+ * @param retryPolicy new retry policy, null to unset any previous value
+ * @return this options object
+ */
+ public EventHubClientOptions setRetryPolicy(RetryPolicy retryPolicy) {
+ this.retryPolicy = retryPolicy;
+ return this;
+ }
+
+ /**
+ * Get the retry policy
+ * @return {@link RetryPolicy} or null if not set
+ */
+ public RetryPolicy getRetryPolicy() {
+ return this.retryPolicy;
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ITokenProvider.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ITokenProvider.java
new file mode 100644
index 000000000000..21c668711ccc
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ITokenProvider.java
@@ -0,0 +1,11 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.azure.eventhubs;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+
+public interface ITokenProvider {
+ CompletableFuture getToken(final String resource, final Duration timeout);
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/JsonSecurityToken.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/JsonSecurityToken.java
new file mode 100644
index 000000000000..8f45de59393a
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/JsonSecurityToken.java
@@ -0,0 +1,31 @@
+package com.microsoft.azure.eventhubs;
+
+import java.text.ParseException;
+import java.util.Date;
+
+import com.microsoft.azure.eventhubs.impl.ClientConstants;
+import com.nimbusds.jwt.JWT;
+import com.nimbusds.jwt.JWTClaimsSet;
+import com.nimbusds.jwt.JWTParser;
+
+/**
+ * Extend SecurityToken with some specifics for a JSon Web Token
+ *
+ */
+public class JsonSecurityToken extends SecurityToken {
+ /**
+ * Construct from a raw JWT.
+ * @param rawToken the JWT token data
+ * @param audience audience of the token
+ * @throws ParseException if the token cannot be parsed
+ */
+ public JsonSecurityToken(final String rawToken, final String audience) throws ParseException {
+ super(rawToken, GetExpirationDateTimeUtcFromToken(rawToken), audience, ClientConstants.JWT_TOKEN_TYPE);
+ }
+
+ static Date GetExpirationDateTimeUtcFromToken(final String token) throws ParseException {
+ JWT jwt = JWTParser.parse(token);
+ JWTClaimsSet claims = jwt.getJWTClaimsSet();
+ return claims.getExpirationTime();
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ManagedIdentityTokenProvider.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ManagedIdentityTokenProvider.java
new file mode 100644
index 000000000000..c8e4b3d324d2
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/ManagedIdentityTokenProvider.java
@@ -0,0 +1,26 @@
+package com.microsoft.azure.eventhubs;
+
+import java.io.IOException;
+import java.text.ParseException;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+import com.microsoft.azure.credentials.MSICredentials;
+import com.microsoft.azure.eventhubs.impl.ClientConstants;
+
+public class ManagedIdentityTokenProvider implements ITokenProvider {
+ final static MSICredentials credentials = new MSICredentials();
+
+ @Override
+ public CompletableFuture getToken(final String resource, final Duration timeout) {
+ return CompletableFuture.supplyAsync(() -> {
+ try {
+ String rawToken = ManagedIdentityTokenProvider.credentials.getToken(ClientConstants.EVENTHUBS_AUDIENCE);
+ return new JsonSecurityToken(rawToken, resource);
+ } catch (IOException | ParseException e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionSender.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionSender.java
index c191c9881698..3ae79cecdb48 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionSender.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/PartitionSender.java
@@ -13,7 +13,7 @@
* if you do not care about sending events to specific partitions. Instead, use {@link EventHubClient#send} method.
*
* @see EventHubClient#createPartitionSender(String)
- * @see EventHubClient#create(String, ScheduledExecutorService)
+ * @see EventHubClient#createFromConnectionString(String, ScheduledExecutorService)
*/
public interface PartitionSender {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/QuotaExceededException.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/QuotaExceededException.java
index e0ff9b50d376..10533ea5c753 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/QuotaExceededException.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/QuotaExceededException.java
@@ -4,8 +4,9 @@
package com.microsoft.azure.eventhubs;
public class QuotaExceededException extends EventHubException {
+ private static final long serialVersionUID = 1L;
- public QuotaExceededException(String message) {
+ public QuotaExceededException(String message) {
super(false, message);
}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/SecurityToken.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/SecurityToken.java
new file mode 100644
index 000000000000..7ee3293bb4eb
--- /dev/null
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/SecurityToken.java
@@ -0,0 +1,63 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.azure.eventhubs;
+
+import java.util.Date;
+
+/**
+ * A generic representation of a security token.
+ *
+ */
+public class SecurityToken {
+ private final String tokenType;
+ private final String token;
+ private final Date validTo;
+ private final String audience;
+
+ /**
+ * Constructor.
+ * @param token the actual token data
+ * @param validTo expiration date/time of the token
+ * @param audience audience of the token
+ * @param tokenType type of the token
+ */
+ public SecurityToken(final String token, final Date validTo, final String audience, final String tokenType) {
+ this.tokenType = tokenType;
+ this.token = token;
+ this.validTo = validTo;
+ this.audience = audience;
+ }
+
+ /**
+ * Get the type of the token.
+ * @return token type
+ */
+ public String getTokenType() {
+ return this.tokenType;
+ }
+
+ /**
+ * Get the token itself.
+ * @return token
+ */
+ public String getToken() {
+ return this.token;
+ }
+
+ /**
+ * Get the expiration date/time of the token.
+ * @return expiration
+ */
+ public Date validTo() {
+ return this.validTo;
+ }
+
+ /**
+ * Get the audience of the token.
+ * @return audience
+ */
+ public String getAudience() {
+ return this.audience;
+ }
+}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ActiveClientTokenManager.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ActiveClientTokenManager.java
index 63a20902717f..4c4bcd884420 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ActiveClientTokenManager.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ActiveClientTokenManager.java
@@ -19,7 +19,7 @@ final class ActiveClientTokenManager {
private final Duration tokenRefreshInterval;
private final SchedulerProvider schedulerProvider;
private final Timer timerScheduler;
- private CompletableFuture timer;
+ private CompletableFuture> timer;
ActiveClientTokenManager(
final ClientEntity clientEntity,
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/AmqpUtil.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/AmqpUtil.java
index e4b1e9a6e2ff..ec97dc294ccb 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/AmqpUtil.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/AmqpUtil.java
@@ -11,6 +11,7 @@
import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
import org.apache.qpid.proton.message.Message;
+import java.util.Date;
import java.util.Locale;
public final class AmqpUtil {
@@ -121,6 +122,10 @@ private static int sizeof(Object obj) {
return Double.BYTES;
}
+ if (obj instanceof Date) {
+ return 32;
+ }
+
throw new IllegalArgumentException(String.format(Locale.US, "Encoding Type: %s is not supported", obj.getClass()));
}
}
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/CBSChannel.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/CBSChannel.java
index 4dcfcd6c5c38..c7e3f073715c 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/CBSChannel.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/CBSChannel.java
@@ -8,18 +8,24 @@
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.message.Message;
+import com.microsoft.azure.eventhubs.SecurityToken;
+
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Consumer;
final class CBSChannel {
-
+ final ScheduledExecutorService executor;
final FaultTolerantObject innerChannel;
CBSChannel(
final SessionProvider sessionProvider,
final AmqpConnection connection,
- final String clientId) {
-
+ final String clientId,
+ final ScheduledExecutorService executor) {
+ this.executor = executor;
RequestResponseCloser closer = new RequestResponseCloser();
this.innerChannel = new FaultTolerantObject<>(
new RequestResponseOpener(sessionProvider, clientId, "cbs-session", "cbs", ClientConstants.CBS_ADDRESS, connection),
@@ -29,27 +35,47 @@ final class CBSChannel {
public void sendToken(
final ReactorDispatcher dispatcher,
- final String token,
+ final CompletableFuture tokenFuture,
final String tokenAudience,
- final OperationResult sendTokenCallback) {
+ final OperationResult sendTokenCallback,
+ final Consumer errorCallback) {
+ tokenFuture.thenAcceptAsync((token) -> {
+ innerSendToken(dispatcher, token, tokenAudience, sendTokenCallback);
+ }, this.executor)
+ .whenCompleteAsync((empty, exception) -> {
+ // TODO: whenCompleteAsync presents a Throwable. But many of the error callbacks expect
+ // an Exception. For now, do a cast here. Will we ever actually get an error that is
+ // not an Exception?
+ if ((exception != null) && (exception instanceof Exception)) {
+ errorCallback.accept((Exception)exception);
+ }
+ }, this.executor);
+ }
+ private void innerSendToken(
+ final ReactorDispatcher dispatcher,
+ final SecurityToken token,
+ final String tokenAudience,
+ final OperationResult sendTokenCallback) {
final Message request = Proton.message();
final Map properties = new HashMap<>();
properties.put(ClientConstants.PUT_TOKEN_OPERATION, ClientConstants.PUT_TOKEN_OPERATION_VALUE);
- properties.put(ClientConstants.PUT_TOKEN_TYPE, ClientConstants.SAS_TOKEN_TYPE);
+ properties.put(ClientConstants.PUT_TOKEN_TYPE, token.getTokenType());
+ properties.put(ClientConstants.PUT_TOKEN_EXPIRY, token.validTo());
properties.put(ClientConstants.PUT_TOKEN_AUDIENCE, tokenAudience);
+
final ApplicationProperties applicationProperties = new ApplicationProperties(properties);
request.setApplicationProperties(applicationProperties);
- request.setBody(new AmqpValue(token));
+ request.setBody(new AmqpValue(token.getToken()));
final MessageOperationResult messageOperation = new MessageOperationResult(response -> sendTokenCallback.onComplete(null), sendTokenCallback::onError);
final OperationResultBase operation = new OperationResultBase<>(
result -> result.request(request, messageOperation),
sendTokenCallback::onError);
-
+
this.innerChannel.runOnOpenedObject(dispatcher, operation);
}
-
+
public void close(
final ReactorDispatcher reactorDispatcher,
final OperationResult closeCallback) {
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ClientConstants.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ClientConstants.java
index 0cd1b70972cf..497d72f0cb05 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ClientConstants.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/ClientConstants.java
@@ -37,7 +37,7 @@ public final class ClientConstants {
public static final String NO_RETRY = "NoRetry";
public static final String DEFAULT_RETRY = "Default";
public static final String PRODUCT_NAME = "MSJavaClient";
- public static final String CURRENT_JAVACLIENT_VERSION = "2.3.1";
+ public static final String CURRENT_JAVACLIENT_VERSION = "3.0.0";
public static final String PLATFORM_INFO = getPlatformInfo();
public static final String FRAMEWORK_INFO = getFrameworkInfo();
public static final String CBS_ADDRESS = "$cbs";
@@ -45,10 +45,12 @@ public final class ClientConstants {
public static final String PUT_TOKEN_OPERATION_VALUE = "put-token";
public static final String PUT_TOKEN_TYPE = "type";
public static final String SAS_TOKEN_TYPE = "servicebus.windows.net:sastoken";
+ public static final String JWT_TOKEN_TYPE = "jwt";
public static final String PUT_TOKEN_AUDIENCE = "name";
public static final String PUT_TOKEN_EXPIRY = "expiration";
public static final String PUT_TOKEN_STATUS_CODE = "status-code";
public static final String PUT_TOKEN_STATUS_DESCRIPTION = "status-description";
+ public final static String EVENTHUBS_AUDIENCE = "https://eventhubs.azure.net/";
public static final String MANAGEMENT_ADDRESS = "$management";
public static final String MANAGEMENT_EVENTHUB_ENTITY_TYPE = AmqpConstants.VENDOR + ":eventhub";
public static final String MANAGEMENT_PARTITION_ENTITY_TYPE = AmqpConstants.VENDOR + ":partition";
diff --git a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/EventHubClientImpl.java b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/EventHubClientImpl.java
index a352172bb1dc..511ae19728c5 100644
--- a/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/EventHubClientImpl.java
+++ b/sdk/eventhubs/microsoft-azure-eventhubs/src/main/java/com/microsoft/azure/eventhubs/impl/EventHubClientImpl.java
@@ -8,24 +8,27 @@
import com.microsoft.azure.eventhubs.EventData;
import com.microsoft.azure.eventhubs.EventDataBatch;
import com.microsoft.azure.eventhubs.EventHubClient;
+import com.microsoft.azure.eventhubs.EventHubClientOptions;
import com.microsoft.azure.eventhubs.EventHubException;
import com.microsoft.azure.eventhubs.EventHubRuntimeInformation;
import com.microsoft.azure.eventhubs.EventPosition;
+import com.microsoft.azure.eventhubs.ITokenProvider;
import com.microsoft.azure.eventhubs.OperationCancelledException;
import com.microsoft.azure.eventhubs.PartitionReceiver;
import com.microsoft.azure.eventhubs.PartitionRuntimeInformation;
import com.microsoft.azure.eventhubs.PartitionSender;
import com.microsoft.azure.eventhubs.ReceiverOptions;
import com.microsoft.azure.eventhubs.RetryPolicy;
+import com.microsoft.azure.eventhubs.impl.MessagingFactory.MessagingFactoryBuilder;
import java.io.IOException;
-import java.security.InvalidKeyException;
-import java.security.NoSuchAlgorithmException;
+import java.net.URI;
import java.time.Duration;
import java.util.Date;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
@@ -49,18 +52,23 @@ public final class EventHubClientImpl extends ClientEntity implements EventHubCl
private CompletableFuture createSender;
- private EventHubClientImpl(final ConnectionStringBuilder connectionString, final ScheduledExecutorService executor) {
+ private EventHubClientImpl(final String eventHubName, final ScheduledExecutorService executor) {
super(StringUtil.getRandomString("EC"), null, executor);
- this.eventHubName = connectionString.getEventHubName();
+ this.eventHubName = eventHubName;
this.senderCreateSync = new Object();
}
public static CompletableFuture create(
final String connectionString, final RetryPolicy retryPolicy, final ScheduledExecutorService executor)
throws IOException {
+ if (StringUtil.isNullOrWhiteSpace(connectionString)) {
+ throw new IllegalArgumentException("Connection string cannot be null or empty");
+ }
+ Objects.requireNonNull(executor, "Executor cannot be null");
+
final ConnectionStringBuilder connStr = new ConnectionStringBuilder(connectionString);
- final EventHubClientImpl eventHubClient = new EventHubClientImpl(connStr, executor);
+ final EventHubClientImpl eventHubClient = new EventHubClientImpl(connStr.getEventHubName(), executor);
return MessagingFactory.createFromConnectionString(connectionString, retryPolicy, executor)
.thenApplyAsync(new Function() {
@@ -73,6 +81,38 @@ public EventHubClient apply(MessagingFactory factory) {
}, executor);
}
+ public static CompletableFuture create(
+ final URI endpoint,
+ final String eventHubName,
+ final ITokenProvider tokenProvider,
+ final ScheduledExecutorService executor,
+ final EventHubClientOptions options) throws IOException {
+ if (StringUtil.isNullOrWhiteSpace(endpoint.getHost())) {
+ throw new IllegalArgumentException("Endpoint must contain a hostname");
+ }
+ if (StringUtil.isNullOrWhiteSpace(eventHubName)) {
+ throw new IllegalArgumentException("Event hub name cannot be null or empty");
+ }
+ Objects.requireNonNull(tokenProvider, "Token provider cannot be null");
+
+ final EventHubClientImpl eventHubClient = new EventHubClientImpl(eventHubName, executor);
+ final MessagingFactoryBuilder builder = new MessagingFactoryBuilder(endpoint.getHost(), tokenProvider, executor);
+ if (options != null) {
+ builder.setOperationTimeout(options.getOperationTimeout()).setTransportType(options.getTransportType()).
+ setRetryPolicy(options.getRetryPolicy());
+ }
+
+ return builder.build()
+ .thenApplyAsync(new Function() {
+ @Override
+ public EventHubClient apply(MessagingFactory factory) {
+ eventHubClient.underlyingFactory = factory;
+ eventHubClient.timer = new Timer(factory);
+ return eventHubClient;
+ }
+ }, executor);
+ }
+
public String getEventHubName() {
return eventHubName;
}
@@ -250,38 +290,24 @@ public void accept(MessageSender a) {
@Override
public CompletableFuture getRuntimeInformation() {
- CompletableFuture future1 = null;
-
throwIfClosed();
Map request = new HashMap();
request.put(ClientConstants.MANAGEMENT_ENTITY_TYPE_KEY, ClientConstants.MANAGEMENT_EVENTHUB_ENTITY_TYPE);
request.put(ClientConstants.MANAGEMENT_ENTITY_NAME_KEY, this.eventHubName);
request.put(ClientConstants.MANAGEMENT_OPERATION_KEY, ClientConstants.READ_OPERATION_VALUE);
- future1 = this.addManagementToken(request);
-
- if (future1 == null) {
- future1 = managementWithRetry(request).thenComposeAsync(new Function