forked from Azure/azure-sdk-for-java
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds support for Event Hubs on v2 stack (Azure#41646)
* [Event hubs] V2 stack onboarding (Azure#40435) * Defining the opt-in flags for event hubs clients on v2 stack, and placeholder for v2 connection cache * move v2 support class outside the builder * Init V2StackSupport in builder to see if spring fails * Adding ConnectionCacheWrapper to delegate to v1 EventHubConnectionProcessor or v2 ReactorConnectionCache, EventHubReactorSession to inspect v2 and create v2 ReceiveLinkHandler2 and update V2StackSupport to create ReactorConnectionCache * use unreleased azure-core-amqp * Integrating the MessageFlux to EventHubsPartitionAsyncConsumer * Integrating the WindowedSubscriber to EventHubConsumerClient * Wire the v2 opt-in in EventHubClientBuilder. * remove unused import in EventHubConsumerAsyncClientTest * Use released version of azure-core-amqp, update changelog about v2 stack integration * Prepare beta release (Azure#41586) * Increment package versions for eventhubs releases (Azure#41595) --------- Co-authored-by: Anu Thomas Chandy <anuamd@hotmail.com> Co-authored-by: Azure SDK Bot <53356347+azure-sdk@users.noreply.github.com>
- Loading branch information
3 people
authored
Aug 26, 2024
1 parent
ac8a9e3
commit 4e61bd6
Showing
29 changed files
with
672 additions
and
127 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
88 changes: 88 additions & 0 deletions
88
...ssaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/ConnectionCacheWrapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| package com.azure.messaging.eventhubs; | ||
|
|
||
| import com.azure.core.amqp.AmqpRetryOptions; | ||
| import com.azure.core.amqp.implementation.ReactorConnectionCache; | ||
| import com.azure.messaging.eventhubs.implementation.EventHubAmqpConnection; | ||
| import com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor; | ||
| import com.azure.messaging.eventhubs.implementation.EventHubManagementNode; | ||
| import com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection; | ||
| import reactor.core.Disposable; | ||
| import reactor.core.publisher.Mono; | ||
|
|
||
| import java.util.Objects; | ||
|
|
||
| import static com.azure.core.amqp.implementation.RetryUtil.withRetry; | ||
|
|
||
| /** | ||
| * Temporary type to support connection cache either in v1 or v2 stack. | ||
| * v2 underlying connection cache is {@link ReactorConnectionCache} | ||
| * v1 underlying connection cache is {@link EventHubConnectionProcessor} | ||
| */ | ||
| final class ConnectionCacheWrapper implements Disposable { | ||
| private final boolean isV2; | ||
| private final ReactorConnectionCache<EventHubReactorAmqpConnection> cache; | ||
| private final EventHubConnectionProcessor processor; | ||
|
|
||
| ConnectionCacheWrapper(ReactorConnectionCache<EventHubReactorAmqpConnection> cache) { | ||
| this.isV2 = true; | ||
| this.cache = Objects.requireNonNull(cache, "'cache' cannot be null."); | ||
| this.processor = null; | ||
| } | ||
|
|
||
| ConnectionCacheWrapper(EventHubConnectionProcessor processor) { | ||
| this.isV2 = false; | ||
| this.processor = Objects.requireNonNull(processor, "'processor' cannot be null."); | ||
| this.cache = null; | ||
| } | ||
|
|
||
| boolean isV2() { | ||
| return isV2; | ||
| } | ||
|
|
||
| Mono<EventHubAmqpConnection> getConnection() { | ||
| return isV2 ? cache.get().cast(EventHubAmqpConnection.class) : processor; | ||
| } | ||
|
|
||
| String getFullyQualifiedNamespace() { | ||
| return isV2 ? cache.getFullyQualifiedNamespace() : processor.getFullyQualifiedNamespace(); | ||
| } | ||
|
|
||
| String getEventHubName() { | ||
| return isV2 ? cache.getEntityPath() : processor.getEventHubName(); | ||
| } | ||
|
|
||
| AmqpRetryOptions getRetryOptions() { | ||
| return isV2 ? cache.getRetryOptions() : processor.getRetryOptions(); | ||
| } | ||
|
|
||
| boolean isChannelClosed() { | ||
| return isV2 ? cache.isCurrentConnectionClosed() : processor.isChannelClosed(); | ||
| } | ||
|
|
||
| Mono<EventHubManagementNode> getManagementNodeWithRetries() { | ||
| if (isV2) { | ||
| final AmqpRetryOptions retryOptions = cache.getRetryOptions(); | ||
| return withRetry(cache.get().flatMap(EventHubReactorAmqpConnection::getManagementNode), | ||
| retryOptions, "Time out creating management node."); | ||
| } else { | ||
| return processor.getManagementNodeWithRetries(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDisposed() { | ||
| return isV2 ? cache.isDisposed() : processor.isDisposed(); | ||
| } | ||
|
|
||
| @Override | ||
| public void dispose() { | ||
| if (isV2) { | ||
| cache.dispose(); | ||
| } else { | ||
| processor.dispose(); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.