AmqpChannelProcessor add constructor overload that creates logger and…

… deprecate old one
Azure · Dec 14, 2021 · 6425523 · 6425523
1 parent 5a18454
commit 6425523
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 7 deletions.
diff --git a/...zure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java b/...zure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java
@@ -18,6 +18,7 @@
 import reactor.core.publisher.Operators;
 
 import java.time.Duration;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.RejectedExecutionException;
@@ -26,6 +27,8 @@
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import java.util.function.Function;
 
+import static com.azure.core.amqp.implementation.AmqpLoggingUtils.createContextWithConnectionId;
+import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY;
 import static com.azure.core.amqp.implementation.ClientConstants.INTERVAL_KEY;
 
 public class AmqpChannelProcessor<T> extends Mono<T> implements Processor<T, T>, CoreSubscriber<T>, Disposable {
@@ -54,6 +57,10 @@ public class AmqpChannelProcessor<T> extends Mono<T> implements Processor<T, T>,
     private volatile Disposable connectionSubscription;
     private volatile Disposable retrySubscription;
 
+    /**
+     * @deprecated Use constructor overload that does not take {@link ClientLogger}
+     */
+    @Deprecated
     public AmqpChannelProcessor(String fullyQualifiedNamespace, String entityPath,
         Function<T, Flux<AmqpEndpointState>> endpointStatesFunction, AmqpRetryPolicy retryPolicy, ClientLogger logger) {
         this.endpointStatesFunction = Objects.requireNonNull(endpointStatesFunction,
@@ -64,6 +71,20 @@ public AmqpChannelProcessor(String fullyQualifiedNamespace, String entityPath,
         this.errorContext = new AmqpErrorContext(fullyQualifiedNamespace);
     }
 
+
+    public AmqpChannelProcessor(String fullyQualifiedNamespace, String entityPath, String connectionId,
+                                Function<T, Flux<AmqpEndpointState>> endpointStatesFunction, AmqpRetryPolicy retryPolicy) {
+        this.endpointStatesFunction = Objects.requireNonNull(endpointStatesFunction,
+            "'endpointStates' cannot be null.");
+        this.retryPolicy = Objects.requireNonNull(retryPolicy, "'retryPolicy' cannot be null.");
+
+        Map<String, Object> loggingContext = createContextWithConnectionId(connectionId);
+        loggingContext.put(ENTITY_PATH_KEY, Objects.requireNonNull(entityPath, "'entityPath' cannot be null."));
+        this.logger = new ClientLogger(AmqpChannelProcessor.class, loggingContext);
+
+        this.errorContext = new AmqpErrorContext(fullyQualifiedNamespace);
+    }
+
     @Override
     public void onSubscribe(Subscription subscription) {
         if (Operators.setOnce(UPSTREAM, this, subscription)) {

diff --git a/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java b/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java
@@ -412,8 +412,8 @@ protected AmqpChannelProcessor<RequestResponseChannel> createRequestResponseChan
         loggingContext.put(ENTITY_PATH_KEY, entityPath);
 
         return createChannel
-            .subscribeWith(new AmqpChannelProcessor<>(connectionId, entityPath,
-                channel -> channel.getEndpointStates(), retryPolicy, new ClientLogger(RequestResponseChannel.class, loggingContext)));
+            .subscribeWith(new AmqpChannelProcessor<>(getFullyQualifiedNamespace(), entityPath, connectionId,
+                channel -> channel.getEndpointStates(), retryPolicy));
     }
 
     @Override

diff --git a/...-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java b/...-core-amqp/src/test/java/com/azure/core/amqp/implementation/AmqpChannelProcessorTest.java
@@ -8,8 +8,6 @@
 import com.azure.core.amqp.exception.AmqpErrorCondition;
 import com.azure.core.amqp.exception.AmqpErrorContext;
 import com.azure.core.amqp.exception.AmqpException;
-import com.azure.core.amqp.implementation.handler.LinkHandlerTest;
-import com.azure.core.util.logging.ClientLogger;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
@@ -58,8 +56,8 @@ class AmqpChannelProcessorTest {
     void setup() {
         mocksCloseable = MockitoAnnotations.openMocks(this);
 
-        channelProcessor = new AmqpChannelProcessor<>("connection-test", "test-path",
-            TestObject::getStates, retryPolicy, new ClientLogger(LinkHandlerTest.class));
+        channelProcessor = new AmqpChannelProcessor<>("namespace-test", "test-path", "connection-test",
+            TestObject::getStates, retryPolicy);
     }
 
     @AfterEach

diff --git a/...ure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ManagementChannelTest.java b/...ure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ManagementChannelTest.java
@@ -95,7 +95,7 @@ public void setup(TestInfo testInfo) {
 
         final AmqpChannelProcessor<RequestResponseChannel> requestResponseMono =
             Mono.defer(() -> Mono.just(requestResponseChannel)).subscribeWith(new AmqpChannelProcessor<>(
-                "foo", "bar", RequestResponseChannel::getEndpointStates, retryPolicy, new ClientLogger(ManagementChannelTest.class)));
+                "foo", "bar", "baz", RequestResponseChannel::getEndpointStates, retryPolicy));
 
         when(tokenManager.authorize()).thenReturn(Mono.just(1000L));
         when(tokenManager.getAuthorizationResults()).thenReturn(tokenProviderResults.flux());