Skip to content

Commit 11cef77

Browse files
anuchandyanuchandy
authored
Removing the dead code in ServiceBusReceiverAsyncClient attempting to deal with session scenario when there is not going to be one (Azure#33937)
1 parent 190c887 commit 11cef77

File tree

5 files changed

+58
-54
lines changed

5 files changed

+58
-54
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -615,7 +615,7 @@ private LinkSubscription<AmqpReceiveLink> getSubscription(String linkName, Strin
615615

616616
receiver.open();
617617

618-
final ReactorReceiver reactorReceiver = createConsumer(entityPath, receiver, receiveLinkHandler,
618+
final AmqpReceiveLink reactorReceiver = createConsumer(entityPath, receiver, receiveLinkHandler,
619619
tokenManager, provider);
620620

621621
final Disposable subscription = reactorReceiver.getEndpointStates().subscribe(state -> {

‎sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ReceiverOptions.java

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -79,15 +79,6 @@ boolean isAutoLockRenewEnabled() {
7979
return maxLockRenewDuration != null && !maxLockRenewDuration.isZero() && !maxLockRenewDuration.isNegative();
8080
}
8181

82-
/**
83-
* Gets whether or not the receiver is a session-aware receiver.
84-
*
85-
* @return true if it is a session-aware receiver; false otherwise.
86-
*/
87-
boolean isSessionReceiver() {
88-
return sessionId != null || maxConcurrentSessions != null;
89-
}
90-
9182
/**
9283
* Gets whether or not this receiver should roll over when a session has completed processing.
9384
*

‎sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java

Lines changed: 28 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,7 @@ public final class ServiceBusReceiverAsyncClient implements AutoCloseable {
236236
private final MessageSerializer messageSerializer;
237237
private final Runnable onClientClose;
238238
private final ServiceBusSessionManager sessionManager;
239+
private final boolean isSessionEnabled;
239240
private final Semaphore completionLock = new Semaphore(1);
240241
private final String identifier;
241242

@@ -268,6 +269,12 @@ public final class ServiceBusReceiverAsyncClient implements AutoCloseable {
268269
this.instrumentation = Objects.requireNonNull(instrumentation, "'tracer' cannot be null");
269270
this.messageSerializer = Objects.requireNonNull(messageSerializer, "'messageSerializer' cannot be null.");
270271
this.onClientClose = Objects.requireNonNull(onClientClose, "'onClientClose' cannot be null.");
272+
this.sessionManager = null;
273+
if (receiverOptions.getSessionId() != null || receiverOptions.getMaxConcurrentSessions() != null) {
274+
// Assert the internal invariant for above 'sessionManager = null' i.e, session-unaware call-sites should not set these options.
275+
throw new IllegalStateException("Session-specific options are not expected to be present on a client for session unaware entity.");
276+
}
277+
this.isSessionEnabled = false;
271278

272279
this.managementNodeLocks = new LockContainer<>(cleanupInterval);
273280
this.renewalContainer = new LockContainer<>(Duration.ofMinutes(2), renewal -> {
@@ -278,7 +285,6 @@ public final class ServiceBusReceiverAsyncClient implements AutoCloseable {
278285
renewal.close();
279286
});
280287

281-
this.sessionManager = null;
282288
this.identifier = identifier;
283289
this.tracer = instrumentation.getTracer();
284290
this.trackSettlementSequenceNumber = instrumentation.startTrackingSettlementSequenceNumber();
@@ -298,6 +304,7 @@ public final class ServiceBusReceiverAsyncClient implements AutoCloseable {
298304
this.messageSerializer = Objects.requireNonNull(messageSerializer, "'messageSerializer' cannot be null.");
299305
this.onClientClose = Objects.requireNonNull(onClientClose, "'onClientClose' cannot be null.");
300306
this.sessionManager = Objects.requireNonNull(sessionManager, "'sessionManager' cannot be null.");
307+
this.isSessionEnabled = true;
301308

302309
this.managementNodeLocks = new LockContainer<>(cleanupInterval);
303310
this.renewalContainer = new LockContainer<>(Duration.ofMinutes(2), renewal -> {
@@ -865,7 +872,7 @@ Flux<ServiceBusMessageContext> receiveMessagesWithContext(int highTide) {
865872
final Flux<ServiceBusMessageContext> messageFluxWithTracing = new FluxTrace(messageFlux, instrumentation);
866873
final Flux<ServiceBusMessageContext> withAutoLockRenewal;
867874

868-
if (!receiverOptions.isSessionReceiver() && receiverOptions.isAutoLockRenewEnabled()) {
875+
if (!isSessionEnabled && receiverOptions.isAutoLockRenewEnabled()) {
869876
withAutoLockRenewal = new FluxAutoLockRenew(messageFluxWithTracing, receiverOptions,
870877
renewalContainer, this::renewMessageLock);
871878
} else {
@@ -1032,15 +1039,15 @@ public Mono<OffsetDateTime> renewMessageLock(ServiceBusReceivedMessage message)
10321039
if (isDisposed.get()) {
10331040
return monoError(LOGGER, new IllegalStateException(
10341041
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "renewMessageLock")));
1042+
} else if (isSessionEnabled) {
1043+
final String errorMessage = "Renewing message lock is an invalid operation when working with sessions.";
1044+
return monoError(LOGGER, new IllegalStateException(errorMessage));
10351045
} else if (Objects.isNull(message)) {
10361046
return monoError(LOGGER, new NullPointerException("'message' cannot be null."));
10371047
} else if (Objects.isNull(message.getLockToken())) {
10381048
return monoError(LOGGER, new NullPointerException("'message.getLockToken()' cannot be null."));
10391049
} else if (message.getLockToken().isEmpty()) {
10401050
return monoError(LOGGER, new IllegalArgumentException("'message.getLockToken()' cannot be empty."));
1041-
} else if (receiverOptions.isSessionReceiver()) {
1042-
final String errorMessage = "Renewing message lock is an invalid operation when working with sessions.";
1043-
return monoError(LOGGER, new IllegalStateException(errorMessage));
10441051
}
10451052

10461053
return tracer.traceMonoWithLink("ServiceBus.renewMessageLock", renewMessageLock(message.getLockToken()), message, message.getContext())
@@ -1088,15 +1095,15 @@ public Mono<Void> renewMessageLock(ServiceBusReceivedMessage message, Duration m
10881095
if (isDisposed.get()) {
10891096
return monoError(LOGGER, new IllegalStateException(
10901097
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "getAutoRenewMessageLock")));
1098+
} else if (isSessionEnabled) {
1099+
final String errorMessage = "Renewing message lock is an invalid operation when working with sessions.";
1100+
return monoError(LOGGER, new IllegalStateException(errorMessage));
10911101
} else if (Objects.isNull(message)) {
10921102
return monoError(LOGGER, new NullPointerException("'message' cannot be null."));
10931103
} else if (Objects.isNull(message.getLockToken())) {
10941104
return monoError(LOGGER, new NullPointerException("'message.getLockToken()' cannot be null."));
10951105
} else if (message.getLockToken().isEmpty()) {
10961106
return monoError(LOGGER, new IllegalArgumentException("'message.getLockToken()' cannot be empty."));
1097-
} else if (receiverOptions.isSessionReceiver()) {
1098-
return monoError(LOGGER, new IllegalStateException(
1099-
String.format("Cannot renew message lock [%s] for a session receiver.", message.getLockToken())));
11001107
} else if (maxLockRenewalDuration == null) {
11011108
return monoError(LOGGER, new NullPointerException("'maxLockRenewalDuration' cannot be null."));
11021109
} else if (maxLockRenewalDuration.isNegative()) {
@@ -1485,6 +1492,9 @@ deadLetterErrorDescription, propertiesToModify, sessionId, getLinkName(sessionId
14851492
}
14861493

14871494
private ServiceBusAsyncConsumer getOrCreateConsumer() {
1495+
if (isSessionEnabled) {
1496+
throw LOGGER.logExceptionAsError(new IllegalStateException("The ServiceBusAsyncConsumer is expected to work only with session unaware entity."));
1497+
}
14881498
final ServiceBusAsyncConsumer existing = consumer.get();
14891499
if (existing != null) {
14901500
return existing;
@@ -1499,19 +1509,14 @@ private ServiceBusAsyncConsumer getOrCreateConsumer() {
14991509
// The Mono, when subscribed, creates a ServiceBusReceiveLink in the ServiceBusAmqpConnection emitted by the connectionProcessor
15001510
//
15011511
final Mono<ServiceBusReceiveLink> receiveLinkMono = connectionProcessor.flatMap(connection -> {
1502-
if (receiverOptions.isSessionReceiver()) {
1503-
return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(),
1504-
null, entityType, identifier, receiverOptions.getSessionId());
1505-
} else {
1506-
return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(),
1507-
null, entityType, identifier);
1508-
}
1512+
return connection.createReceiveLink(linkName, entityPath, receiverOptions.getReceiveMode(),
1513+
null, entityType, identifier);
15091514
}).doOnNext(next -> {
15101515
LOGGER.atVerbose()
15111516
.addKeyValue(LINK_NAME_KEY, linkName)
15121517
.addKeyValue(ENTITY_PATH_KEY, next.getEntityPath())
15131518
.addKeyValue("mode", receiverOptions.getReceiveMode())
1514-
.addKeyValue("isSessionEnabled", receiverOptions.isSessionReceiver())
1519+
.addKeyValue("isSessionEnabled", false)
15151520
.addKeyValue(ENTITY_TYPE_KEY, entityType)
15161521
.log("Created consumer for Service Bus resource.");
15171522
});
@@ -1576,10 +1581,8 @@ private ServiceBusAsyncConsumer getOrCreateConsumer() {
15761581
* @return The name of the receive link, or null of it has not connected via a receive link.
15771582
*/
15781583
private String getLinkName(String sessionId) {
1579-
if (sessionManager != null && !CoreUtils.isNullOrEmpty(sessionId)) {
1580-
return sessionManager.getLinkName(sessionId);
1581-
} else if (!CoreUtils.isNullOrEmpty(sessionId) && !receiverOptions.isSessionReceiver()) {
1582-
return null;
1584+
if (!CoreUtils.isNullOrEmpty(sessionId)) {
1585+
return isSessionEnabled ? sessionManager.getLinkName(sessionId) : null;
15831586
} else {
15841587
final ServiceBusAsyncConsumer existing = consumer.get();
15851588
return existing != null ? existing.getLinkName() : null;
@@ -1590,12 +1593,10 @@ Mono<OffsetDateTime> renewSessionLock(String sessionId) {
15901593
if (isDisposed.get()) {
15911594
return monoError(LOGGER, new IllegalStateException(
15921595
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "renewSessionLock")));
1593-
} else if (!receiverOptions.isSessionReceiver()) {
1596+
} else if (!isSessionEnabled) {
15941597
return monoError(LOGGER, new IllegalStateException("Cannot renew session lock on a non-session receiver."));
15951598
}
1596-
final String linkName = sessionManager != null
1597-
? sessionManager.getLinkName(sessionId)
1598-
: null;
1599+
final String linkName = sessionManager.getLinkName(sessionId);
15991600

16001601
return tracer.traceMono("ServiceBus.renewSessionLock", connectionProcessor
16011602
.flatMap(connection -> connection.getManagementNode(entityPath, entityType))
@@ -1607,7 +1608,7 @@ Mono<Void> renewSessionLock(String sessionId, Duration maxLockRenewalDuration) {
16071608
if (isDisposed.get()) {
16081609
return monoError(LOGGER, new IllegalStateException(
16091610
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "renewSessionLock")));
1610-
} else if (!receiverOptions.isSessionReceiver()) {
1611+
} else if (!isSessionEnabled) {
16111612
return monoError(LOGGER, new IllegalStateException(
16121613
"Cannot renew session lock on a non-session receiver."));
16131614
} else if (maxLockRenewalDuration == null) {
@@ -1632,7 +1633,7 @@ Mono<Void> setSessionState(String sessionId, byte[] sessionState) {
16321633
if (isDisposed.get()) {
16331634
return monoError(LOGGER, new IllegalStateException(
16341635
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "setSessionState")));
1635-
} else if (!receiverOptions.isSessionReceiver()) {
1636+
} else if (!isSessionEnabled) {
16361637
return monoError(LOGGER, new IllegalStateException("Cannot set session state on a non-session receiver."));
16371638
}
16381639
final String linkName = sessionManager != null
@@ -1649,7 +1650,7 @@ Mono<byte[]> getSessionState(String sessionId) {
16491650
if (isDisposed.get()) {
16501651
return monoError(LOGGER, new IllegalStateException(
16511652
String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "getSessionState")));
1652-
} else if (!receiverOptions.isSessionReceiver()) {
1653+
} else if (!isSessionEnabled) {
16531654
return monoError(LOGGER, new IllegalStateException("Cannot get session state on a non-session receiver."));
16541655
}
16551656

‎sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorSession.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,7 @@ private Mono<ServiceBusReceiveLink> createConsumer(String linkName, String entit
209209
receiverSettleMode)
210210
.cast(ServiceBusReceiveLink.class));
211211
} else {
212-
return createConsumer(linkName, entityPath, timeout, retry, filter, linkProperties,
212+
return super.createConsumer(linkName, entityPath, timeout, retry, filter, linkProperties,
213213
null, senderSettleMode, receiverSettleMode).cast(ServiceBusReceiveLink.class);
214214
}
215215
}

0 commit comments

Comments
 (0)