Azure · EldertGrootenboer · Apr 28, 2026 · Apr 28, 2026 · Apr 28, 2026 · Apr 28, 2026
@@ -16,10 +16,14 @@
 import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusReceiverInstrumentation;
 import com.azure.messaging.servicebus.implementation.instrumentation.ServiceBusTracer;
 import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
 import java.util.Objects;
 
+import static com.azure.core.util.FluxUtil.fluxError;
 import static com.azure.core.util.FluxUtil.monoError;
 import static com.azure.messaging.servicebus.ReceiverOptions.createNamedSessionOptions;
 
@@ -292,6 +296,55 @@ private Mono<ServiceBusReceiverAsyncClient> acquireSpecificOrNextSession(String
         return tracer.traceMono("ServiceBus.acceptSession", acquireSessionReceiver);
     }
 
+    /**
+     * Lists the IDs of sessions that have active messages 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>
+     *
+     * @return A {@link Flux} of session ID strings.
+     */
+    @ServiceMethod(returns = ReturnType.COLLECTION)
+    public Flux<String> listSessions() {
+        // Use year 9999 as sentinel for "active messages" mode.
+        // Cannot use OffsetDateTime.MAX because its year (999999999) overflows java.util.Date.
+        return listSessionsInternal(OffsetDateTime.of(9999, 12, 31, 23, 59, 59, 999999999, ZoneOffset.UTC));
+    }
+
+    /**
+     * Lists the IDs of sessions whose state was updated after the specified time.
+     *
+     * @param updatedAfter Only sessions whose session state was updated after this time are returned.
+     * @return A {@link Flux} of session ID strings.
+     * @throws NullPointerException if {@code updatedAfter} is null.
+     */
+    @ServiceMethod(returns = ReturnType.COLLECTION)
+    public Flux<String> listSessions(OffsetDateTime updatedAfter) {
+        if (updatedAfter == null) {
+            return fluxError(LOGGER, new NullPointerException("'updatedAfter' cannot be null."));
+        }
+        return listSessionsInternal(updatedAfter);
+    }
+
+    private Flux<String> listSessionsInternal(OffsetDateTime lastUpdatedTime) {
+        final int pageSize = 100;
+        return connectionCacheWrapper.getConnection()
+            .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
+            .flatMapMany(managementNode -> {
+                final int[] currentSkip = { 0 };
+                return Flux
+                    .defer(() -> managementNode.getMessageSessions(lastUpdatedTime, currentSkip[0], pageSize, null))
+                    .repeat()
+                    .takeWhile(page -> !page.isEmpty())
+                    .map(page -> {
+                        currentSkip[0] += page.size();
+                        return page;
+                    })
+                    .flatMapIterable(page -> page);
+            });
+    }
+
     @Override
     public void close() {
         if (this.unNamedSessionManager != null) {

@@ -12,6 +12,8 @@
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -201,6 +203,28 @@ public ServiceBusReceiverClient acceptSession(String sessionId) {
             .block();
     }
 
+    /**
+     * Lists the IDs of sessions that have active messages in this entity.
+     *
+     * @return A list of session ID strings.
+     */
+    @ServiceMethod(returns = ReturnType.COLLECTION)
+    public List<String> listSessions() {
+        return sessionAsyncClient.listSessions().collectList().block(operationTimeout);
+    }
+
+    /**
+     * Lists the IDs of sessions whose state was updated after the specified time.
+     *
+     * @param updatedAfter Only sessions whose session state was updated after this time are returned.
+     * @return A list of session ID strings.
+     * @throws NullPointerException if {@code updatedAfter} is null.
+     */
+    @ServiceMethod(returns = ReturnType.COLLECTION)
+    public List<String> listSessions(OffsetDateTime updatedAfter) {
+        return sessionAsyncClient.listSessions(updatedAfter).collectList().block(operationTimeout);
+    }
+
     @Override
     public void close() {
         sessionAsyncClient.close();

@@ -57,6 +57,7 @@
 import static com.azure.core.util.FluxUtil.fluxError;
 import static com.azure.core.util.FluxUtil.monoError;
 import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_ADD_RULE;
+import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_MESSAGE_SESSIONS;
 import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_RULES;
 import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_GET_SESSION_STATE;
 import static com.azure.messaging.servicebus.implementation.ManagementConstants.OPERATION_PEEK;
@@ -504,6 +505,77 @@ public Mono<Void> deleteRule(String ruleName) {
         })).then();
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public Mono<List<String>> getMessageSessions(OffsetDateTime lastUpdatedTime, int skip, int top,
+        String lastSessionId) {
+        if (lastUpdatedTime == null) {
+            return monoError(logger, new NullPointerException("'lastUpdatedTime' cannot be null."));
+        }
+
+        // The service checks `lastUpdatedTime == DateTime.MaxValue` (C# 9999-12-31T23:59:59.9999999)
+        // to switch between "active messages" mode and "updated since" mode. On the AMQP wire,
+        // timestamps have millisecond precision, so DateTime.MaxValue becomes 253402300799999 ms.
+        // Year 9999 with .999999999 nanos in UTC produces this same ms value via Date.from(instant),
+        // ensuring the service-side comparison matches. Cap to the max supported instant in UTC to
+        // avoid java.util.Date overflow (OffsetDateTime.MAX year 999999999 exceeds Date's range)
+        // while keeping the serialized sentinel stable even when the source offset is non-UTC.
+        final OffsetDateTime cappedTime = lastUpdatedTime.getYear() > 9999
+            ? OffsetDateTime.of(9999, 12, 31, 23, 59, 59, 999999999, ZoneOffset.UTC)
+            : lastUpdatedTime;
+
+        return isAuthorized(OPERATION_GET_MESSAGE_SESSIONS).then(channelCache.get().flatMap(channel -> {
+            // No associated link name for entity-level operations
+            final Message message = createManagementMessage(OPERATION_GET_MESSAGE_SESSIONS, null);
+
+            final Map<String, Object> body = new HashMap<>();
+            body.put(ManagementConstants.LAST_UPDATED_TIME, Date.from(cappedTime.toInstant()));
+            body.put(ManagementConstants.SKIP, skip);
+            body.put(ManagementConstants.TOP, top);
+            if (!CoreUtils.isNullOrEmpty(lastSessionId)) {
+                body.put(ManagementConstants.LAST_SESSION_ID, lastSessionId);
+            }
+
+            message.setBody(new AmqpValue(body));
+
+            return sendWithVerify(channel, message, null);
+        })).flatMap(response -> {
+            final AmqpResponseCode statusCode = RequestResponseUtils.getStatusCode(response);
+
+            if (statusCode == AmqpResponseCode.OK) {
+                final Object value = ((AmqpValue) response.getBody()).getValue();
+                if (value instanceof Map) {
+                    @SuppressWarnings("unchecked")
+                    final Map<String, Object> map = (Map<String, Object>) value;
+                    final Object sessionsObj = map.get(ManagementConstants.SESSIONS_IDS);
+
+                    if (sessionsObj instanceof Object[]) {
+                        final Object[] sessionArray = (Object[]) sessionsObj;
+                        final List<String> sessionIds = new ArrayList<>(sessionArray.length);
+                        for (Object id : sessionArray) {
+                            sessionIds.add(String.valueOf(id));
+                        }
+                        return Mono.just(sessionIds);
+                    }
+                }
+                return Mono.just(Collections.emptyList());
+            } else if (statusCode == AmqpResponseCode.NO_CONTENT) {
+                return Mono.just(Collections.emptyList());
+            } else if (statusCode == AmqpResponseCode.NOT_FOUND) {
+                // 404 + SessionNotFound means no sessions exist. sendWithVerify already passes
+                // this through as a successful response rather than an error.
+                return Mono.just(Collections.emptyList());
+            } else {
+                final String statusDescription = RequestResponseUtils.getStatusDescription(response);
+                throw logger.logExceptionAsWarning(new AmqpException(true,
+                    "Get message sessions failed. Status: " + statusCode + " Description: " + statusDescription,
+                    getErrorContext()));
+            }
+        });
+    }
+
     /**
      * {@inheritDoc}
      */

@@ -25,6 +25,9 @@ public class ManagementConstants {
     public static final String SEQUENCE_NUMBERS = "sequence-numbers";
     public static final String SESSION_ID = "session-id";
     public static final String SESSION_STATE = "session-state";
+    public static final String SESSIONS_IDS = "sessions-ids";
+    public static final String LAST_UPDATED_TIME = "last-updated-time";
+    public static final String LAST_SESSION_ID = "last-session-id";
     public static final String VIA_PARTITION_KEY = "via-partition-key";
     public static final String RULE_NAME = "rule-name";
     public static final String RULE_DESCRIPTION = "rule-description";
@@ -61,6 +64,7 @@ public class ManagementConstants {
     static final String OPERATION_ADD_RULE = AmqpConstants.VENDOR + ":add-rule";
     static final String OPERATION_REMOVE_RULE = AmqpConstants.VENDOR + ":remove-rule";
     static final String OPERATION_GET_RULES = AmqpConstants.VENDOR + ":enumerate-rules";
+    static final String OPERATION_GET_MESSAGE_SESSIONS = AmqpConstants.VENDOR + ":get-message-sessions";
 
     static final String SERVER_TIMEOUT = AmqpConstants.VENDOR + ":server-timeout";
 

@@ -148,6 +148,22 @@ 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.
+     *
+     * @param lastUpdatedTime Filter timestamp. To get sessions with active messages, pass the UTC sentinel
+     *     {@code 9999-12-31T23:59:59.999999999Z}; pass a real timestamp to get sessions updated since that
+     *     time. The implementation caps {@code lastUpdatedTime} values beyond year 9999 (such as
+     *     {@link java.time.OffsetDateTime#MAX}) to that UTC sentinel so the service-side sentinel
+     *     comparison matches.
+     * @param skip Pagination offset.
+     * @param top Page size.
+     * @param lastSessionId Last session ID from the previous page (for cursor-based pagination), or null.
+     * @return A list of session ID strings for this page.
+     */
+    Mono<List<String>> getMessageSessions(OffsetDateTime lastUpdatedTime, int skip, int top, String lastSessionId);
+
     @Override
     void close();
 }