Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

### Features Added

- Added `listSessions()` and `listSessions(OffsetDateTime sessionStateUpdatedAfter)` to `ServiceBusSessionReceiverAsyncClient` (returning `PagedFlux<String>`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable<String>`). 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<String>`) and `ServiceBusSessionReceiverClient` (returning `PagedIterable<String>`). 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.

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -313,11 +313,11 @@ private Mono<ServiceBusReceiverAsyncClient> 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.
*
* <p>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.</p>
* <p>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
Comment thread
EldertGrootenboer marked this conversation as resolved.
* returned.</p>
*
* <p>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
Expand All @@ -330,26 +330,19 @@ private Mono<ServiceBusReceiverAsyncClient> acquireSpecificOrNextSession(String
@ServiceMethod(returns = ReturnType.COLLECTION)
public PagedFlux<String> 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.
*
* <p>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)}.</p>
*
* <p>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.</p>
*
* @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.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
* <p>Sessions with neither active messages nor stored session state are excluded.</p>
*
* <p>The returned {@link PagedIterable} fetches additional pages from the broker on demand;
* iterate the {@code PagedIterable} (or call {@link PagedIterable#stream()}) to receive every
Expand All @@ -228,21 +230,14 @@ public PagedIterable<String> 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.
*
* <p>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}).</p>
*
* <p>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.</p>
*
* @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.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -521,17 +521,18 @@ public Mono<MessageSessionsResult> 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 -> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,16 +36,17 @@ 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)}
Comment thread
EldertGrootenboer marked this conversation as resolved.
* (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
* value risks the broker not switching into the proven mode. Defined as {@link OffsetDateTime}
* 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";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -149,17 +149,17 @@ Mono<Void> updateDisposition(String lockToken, DispositionStatus dispositionStat
Flux<RuleProperties> 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.
*
* <p>Pagination follows the cursor semantics of Track 1's
* {@code com.microsoft.azure.servicebus.SessionBrowser}: the caller threads {@code skip} from
Comment thread
EldertGrootenboer marked this conversation as resolved.
* {@link MessageSessionsResult#getNextSkip()} of the previous response and {@code lastSessionId}
* (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).</p>
*
* @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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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();
}
}
}
Original file line number Diff line number Diff line change
@@ -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));
}
}
Loading
Loading