Defer SwitchIfEmpty Return Values (Azure#24621)

Defer SwitchIfEmpty Return Values

Author: alzimmermsft

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ClaimsBasedSecurityChannel.java

@@ -77,9 +77,9 @@ public Mono<OffsetDateTime> authorize(String tokenAudience, String scopes) {
                                 sink.error(error);
                             }
                         })
-                        .switchIfEmpty(Mono.error(new AmqpException(true, String.format(
+                        .switchIfEmpty(Mono.defer(() -> Mono.error(new AmqpException(true, String.format(
                             "No response received from CBS node. tokenAudience: '%s'. scopes: '%s'",
-                            tokenAudience, scopes), channel.getErrorContext())));
+                            tokenAudience, scopes), channel.getErrorContext()))));
                 }));
     }
 

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ManagementChannel.java

@@ -47,9 +47,9 @@ public Mono<AmqpAnnotatedMessage> send(AmqpAnnotatedMessage message) {
             return channel.sendWithAck(protonJMessage)
                 .handle((Message responseMessage, SynchronousSink<AmqpAnnotatedMessage> sink) ->
                     handleResponse(responseMessage, sink, channel.getErrorContext()))
-                .switchIfEmpty(Mono.error(new AmqpException(true, String.format(
+                .switchIfEmpty(Mono.defer(() -> Mono.error(new AmqpException(true, String.format(
                     "entityPath[%s] No response received from management channel.", entityPath),
-                    channel.getErrorContext())));
+                    channel.getErrorContext()))));
         }));
     }
 
@@ -62,9 +62,9 @@ public Mono<AmqpAnnotatedMessage> send(AmqpAnnotatedMessage message, DeliveryOut
             return channel.sendWithAck(protonJMessage, protonJDeliveryState)
                 .handle((Message responseMessage, SynchronousSink<AmqpAnnotatedMessage> sink) ->
                     handleResponse(responseMessage, sink, channel.getErrorContext()))
-                .switchIfEmpty(Mono.error(new AmqpException(true, String.format(
+                .switchIfEmpty(Mono.defer(() -> Mono.error(new AmqpException(true, String.format(
                     "entityPath[%s] outcome[%s] No response received from management channel.", entityPath,
-                    deliveryOutcome.getDeliveryState()), channel.getErrorContext())));
+                    deliveryOutcome.getDeliveryState()), channel.getErrorContext()))));
         }));
     }
 
@@ -114,7 +114,8 @@ private void handleResponse(Message response, SynchronousSink<AmqpAnnotatedMessa
     private Mono<Void> isAuthorized() {
         return tokenManager.getAuthorizationResults()
             .next()
-            .switchIfEmpty(Mono.error(new AmqpException(false, "Did not get response from tokenManager: " + entityPath, getErrorContext())))
+            .switchIfEmpty(Mono.defer(() -> Mono.error(
+                new AmqpException(false, "Did not get response from tokenManager: " + entityPath, getErrorContext()))))
             .handle((response, sink) -> {
                 if (RequestResponseUtils.isSuccessful(response)) {
                     sink.complete();

sdk/core/azure-core-experimental/src/main/java/com/azure/core/experimental/implementation/AccessTokenCacheImpl.java

@@ -143,7 +143,7 @@ private Function<Signal<AccessToken>, Mono<? extends AccessToken>> processTokenR
             } else if (signal.isOnError() && error != null) { // ERROR
                 logger.error(refreshLog(cache, now, "Failed to acquire a new access token"));
                 nextTokenRefresh = OffsetDateTime.now().plus(REFRESH_DELAY);
-                return fallback.switchIfEmpty(Mono.error(error));
+                return fallback.switchIfEmpty(Mono.defer(() -> Mono.error(error)));
             } else { // NO REFRESH
                 sinksOne.tryEmitEmpty();
                 return fallback;

sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClient.java

@@ -145,7 +145,7 @@ private static Mono<ByteString> toByteString(Flux<ByteBuffer> bbFlux) {
                     throw Exceptions.propagate(ioe);
                 }
             }).map(b -> ByteString.of(b.readByteArray())), okio.Buffer::clear)
-            .switchIfEmpty(EMPTY_BYTE_STRING_MONO);
+            .switchIfEmpty(Mono.defer(() -> EMPTY_BYTE_STRING_MONO));
     }
 
     private static class OkHttpCallback implements okhttp3.Callback {

sdk/core/azure-core/src/main/java/com/azure/core/credential/SimpleTokenCache.java

@@ -94,7 +94,7 @@ public Mono<AccessToken> getToken() {
                             } else if (signal.isOnError() && error != null) { // ERROR
                                 logger.error(refreshLog(cache, now, "Failed to acquire a new access token"));
                                 nextTokenRefresh = OffsetDateTime.now().plus(REFRESH_DELAY);
-                                return fallback.switchIfEmpty(Mono.error(error));
+                                return fallback.switchIfEmpty(Mono.defer(() -> Mono.error(error)));
                             } else { // NO REFRESH
                                 sinksOne.tryEmitEmpty();
                                 return fallback;

sdk/core/azure-core/src/main/java/com/azure/core/implementation/AccessTokenCache.java

@@ -167,7 +167,7 @@ private Function<Signal<AccessToken>, Mono<? extends AccessToken>> processTokenR
             } else if (signal.isOnError() && error != null) { // ERROR
                 logger.error(refreshLog(cache, now, "Failed to acquire a new access token"));
                 nextTokenRefresh = OffsetDateTime.now().plus(REFRESH_DELAY);
-                return fallback.switchIfEmpty(Mono.error(error));
+                return fallback.switchIfEmpty(Mono.defer(() -> Mono.error(error)));
             } else { // NO REFRESH
                 sinksOne.tryEmitEmpty();
                 return fallback;