diff --git a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md index 2272155fb763..ed54c3b8d0a0 100644 --- a/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md +++ b/sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md @@ -4,7 +4,7 @@ ### Features Added -- Added `listSessions()` and `listSessions(OffsetDateTime sessionStateUpdatedAfter)` to `ServiceBusSessionReceiverAsyncClient` (returning `PagedFlux`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable`). The no-arg overload returns sessions with active messages; the `sessionStateUpdatedAfter` overload returns sessions whose session state was updated after the given timestamp. Implements the `com.microsoft:get-message-sessions` AMQP management operation. ([#48956](https://github.com/Azure/azure-sdk-for-java/pull/48956)) +- Added `listSessions()` and `listSessions(OffsetDateTime sessionStateUpdatedAfter)` to `ServiceBusSessionReceiverAsyncClient` (returning `PagedFlux`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable`). The no-arg overload returns sessions with active messages or stored session state; the `sessionStateUpdatedAfter` overload returns sessions whose session state was set or updated after the given timestamp. Implements the `com.microsoft:get-message-sessions` AMQP management operation. ([#48956](https://github.com/Azure/azure-sdk-for-java/pull/48956)) - Added `getSqlFilterCount()` and `getCorrelationFilterCount()` to `TopicRuntimeProperties`, exposing the total number of SQL filters and correlation filters across all of a topic's subscriptions. - Added `ServiceBusServiceVersion.V2024_05` and made it the latest version. The administration client now uses `api-version=2024-05` by default, which is required for the topic filter counts above. diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java index 4d45b517d086..f3eb733eafbe 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClient.java @@ -313,11 +313,11 @@ private Mono acquireSpecificOrNextSession(String } /** - * Lists the IDs of sessions that have active messages in this entity. + * Lists the IDs of sessions that have active messages or stored session state in this entity. * - *

Only sessions with active messages in the queue or subscription are returned. - * Sessions on the dead-letter queue or sessions having only a session state (but no messages) - * are not returned.

+ *

Sessions with active messages or stored session state in the queue or subscription are + * returned. Sessions with neither are excluded. Sessions on the dead-letter queue are not + * returned.

* *

The returned {@link PagedFlux} fetches additional pages from the broker on demand using * cursor-based pagination (server-returned {@code skip} plus {@code lastSessionId} of the @@ -330,26 +330,19 @@ private Mono acquireSpecificOrNextSession(String @ServiceMethod(returns = ReturnType.COLLECTION) public PagedFlux listSessions() { // Wire value matches Track 1's SessionBrowser.MAXDATE so the broker switches into the - // active-messages mode it has historically been validated against. - return listSessionsInternal(ManagementConstants.ACTIVE_MESSAGES_SENTINEL); + // default listing mode for sessions with active messages or stored session state. + return listSessionsInternal(ManagementConstants.DEFAULT_LISTING_SENTINEL); } /** - * Lists the IDs of sessions whose state was updated after the specified time. + * Lists the IDs of sessions whose state was set or updated after the specified time. * *

The returned {@link PagedFlux} fetches additional pages from the broker on demand using * cursor-based pagination (server-returned {@code skip} plus {@code lastSessionId} of the * previous page) and terminates when the broker returns a page smaller than the requested page * size (a short or empty page signals the end). The default page size is 100; callers can * request a different size via {@link PagedFlux#byPage(int)}.

- * - *

Values at or beyond the active-messages sentinel value - * ({@code new Date(253402300800000L)}, rendered by {@code OffsetDateTime.toString()} as - * {@code +10000-01-01T00:00Z}, matching Track 1's {@code SessionBrowser.MAXDATE}) are clamped - * to that sentinel and behave the same as {@link #listSessions()}, returning sessions that - * have active messages.

- * - * @param sessionStateUpdatedAfter Only sessions whose session state was updated after this time are returned. + * @param sessionStateUpdatedAfter Only sessions whose session state was set or updated after this time are returned. * @return A {@link PagedFlux} of session ID strings. * @throws NullPointerException if {@code sessionStateUpdatedAfter} is null. */ diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java index 6bb7f1130efc..603025d39b90 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverClient.java @@ -212,7 +212,9 @@ public ServiceBusReceiverClient acceptSession(String sessionId) { } /** - * Lists the IDs of sessions that have active messages in this entity. + * Lists the IDs of sessions that have active messages or stored session state in this entity. + * + *

Sessions with neither active messages nor stored session state are excluded.

* *

The returned {@link PagedIterable} fetches additional pages from the broker on demand; * iterate the {@code PagedIterable} (or call {@link PagedIterable#stream()}) to receive every @@ -228,21 +230,14 @@ public PagedIterable listSessions() { } /** - * Lists the IDs of sessions whose state was updated after the specified time. + * Lists the IDs of sessions whose state was set or updated after the specified time. * *

The returned {@link PagedIterable} fetches additional pages from the broker on demand; * iterate the {@code PagedIterable} (or call {@link PagedIterable#stream()}) to receive every * session ID. Pages are fetched lazily as the iterator advances. The default page size is 100; * callers can request a different size via {@link PagedIterable#iterableByPage(int)} (or the * equivalent on the underlying {@code PagedFlux}).

- * - *

Values at or beyond the active-messages sentinel value - * ({@code new Date(253402300800000L)}, rendered by {@code OffsetDateTime.toString()} as - * {@code +10000-01-01T00:00Z}, matching Track 1's {@code SessionBrowser.MAXDATE}) are clamped - * to that sentinel and behave the same as {@link #listSessions()}, returning sessions that - * have active messages.

- * - * @param sessionStateUpdatedAfter Only sessions whose session state was updated after this time are returned. + * @param sessionStateUpdatedAfter Only sessions whose session state was set or updated after this time are returned. * @return A {@link PagedIterable} of session ID strings. * @throws NullPointerException if {@code sessionStateUpdatedAfter} is null. */ diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java index 9331097ec5d7..5a8ab347a07f 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java @@ -521,17 +521,18 @@ public Mono getMessageSessions(OffsetDateTime lastUpdated return monoError(logger, new IllegalArgumentException("'top' must be positive; got " + top + ".")); } - // Track 1's SessionBrowser uses new Date(253402300800000L) as the active-messages sentinel + // Track 1's SessionBrowser uses new Date(253402300800000L) as the default-listing sentinel // (1ms past 9999-12-31T23:59:59.999Z, rendered by OffsetDateTime.toString() as // +10000-01-01T00:00Z). This is the wire value the broker has been validated against for - // years; align with it here. Any input at or beyond that instant (including + // listing sessions with active messages or stored session state; align with it here. Any + // input at or beyond that instant (including // OffsetDateTime.MAX, whose nanosecond precision and year-999_999_999 value would otherwise // overflow java.util.Date) is clamped to it so the sentinel comparison and Date.from(...) // both stay well-defined. Comparing with >= so the sentinel-equal case is also routed // through the clamp explicitly (it's a no-op for equal values, but keeps the comment/code // contract precise). - final OffsetDateTime cappedTime = lastUpdatedTime.compareTo(ManagementConstants.ACTIVE_MESSAGES_SENTINEL) >= 0 - ? ManagementConstants.ACTIVE_MESSAGES_SENTINEL + final OffsetDateTime cappedTime = lastUpdatedTime.compareTo(ManagementConstants.DEFAULT_LISTING_SENTINEL) >= 0 + ? ManagementConstants.DEFAULT_LISTING_SENTINEL : lastUpdatedTime; return isAuthorized(OPERATION_GET_MESSAGE_SESSIONS).then(channelCache.get().flatMap(channel -> { diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java index 728e9efb8a16..79c3c2264eaa 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementConstants.java @@ -36,8 +36,9 @@ public class ManagementConstants { public static final String SESSION_IDS = "sessions-ids"; /** - * Sentinel timestamp the broker recognizes as "list sessions with active messages" mode for the - * {@code OPERATION_GET_MESSAGE_SESSIONS} operation. Matches Track 1's + * Sentinel timestamp the broker recognizes as the default list mode for the + * {@code OPERATION_GET_MESSAGE_SESSIONS} operation. This mode returns sessions with active + * messages or stored session state and excludes sessions with neither. Matches Track 1's * {@code SessionBrowser.MAXDATE = new Date(253402300800000L)} * (rendered by {@link OffsetDateTime#toString()} as {@code +10000-01-01T00:00Z} - the leading * {@code +} is required by ISO 8601 for years with more than four digits); using any other @@ -45,7 +46,7 @@ public class ManagementConstants { * so callers and the implementation can clamp inputs via {@link OffsetDateTime#compareTo} * without each owning their own copy. */ - public static final OffsetDateTime ACTIVE_MESSAGES_SENTINEL + public static final OffsetDateTime DEFAULT_LISTING_SENTINEL = OffsetDateTime.of(10000, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); public static final String LAST_UPDATED_TIME = "last-updated-time"; public static final String LAST_SESSION_ID = "last-session-id"; diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java index e05facd030f0..3a29712bd749 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/implementation/ServiceBusManagementNode.java @@ -149,8 +149,7 @@ Mono updateDisposition(String lockToken, DispositionStatus dispositionStat Flux listRules(); /** - * Lists the session IDs for sessions that have active messages or whose state was updated - * since the given time. + * Lists session IDs using either the default listing mode or a session-state update cutoff. * *

Pagination follows the cursor semantics of Track 1's * {@code com.microsoft.azure.servicebus.SessionBrowser}: the caller threads {@code skip} from @@ -158,8 +157,9 @@ Mono updateDisposition(String lockToken, DispositionStatus dispositionStat * (the last entry of the previous page) into the next request, and stops when the broker returns * a page smaller than the requested page size (a short or empty page signals the end).

* - * @param lastUpdatedTime Filter timestamp. To get sessions with active messages, pass the - * {@link ManagementConstants#ACTIVE_MESSAGES_SENTINEL} sentinel (the implementation also + * @param lastUpdatedTime Filter timestamp. To use the default listing mode for sessions with + * active messages or stored session state, pass the + * {@link ManagementConstants#DEFAULT_LISTING_SENTINEL} sentinel (the implementation also * accepts {@link OffsetDateTime#MAX} and clamps it to that sentinel), which matches the * Track 1 Java sentinel value ({@code new Date(253402300800000L)}, rendered by * {@code OffsetDateTime.toString()} as {@code +10000-01-01T00:00Z}). Pass a real timestamp diff --git a/sdk/servicebus/azure-messaging-servicebus/src/samples/README.md b/sdk/servicebus/azure-messaging-servicebus/src/samples/README.md index 713a2d33b116..6c5f90909fbf 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/samples/README.md +++ b/sdk/servicebus/azure-messaging-servicebus/src/samples/README.md @@ -51,6 +51,8 @@ connection string value can be obtained by: - [Process all session messages using processor][ServiceBusSessionProcessorSample] - [Receive messages from a specific session][ReceiveNamedSessionAsyncSample] - [Receive messages from the first available session][ReceiveSingleSessionAsyncSample] +- [List session IDs synchronously][ListSessionsSample] +- [List session IDs asynchronously][ListSessionsAsyncSample] ### Synchronous Administration Client operations - [Update queue properties synchronously][AdministrationClientUpdateQueueSample] @@ -79,6 +81,8 @@ Guidelines](https://github.com/Azure/azure-sdk-for-java/blob/main/CONTRIBUTING.m [sdk_readme_next_steps]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/README.md#next-steps [PeekMessageAsyncSample]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/PeekMessageAsyncSample.java +[ListSessionsAsyncSample]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsAsyncSample.java +[ListSessionsSample]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsSample.java [ReceiveMessageAndSettleAsyncSample]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ReceiveMessageAndSettleAsyncSample.java [ReceiveMessageAsyncSample]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ReceiveMessageAsyncSample.java [ReceiveMessageAutoLockRenewal]: https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ReceiveMessageAutoLockRenewal.java diff --git a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsAsyncSample.java b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsAsyncSample.java new file mode 100644 index 000000000000..da716cf9dc75 --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsAsyncSample.java @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus; + +import org.junit.jupiter.api.Test; + +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +/** + * Demonstrates how to asynchronously list sessions using both supported modes. Default listing returns sessions with + * active messages or stored session state and excludes sessions with neither. A cutoff returns only sessions whose + * stored state was set or updated after that time. + */ +public class ListSessionsAsyncSample { + String connectionString = System.getenv("AZURE_SERVICEBUS_NAMESPACE_CONNECTION_STRING"); + String topicName = System.getenv("AZURE_SERVICEBUS_SAMPLE_TOPIC_NAME"); + String subscriptionName = System.getenv("AZURE_SERVICEBUS_SAMPLE_SESSION_SUBSCRIPTION_NAME"); + + /** + * Main method to invoke this demo on how to list session IDs in a Service Bus topic subscription. + * + * @param args Unused arguments to the program. + * @throws InterruptedException If the program is interrupted while waiting for the operation to complete. + */ + public static void main(String[] args) throws InterruptedException { + ListSessionsAsyncSample sample = new ListSessionsAsyncSample(); + sample.run(); + } + + /** + * Lists sessions using both supported modes. + * + * @throws InterruptedException If the program is interrupted while waiting for the operation to complete. + */ + @Test + public void run() throws InterruptedException { + CountDownLatch countdownLatch = new CountDownLatch(1); + ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder() + .connectionString(connectionString) + .sessionReceiver() + .topicName(topicName) + .subscriptionName(subscriptionName) + .buildAsyncClient(); + + try { + OffsetDateTime sessionStateUpdatedAfter = OffsetDateTime.now(ZoneOffset.UTC).minusDays(7); + + sessionReceiver.listSessions() + .doOnNext(sessionId -> System.out.println("Session ID: " + sessionId)) + .thenMany(sessionReceiver.listSessions(sessionStateUpdatedAfter)) + .subscribe( + sessionId -> System.out.println("Recently updated session ID: " + sessionId), + error -> { + System.err.println("Error occurred: " + error); + countdownLatch.countDown(); + }, + countdownLatch::countDown); + + countdownLatch.await(30, TimeUnit.SECONDS); + } finally { + sessionReceiver.close(); + } + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsSample.java b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsSample.java new file mode 100644 index 000000000000..525d98e65ee7 --- /dev/null +++ b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ListSessionsSample.java @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.messaging.servicebus; + +import org.junit.jupiter.api.Test; + +import java.time.OffsetDateTime; +import java.time.ZoneOffset; + +/** + * Demonstrates how to list sessions using both supported modes. Default listing returns sessions with active messages + * or stored session state and excludes sessions with neither. A cutoff returns only sessions whose stored state was set + * or updated after that time. + */ +public class ListSessionsSample { + String connectionString = System.getenv("AZURE_SERVICEBUS_NAMESPACE_CONNECTION_STRING"); + String queueName = System.getenv("AZURE_SERVICEBUS_SAMPLE_SESSION_QUEUE_NAME"); + + /** + * Main method to invoke this demo on how to list session IDs in a Service Bus queue. + * + * @param args Unused arguments to the program. + */ + public static void main(String[] args) { + ListSessionsSample sample = new ListSessionsSample(); + sample.run(); + } + + /** + * Lists sessions using both supported modes. + */ + @Test + public void run() { + ServiceBusSessionReceiverClient sessionReceiver = new ServiceBusClientBuilder() + .connectionString(connectionString) + .sessionReceiver() + .queueName(queueName) + .buildClient(); + + try { + listSessionsWithMessagesOrState(sessionReceiver); + listSessionsWithRecentlyUpdatedState(sessionReceiver); + } finally { + sessionReceiver.close(); + } + } + + private static void listSessionsWithMessagesOrState(ServiceBusSessionReceiverClient sessionReceiver) { + // Omitting the cutoff returns sessions with active messages or stored session state. + sessionReceiver.listSessions() + .forEach(sessionId -> System.out.println("Session ID: " + sessionId)); + } + + private static void listSessionsWithRecentlyUpdatedState(ServiceBusSessionReceiverClient sessionReceiver) { + // Supplying a cutoff returns only sessions whose stored state was set or updated after that time. + OffsetDateTime sessionStateUpdatedAfter = OffsetDateTime.now(ZoneOffset.UTC).minusDays(7); + sessionReceiver.listSessions(sessionStateUpdatedAfter) + .forEach(sessionId -> System.out.println("Recently updated session ID: " + sessionId)); + } +} diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClientTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClientTest.java index a535665e38ee..9815ecc9d93d 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClientTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusSessionReceiverAsyncClientTest.java @@ -401,19 +401,19 @@ private static List fullPage(String prefix) { } /** - * Verifies the no-arg listSessions() drives the broker with the active-messages sentinel and + * Verifies the no-arg listSessions() drives the broker with the default-listing sentinel and * collects every page until the broker returns a short page (fewer IDs than the requested page * size), which terminates pagination. */ @Test - void listSessionsActiveModeStreamsAllPagesUntilShortPage() { + void listSessionsDefaultListingModeStreamsAllPagesUntilShortPage() { // First page: a full page (100 sessions) continues; server-returned skip = 100. final List firstPage = fullPage("s"); - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(0), eq(100), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(0), eq(100), isNull())).thenReturn(Mono.just(new MessageSessionsResult(firstPage, 100))); // Cursor for the second page is server-skip (100) + base64url(lastSessionId "s99"). The second // page is short (2 < 100), which terminates pagination. - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(100), eq(100), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(100), eq(100), eq("s99"))).thenReturn(Mono.just(new MessageSessionsResult(Arrays.asList("t1", "t2"), 102))); final ServiceBusSessionReceiverAsyncClient client = newSessionReceiver(); @@ -479,11 +479,11 @@ void listSessionsRoundTripsArbitrarySessionIdsThroughCursor() { // page encodes it; a full page also drives the second request under short-page termination. final List firstPage = fullPage("x", 99); firstPage.add(sessionWithPipe); - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(0), eq(100), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(0), eq(100), isNull())).thenReturn(Mono.just(new MessageSessionsResult(firstPage, 100))); // The second-page request must decode the cursor back to lastSessionId=sessionWithPipe intact // (pipe and all); the short (empty) page then terminates pagination. - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(100), eq(100), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(100), eq(100), eq(sessionWithPipe))).thenReturn(Mono.just(new MessageSessionsResult(Collections.emptyList(), 100))); final ServiceBusSessionReceiverAsyncClient client = newSessionReceiver(); @@ -563,9 +563,9 @@ void listSessionsHonorsCallerPageSize() { // page is full (25 items) so a second page is requested; the short second page (1 < 25) // terminates pagination. final List firstPage = fullPage("s", 25); - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(0), eq(25), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(0), eq(25), isNull())).thenReturn(Mono.just(new MessageSessionsResult(firstPage, 25))); - when(managementNode.getMessageSessions(eq(ManagementConstants.ACTIVE_MESSAGES_SENTINEL), eq(25), eq(25), + when(managementNode.getMessageSessions(eq(ManagementConstants.DEFAULT_LISTING_SENTINEL), eq(25), eq(25), eq("s24"))).thenReturn(Mono.just(new MessageSessionsResult(Collections.singletonList("t1"), 26))); final ServiceBusSessionReceiverAsyncClient client = newSessionReceiver(); diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java index 6026ba0a218e..85d195a4bdf7 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/implementation/ManagementChannelTests.java @@ -1116,14 +1116,14 @@ void getMessageSessionsSessionStateUpdatedAfterMode() { } /** - * Verifies getMessageSessions in active-messages mode uses the Track 1 active-messages sentinel. + * Verifies getMessageSessions in default listing mode uses the Track 1 default-listing sentinel. * Track 1's {@code SessionBrowser.MAXDATE} is {@code new Date(253402300800000L)} * (10000-01-01T00:00:00Z UTC, 1 ms past 9999-12-31T23:59:59.999Z), which the broker recognizes - * as the "list sessions with active messages" mode. + * as the mode that lists sessions with active messages or stored session state. */ @Test - void getMessageSessionsActiveMessagesMode() { - // Arrange - Track 1 active-messages sentinel (10000-01-01T00:00:00Z UTC). + void getMessageSessionsDefaultListingMode() { + // Arrange - Track 1 default-listing sentinel (10000-01-01T00:00:00Z UTC). final OffsetDateTime sentinel = OffsetDateTime.of(10000, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC); final String[] sessionIds = new String[] { "active-1", "active-2" }; @@ -1271,7 +1271,7 @@ void getMessageSessionsRejectsUnexpectedSessionIdsPayloadType() { /** * Verifies that getMessageSessions clamps inputs at or beyond the Track 1 sentinel down to * the sentinel itself, both to avoid {@link java.util.Date} overflow for {@link OffsetDateTime#MAX} - * and to keep the broker's active-messages comparison stable. + * and to keep the broker's default-listing comparison stable. */ @Test void getMessageSessionsCapsYear() { @@ -1291,7 +1291,7 @@ void getMessageSessionsCapsYear() { .expectComplete() .verify(TIMEOUT); - // Verify the sent timestamp is capped to the Track 1 active-messages sentinel. + // Verify the sent timestamp is capped to the Track 1 default-listing sentinel. verify(requestResponseChannel).sendWithAck(messageCaptor.capture(), isNull()); @SuppressWarnings("unchecked") final Map body