Skip to content

[Event Hubs Client] Shared Key Credential + Connection String Parser #16044

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jsquire merged 1 commit into from
Oct 23, 2020
Merged

[Event Hubs Client] Shared Key Credential + Connection String Parser #16044

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions ...e.Messaging.EventHubs.Processor/api/Azure.Messaging.EventHubs.Processor.netstandard2.0.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointSt
public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointStore, string consumerGroup, string connectionString, Azure.Messaging.EventHubs.EventProcessorClientOptions clientOptions) { }
public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointStore, string consumerGroup, string connectionString, string eventHubName) { }
public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointStore, string consumerGroup, string fullyQualifiedNamespace, string eventHubName, Azure.Core.TokenCredential credential, Azure.Messaging.EventHubs.EventProcessorClientOptions clientOptions = null) { }
public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointStore, string consumerGroup, string fullyQualifiedNamespace, string eventHubName, Azure.Messaging.EventHubs.EventHubsSharedAccessKeyCredential credential, Azure.Messaging.EventHubs.EventProcessorClientOptions clientOptions = null) { }
public EventProcessorClient(Azure.Storage.Blobs.BlobContainerClient checkpointStore, string consumerGroup, string connectionString, string eventHubName, Azure.Messaging.EventHubs.EventProcessorClientOptions clientOptions) { }
public new string ConsumerGroup { get { throw null; } }
public new string EventHubName { get { throw null; } }
Expand Down
57 changes: 57 additions & 0 deletions sdk/eventhub/Azure.Messaging.EventHubs.Processor/src/EventProcessorClient.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,6 +405,33 @@ public EventProcessorClient(BlobContainerClient checkpointStore,
StorageManager = CreateStorageManager(checkpointStore);
}

/// <summary>
/// Initializes a new instance of the <see cref="EventProcessorClient"/> class.
/// </summary>
///
/// <param name="checkpointStore">The client responsible for persisting checkpoints and processor state to durable storage. The associated container is expected to exist.</param>
/// <param name="consumerGroup">The name of the consumer group this processor is associated with. Events are read in the context of this group.</param>
/// <param name="fullyQualifiedNamespace">The fully qualified Event Hubs namespace to connect to. This is likely to be similar to <c>{yournamespace}.servicebus.windows.net</c>.</param>
/// <param name="eventHubName">The name of the specific Event Hub to associate the processor with.</param>
/// <param name="credential">The Event Hubs shared access key credential to use for authorization. Access controls may be specified by the Event Hubs namespace or the requested Event Hub, depending on Azure configuration.</param>
/// <param name="clientOptions">The set of options to use for this processor.</param>
///
/// <remarks>
/// The container associated with the <paramref name="checkpointStore" /> is expected to exist; the <see cref="EventProcessorClient" />
/// does not assume the ability to manage the storage account and is safe to run without permission to manage the storage account.
Comment thread
jsquire marked this conversation as resolved.
/// </remarks>
///
public EventProcessorClient(BlobContainerClient checkpointStore,
string consumerGroup,
string fullyQualifiedNamespace,
string eventHubName,
EventHubsSharedAccessKeyCredential credential,
EventProcessorClientOptions clientOptions = default) : base((clientOptions ?? DefaultClientOptions).CacheEventCount, consumerGroup, fullyQualifiedNamespace, eventHubName, credential, CreateOptions(clientOptions))
Copy link
Copy Markdown
Member

@christothes christothes Oct 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it ok that the first arg to the base ctor conditionally passes DefaultClientOptions but the last arg doesn't?

Copy link
Copy Markdown
Member Author

@jsquire jsquire Oct 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Scared me there. The first argument sets the batch size using either the specified option or the default size. The latter translates the derived options into the base type and passes it as the options set.

The CacheEventCount used for the batch size in the first argument does not exist as part of the base options, due to it being batch-driven natively and expecting that callers will need to specify the size of batches that they want to process.

{
Argument.AssertNotNull(checkpointStore, nameof(checkpointStore));
StorageManager = CreateStorageManager(checkpointStore);
}

/// <summary>
/// Initializes a new instance of the <see cref="EventProcessorClient"/> class.
/// </summary>
Expand Down Expand Up @@ -432,6 +459,36 @@ public EventProcessorClient(BlobContainerClient checkpointStore,
StorageManager = CreateStorageManager(checkpointStore);
}

/// <summary>
/// Initializes a new instance of the <see cref="EventProcessorClient"/> class.
/// </summary>
///
/// <param name="storageManager">Responsible for creation of checkpoints and for ownership claim.</param>
/// <param name="consumerGroup">The name of the consumer group this processor is associated with. Events are read in the context of this group.</param>
/// <param name="fullyQualifiedNamespace">The fully qualified Event Hubs namespace to connect to. This is likely to be similar to <c>{yournamespace}.servicebus.windows.net</c>.</param>
/// <param name="eventHubName">The name of the specific Event Hub to associate the processor with.</param>
/// <param name="cacheEventCount">The maximum number of events that will be read from the Event Hubs service and held in a local memory cache when reading is active and events are being emitted to an enumerator for processing.</param>
/// <param name="credential">A shared access key credential to satisfy base class requirements; this credential may not be <c>null</c> but will only be used in the case that <see cref="CreateConnection" /> has not been overridden.</param>
/// <param name="clientOptions">The set of options to use for this processor.</param>
///
/// <remarks>
/// This constructor is intended only to support functional testing and mocking; it should not be used for production scenarios.
/// </remarks>
///
internal EventProcessorClient(StorageManager storageManager,
string consumerGroup,
string fullyQualifiedNamespace,
string eventHubName,
int cacheEventCount,
EventHubsSharedAccessKeyCredential credential,
EventProcessorOptions clientOptions) : base(cacheEventCount, consumerGroup, fullyQualifiedNamespace, eventHubName, credential, clientOptions)
{
Argument.AssertNotNull(storageManager, nameof(storageManager));

DefaultStartingPosition = (clientOptions?.DefaultStartingPosition ?? DefaultStartingPosition);
StorageManager = storageManager;
}

/// <summary>
/// Initializes a new instance of the <see cref="EventProcessorClient"/> class.
/// </summary>
Expand Down
84 changes: 84 additions & 0 deletions ...thub/Azure.Messaging.EventHubs.Processor/tests/Processor/EventProcessorClientLiveTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,56 @@ public async Task EventsCanBeReadByOneProcessorClientUsingAnIdentityCredential()
}
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
///
[Test]
public async Task EventsCanBeReadByOneProcessorClientUsingTheSharedKeyCredential()
{
// Setup the environment.

await using EventHubScope scope = await EventHubScope.CreateAsync(2);
var connectionString = EventHubsTestEnvironment.Instance.BuildConnectionStringForEventHub(scope.EventHubName);

using var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(EventHubsTestEnvironment.Instance.TestExecutionTimeLimit);

// Send a set of events.

var sourceEvents = EventGenerator.CreateEvents(50).ToList();
var sentCount = await SendEvents(connectionString, sourceEvents, cancellationSource.Token);

Assert.That(sentCount, Is.EqualTo(sourceEvents.Count), "Not all of the source events were sent.");

// Attempt to read back the events.

var processedEvents = new ConcurrentDictionary<string, EventData>();
var completionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
var options = new EventProcessorOptions { LoadBalancingUpdateInterval = TimeSpan.FromMilliseconds(250) };
var processor = CreateProcessorWithSharedAccessKey(scope.ConsumerGroups.First(), scope.EventHubName, options: options);

processor.ProcessErrorAsync += CreateAssertingErrorHandler();
processor.ProcessEventAsync += CreateEventTrackingHandler(sentCount, processedEvents, completionSource, cancellationSource.Token);

await processor.StartProcessingAsync(cancellationSource.Token);

await Task.WhenAny(completionSource.Task, Task.Delay(Timeout.Infinite, cancellationSource.Token));
Assert.That(cancellationSource.IsCancellationRequested, Is.False, $"The cancellation token should not have been signaled. { processedEvents.Count } events were processed.");

await processor.StopProcessingAsync(cancellationSource.Token);
cancellationSource.Cancel();

// Validate the events that were processed.

foreach (var sourceEvent in sourceEvents)
{
var sourceId = sourceEvent.Properties[EventGenerator.IdPropertyName].ToString();
Assert.That(processedEvents.TryGetValue(sourceId, out var processedEvent), Is.True, $"The event with custom identifier [{ sourceId }] was not processed." );
Assert.That(sourceEvent.IsEquivalentTo(processedEvent), $"The event with custom identifier [{ sourceId }] did not match the corresponding processed event.");
}
}

/// <summary>
/// Verifies that the <see cref="EventProcessorClient" /> can read a set of published events.
/// </summary>
Expand Down Expand Up @@ -477,6 +527,29 @@ private EventProcessorClient CreateProcessorWithIdentity(string consumerGroup,
return new TestEventProcessorClient(storageManager, consumerGroup, EventHubsTestEnvironment.Instance.FullyQualifiedNamespace, eventHubName, credential, createConnection, options);
}

/// <summary>
/// Creates an <see cref="EventProcessorClient" /> that uses mock storage and
/// a connection based on an identity credential.
/// </summary>
///
/// <param name="consumerGroup">The consumer group for the processor.</param>
/// <param name="eventHubName">The name of the Event Hub for the processor.</param>
/// <param name="options">The set of client options to pass.</param>
///
/// <returns>The processor instance.</returns>
///
private EventProcessorClient CreateProcessorWithSharedAccessKey(string consumerGroup,
string eventHubName,
StorageManager storageManager = default,
EventProcessorOptions options = default)
{
var credential = new EventHubsSharedAccessKeyCredential(EventHubsTestEnvironment.Instance.SharedAccessKeyName, EventHubsTestEnvironment.Instance.SharedAccessKey);
EventHubConnection createConnection() => new EventHubConnection(EventHubsTestEnvironment.Instance.FullyQualifiedNamespace, eventHubName, credential);

storageManager ??= new InMemoryStorageManager(_=> {});
return new TestEventProcessorClient(storageManager, consumerGroup, EventHubsTestEnvironment.Instance.FullyQualifiedNamespace, eventHubName, credential, createConnection, options);
}

/// <summary>
/// Sends a set of events using a new producer to do so.
/// </summary>
Expand Down Expand Up @@ -575,6 +648,17 @@ public class TestEventProcessorClient : EventProcessorClient
{
private readonly Func<EventHubConnection> InjectedConnectionFactory;

internal TestEventProcessorClient(StorageManager storageManager,
string consumerGroup,
string fullyQualifiedNamespace,
string eventHubName,
EventHubsSharedAccessKeyCredential credential,
Func<EventHubConnection> connectionFactory,
EventProcessorOptions options) : base(storageManager, consumerGroup, fullyQualifiedNamespace, eventHubName, 100, credential, options)
{
InjectedConnectionFactory = connectionFactory;
}

internal TestEventProcessorClient(StorageManager storageManager,
string consumerGroup,
string fullyQualifiedNamespace,
Expand Down
15 changes: 10 additions & 5 deletions ...eventhub/Azure.Messaging.EventHubs.Processor/tests/Processor/EventProcessorClientTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,8 @@ public class EventProcessorClientTests
public void ConstructorsValidateTheConsumerGroup(string consumerGroup)
{
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), consumerGroup, "dummyConnection", new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentException>(), "The connection string constructor should validate the consumer group.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), consumerGroup, "dummyNamespace", "dummyEventHub", Mock.Of<TokenCredential>(), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentException>(), "The namespace constructor should validate the consumer group.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), consumerGroup, "dummyNamespace", "dummyEventHub", Mock.Of<TokenCredential>(), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentException>(), "The token credential constructor should validate the consumer group.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), consumerGroup, "dummyNamespace", "dummyEventHub", new EventHubsSharedAccessKeyCredential("key", "value"), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentException>(), "The shared key credential constructor should validate the consumer group.");
}

/// <summary>
Expand All @@ -51,7 +52,8 @@ public void ConstructorsValidateTheBlobContainerClient()
var fakeConnection = "Endpoint=sb://not-real.servicebus.windows.net/;SharedAccessKeyName=DummyKey;SharedAccessKey=[not_real];EntityPath=fake";

Assert.That(() => new EventProcessorClient(null, "consumerGroup", fakeConnection, new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentNullException>(), "The connection string constructor should validate the blob container client.");
Assert.That(() => new EventProcessorClient(null, "consumerGroup", "dummyNamespace", "dummyEventHub", Mock.Of<TokenCredential>(), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentNullException>(), "The namespace constructor should validate the blob container client.");
Assert.That(() => new EventProcessorClient(null, "consumerGroup", "dummyNamespace", "dummyEventHub", Mock.Of<TokenCredential>(), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentNullException>(), "The token credential constructor should validate the blob container client.");
Assert.That(() => new EventProcessorClient(null, "consumerGroup", "dummyNamespace", "dummyEventHub", new EventHubsSharedAccessKeyCredential("key", "value"), new EventProcessorClientOptions()), Throws.InstanceOf<ArgumentNullException>(), "The shared key credential constructor should validate the blob container client.");
}

/// <summary>
Expand All @@ -77,7 +79,8 @@ public void ConstructorsValidateTheConnectionString(string connectionString)
[TestCase("http://namspace.servciebus.windows.com")]
public void ConstructorValidatesTheNamespace(string constructorArgument)
{
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, constructorArgument, "dummy", Mock.Of<TokenCredential>()), Throws.InstanceOf<ArgumentException>());
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, constructorArgument, "dummy", Mock.Of<TokenCredential>()), Throws.InstanceOf<ArgumentException>(), "The token credential should validate.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, constructorArgument, "dummy", new EventHubsSharedAccessKeyCredential("key", "value")), Throws.InstanceOf<ArgumentException>(), "The shared key credential should validate.");
}

/// <summary>
Expand All @@ -89,7 +92,8 @@ public void ConstructorValidatesTheNamespace(string constructorArgument)
[TestCase("")]
public void ConstructorValidatesTheEventHub(string constructorArgument)
{
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", constructorArgument, Mock.Of<TokenCredential>()), Throws.InstanceOf<ArgumentException>());
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", constructorArgument, Mock.Of<TokenCredential>()), Throws.InstanceOf<ArgumentException>(), "The token credential should validate.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", constructorArgument, new EventHubsSharedAccessKeyCredential("key", "value")), Throws.InstanceOf<ArgumentException>(), "The shared key credential should validate.");
}

/// <summary>
Expand All @@ -99,7 +103,8 @@ public void ConstructorValidatesTheEventHub(string constructorArgument)
[Test]
public void ConstructorValidatesTheCredential()
{
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", "hubName", default(TokenCredential)), Throws.ArgumentNullException);
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", "hubName", default(TokenCredential)), Throws.ArgumentNullException, "The token credential should validate.");
Assert.That(() => new EventProcessorClient(Mock.Of<BlobContainerClient>(), EventHubConsumerClient.DefaultConsumerGroupName, "namespace", "hubName", default(EventHubsSharedAccessKeyCredential)), Throws.ArgumentNullException, "The shared key credential should validate.");
}

/// <summary>
Expand Down
Loading