Add ability to manage rules with the ServiceBusReceiver (Azure#30271)

* Create rule manager

Author: ZejiaJiang

sdk/servicebus/azure-messaging-servicebus/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 7.13.0-beta.1 (Unreleased)
 
 ### Features Added
+- Added rule manager client to manage rules for ServiceBus subscription with listen claims. ([#27711](https://github.com/Azure/azure-sdk-for-java/issues/27711))
 
 ### Breaking Changes
 

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusClientBuilder.java

@@ -649,6 +649,15 @@ public ServiceBusSessionProcessorClientBuilder sessionProcessor() {
         return new ServiceBusSessionProcessorClientBuilder();
     }
 
+    /**
+     * A new instance of {@link ServiceBusRuleManagerBuilder} used to configure a Service Bus rule manager instance.
+     *
+     * @return A new instance of {@link ServiceBusRuleManagerBuilder}.
+     */
+    public ServiceBusRuleManagerBuilder ruleManager() {
+        return new ServiceBusRuleManagerBuilder();
+    }
+
     /**
      * Called when a child client is closed. Disposes of the shared connection if there are no more clients.
      */
@@ -2021,6 +2030,77 @@ ServiceBusReceiverAsyncClient buildAsyncClient(boolean isAutoCompleteAllowed, bo
         }
     }
 
+    /**
+     * Builder for creating {@link ServiceBusRuleManagerAsyncClient}  to manage Service Bus subscription rules.
+     *
+     * @see ServiceBusRuleManagerAsyncClient
+     */
+    @ServiceClientBuilder(serviceClients = {ServiceBusRuleManagerAsyncClient.class})
+    public final class ServiceBusRuleManagerBuilder {
+        private String subscriptionName;
+        private String topicName;
+
+        private ServiceBusRuleManagerBuilder() {
+        }
+
+        /**
+         * Sets the name of the topic. <b>{@link #subscriptionName(String)} must also be set.</b>
+         *
+         * @param topicName Name of the topic.
+         *
+         * @return The modified {@link ServiceBusRuleManagerBuilder} object.
+         * @see #subscriptionName A subscription name should be set as well.
+         */
+        public ServiceBusRuleManagerBuilder topicName(String topicName) {
+            this.topicName = topicName;
+            return this;
+        }
+
+        /**
+         * Sets the name of the subscription in the topic to manage its rules. <b>{@link #topicName(String)} must also be set.
+         * </b>
+         * @param subscriptionName Name of the subscription.
+         *
+         * @return The modified {@link ServiceBusRuleManagerBuilder} object.
+         * @see #topicName A topic name should be set as well.
+         */
+        public ServiceBusRuleManagerBuilder subscriptionName(String subscriptionName) {
+            this.subscriptionName = subscriptionName;
+            return this;
+        }
+
+        /**
+         * Creates an <b>asynchronous</b> {@link ServiceBusRuleManagerAsyncClient} for managing rules of the specific subscription.
+         *
+         * @return A new {@link ServiceBusRuleManagerAsyncClient} that manages rules for specific subscription.
+         * @throws IllegalStateException if {@code topicName} or {@code subscriptionName} is null or empty. It is also
+         * thrown if the Service Bus {@link #connectionString(String) connectionString} contains an {@code EntityPath}
+         * that does not match one set in {@link #topicName(String) topicName}.
+         */
+        public ServiceBusRuleManagerAsyncClient buildAsyncClient() {
+            final MessagingEntityType entityType = validateEntityPaths(connectionStringEntityName, topicName,
+                null);
+            final String entityPath = getEntityPath(entityType, null, topicName, subscriptionName,
+                null);
+            final ServiceBusConnectionProcessor connectionProcessor = getOrCreateConnectionProcessor(messageSerializer);
+
+            return new ServiceBusRuleManagerAsyncClient(entityPath, entityType, connectionProcessor,
+                ServiceBusClientBuilder.this::onClientClose);
+        }
+
+        /**
+         * Creates a <b>synchronous</b> {@link ServiceBusRuleManagerClient} for managing rules of the specific subscription.
+         *
+         * @return A new {@link ServiceBusRuleManagerClient} that manages rules for specific subscription.
+         * @throws IllegalStateException if {@code topicName} or {@code subscriptionName} is null or empty. It is also
+         * thrown if the Service Bus {@link #connectionString(String) connectionString} contains an {@code EntityPath}
+         * that does not match one set in {@link #topicName(String) topicName}.
+         */
+        public ServiceBusRuleManagerClient buildClient() {
+            return new ServiceBusRuleManagerClient(buildAsyncClient(), MessageUtils.getTotalTimeout(retryOptions));
+        }
+    }
+
     private void validateAndThrow(int prefetchCount) {
         if (prefetchCount < 0) {
             throw LOGGER.logExceptionAsError(new IllegalArgumentException(String.format(

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusRuleManagerAsyncClient.java

@@ -0,0 +1,243 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.messaging.servicebus;
+
+import com.azure.core.annotation.ServiceClient;
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.messaging.servicebus.administration.ServiceBusAdministrationAsyncClient;
+import com.azure.messaging.servicebus.administration.models.CorrelationRuleFilter;
+import com.azure.messaging.servicebus.administration.models.CreateRuleOptions;
+import com.azure.messaging.servicebus.administration.models.RuleFilter;
+import com.azure.messaging.servicebus.administration.models.RuleProperties;
+import com.azure.messaging.servicebus.administration.models.SqlRuleAction;
+import com.azure.messaging.servicebus.administration.models.SqlRuleFilter;
+import com.azure.messaging.servicebus.implementation.MessagingEntityType;
+import com.azure.messaging.servicebus.implementation.ServiceBusConnectionProcessor;
+import com.azure.messaging.servicebus.implementation.ServiceBusManagementNode;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import static com.azure.core.util.FluxUtil.fluxError;
+import static com.azure.core.util.FluxUtil.monoError;
+import static com.azure.messaging.servicebus.implementation.Messages.INVALID_OPERATION_DISPOSED_RULE_MANAGER;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * An <b>asynchronous</b> rule manager responsible for managing rules for a specific topic subscription. The rule manager
+ * requires only Listen claims, whereas the {@link ServiceBusAdministrationAsyncClient} requires Manage claims.
+ *
+ * <p><strong>Create an instance of rule manager</strong></p>
+ * <!-- src_embed com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.instantiation -->
+ * <pre>
+ * &#47;&#47; The required parameters is connectionString, a way to authenticate with Service Bus using credentials.
+ * &#47;&#47; The connectionString&#47;queueName must be set by the application. The 'connectionString' format is shown below.
+ * &#47;&#47; &quot;Endpoint=&#123;fully-qualified-namespace&#125;;SharedAccessKeyName=&#123;policy-name&#125;;SharedAccessKey=&#123;key&#125;&quot;
+ *
+ * ServiceBusRuleManagerAsyncClient ruleManager = new ServiceBusClientBuilder&#40;&#41;
+ *     .connectionString&#40;connectionString&#41;
+ *     .ruleManager&#40;&#41;
+ *     .topicName&#40;topicName&#41;
+ *     .subscriptionName&#40;subscriptionName&#41;
+ *     .buildAsyncClient&#40;&#41;;
+ * </pre>
+ * <!-- end com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.instantiation -->
+ *
+ * <p><strong>Create a rule to a Service Bus subscription</strong></p>
+ * <!-- src_embed com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.createRule -->
+ * <pre>
+ * RuleFilter trueRuleFilter = new TrueRuleFilter&#40;&#41;;
+ * CreateRuleOptions options = new CreateRuleOptions&#40;trueRuleFilter&#41;;
+ * ruleManager.createRule&#40;&quot;new-rule&quot;, options&#41;.subscribe&#40;
+ *     unused -&gt; &#123; &#125;,
+ *     err -&gt; System.err.println&#40;&quot;Error occurred when create a rule, err: &quot; + err&#41;,
+ *     &#40;&#41; -&gt; System.out.println&#40;&quot;Create complete.&quot;&#41;
+ * &#41;;
+ * </pre>
+ * <!-- end com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.createRule -->
+ *
+ * <p><strong>Fetch all rules.</strong></p>
+ * <!-- src_embed com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.getRules -->
+ * <pre>
+ * ruleManager.getRules&#40;&#41;.subscribe&#40;ruleProperties -&gt; System.out.println&#40;ruleProperties.getName&#40;&#41;&#41;&#41;;
+ * </pre>
+ * <!-- end com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.getRules -->
+ *
+ * <p><strong>Delete a rule.</strong></p>
+ * <!-- src_embed com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.deleteRule -->
+ * <pre>
+ * ruleManager.deleteRule&#40;&quot;exist-rule&quot;&#41;.subscribe&#40;
+ *     unused -&gt; &#123; &#125;,
+ *     err -&gt; System.err.println&#40;&quot;Error occurred when delete rule, err: &quot; + err&#41;,
+ *     &#40;&#41; -&gt; System.out.println&#40;&quot;Delete complete.&quot;&#41;
+ * &#41;;
+ * </pre>
+ * <!-- end com.azure.messaging.servicebus.servicebusrulemanagerasyncclient.deleteRule -->
+ * @see ServiceBusClientBuilder
+ */
+@ServiceClient(builder = ServiceBusClientBuilder.class, isAsync = true)
+public class ServiceBusRuleManagerAsyncClient implements AutoCloseable {
+    private static final ClientLogger LOGGER = new ClientLogger(ServiceBusRuleManagerAsyncClient.class);
+
+    private final String entityPath;
+    private final MessagingEntityType entityType;
+    private final ServiceBusConnectionProcessor connectionProcessor;
+    private final Runnable onClientClose;
+    private final AtomicBoolean isDisposed = new AtomicBoolean();
+
+    /**
+     * Creates a rule manager that manages rules for a Service Bus subscription.
+     *
+     * @param entityPath The name of the topic and subscription.
+     * @param entityType The type of the Service Bus resource.
+     * @param connectionProcessor The AMQP connection to the Service Bus resource.
+     * @param onClientClose Operation to run when the client completes.
+     */
+    ServiceBusRuleManagerAsyncClient(String entityPath, MessagingEntityType entityType,
+        ServiceBusConnectionProcessor connectionProcessor, Runnable onClientClose) {
+        this.entityPath = Objects.requireNonNull(entityPath, "'entityPath' cannot be null.");
+        this.entityType = Objects.requireNonNull(entityType, "'entityType' cannot be null.");
+        this.connectionProcessor = Objects.requireNonNull(connectionProcessor,
+            "'connectionProcessor' cannot be null.");
+        this.onClientClose = onClientClose;
+    }
+
+    /**
+     * Gets the fully qualified namespace.
+     *
+     * @return The fully qualified namespace.
+     */
+    public String getFullyQualifiedNamespace() {
+        return connectionProcessor.getFullyQualifiedNamespace();
+    }
+
+    /**
+     * Gets the name of the Service Bus resource.
+     *
+     * @return The name of the Service Bus resource.
+     */
+    public String getEntityPath() {
+        return entityPath;
+    }
+
+    /**
+     * Creates a rule to the current subscription to filter the messages reaching from topic to the subscription.
+     *
+     * @param ruleName Name of rule.
+     * @param options The options for the rule to add.
+     * @return A Mono that completes when the rule is created.
+     *
+     * @throws NullPointerException if {@code options}, {@code ruleName} is null.
+     * @throws IllegalStateException if client is disposed.
+     * @throws IllegalArgumentException if {@code ruleName} is empty string, action of {@code options} is not null and not
+     * instanceof {@link SqlRuleAction}, filter of {@code options} is not instanceof {@link SqlRuleFilter} or
+     * {@link CorrelationRuleFilter}.
+     * @throws ServiceBusException if filter matches {@code ruleName} is already created in subscription.
+     */
+    public Mono<Void> createRule(String ruleName, CreateRuleOptions options) {
+        if (Objects.isNull(options)) {
+            return monoError(LOGGER, new NullPointerException("'options' cannot be null."));
+        }
+        return createRuleInternal(ruleName, options);
+    }
+
+    /**
+     * Creates a rule to the current subscription to filter the messages reaching from topic to the subscription.
+     *
+     * @param ruleName Name of rule.
+     * @param filter The filter expression against which messages will be matched.
+     * @return A Mono that completes when the rule is created.
+     *
+     * @throws NullPointerException if {@code filter}, {@code ruleName} is null.
+     * @throws IllegalStateException if client is disposed.
+     * @throws IllegalArgumentException if ruleName is empty string, {@code filter} is not instanceof {@link SqlRuleFilter} or
+     * {@link CorrelationRuleFilter}.
+     * @throws ServiceBusException if filter matches {@code ruleName} is already created in subscription.
+     */
+    public Mono<Void> createRule(String ruleName, RuleFilter filter) {
+        CreateRuleOptions options = new CreateRuleOptions(filter);
+        return createRuleInternal(ruleName, options);
+    }
+
+    /**
+     * Fetches all rules associated with the topic and subscription.
+     *
+     * @return A list of rules associated with the topic and subscription.
+     *
+     * @throws IllegalStateException if client is disposed.
+     * @throws UnsupportedOperationException if client cannot support filter with descriptor in message body.
+     */
+    public Flux<RuleProperties> getRules() {
+        if (isDisposed.get()) {
+            return fluxError(LOGGER, new IllegalStateException(
+                String.format(INVALID_OPERATION_DISPOSED_RULE_MANAGER, "getRules")
+            ));
+        }
+
+        return connectionProcessor
+            .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
+            .flatMapMany(ServiceBusManagementNode::getRules);
+    }
+
+    /**
+     * Removes the rule on the subscription identified by {@code ruleName}.
+     *
+     * @param ruleName Name of rule to delete.
+     * @return A Mono that completes when the rule is deleted.
+     *
+     * @throws NullPointerException if {@code ruleName} is null.
+     * @throws IllegalStateException if client is disposed.
+     * @throws IllegalArgumentException if {@code ruleName} is empty string.
+     * @throws ServiceBusException if cannot find filter matches {@code ruleName} in subscription.
+     */
+    public Mono<Void> deleteRule(String ruleName) {
+        if (isDisposed.get()) {
+            return monoError(LOGGER, new IllegalStateException(
+                String.format(INVALID_OPERATION_DISPOSED_RULE_MANAGER, "deleteRule")
+            ));
+        }
+
+        if (ruleName == null) {
+            return monoError(LOGGER, new NullPointerException("'ruleName' cannot be null."));
+        } else if (ruleName.isEmpty()) {
+            return monoError(LOGGER, new IllegalArgumentException("'ruleName' cannot be an empty string."));
+        }
+
+        return connectionProcessor
+            .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
+            .flatMap(managementNode -> managementNode.deleteRule(ruleName));
+    }
+
+    /**
+     * Disposes of the {@link ServiceBusRuleManagerAsyncClient}. If the client has a dedicated connection, the underlying
+     * connection is also closed.
+     */
+    @Override
+    public void close() {
+        if (isDisposed.getAndSet(true)) {
+            return;
+        }
+
+        onClientClose.run();
+    }
+
+    private Mono<Void> createRuleInternal(String ruleName, CreateRuleOptions options) {
+        if (isDisposed.get()) {
+            return monoError(LOGGER, new IllegalStateException(
+                String.format(INVALID_OPERATION_DISPOSED_RULE_MANAGER, "createRule")
+            ));
+        }
+
+        if (ruleName == null) {
+            return monoError(LOGGER, new NullPointerException("'ruleName' cannot be null."));
+        } else if (ruleName.isEmpty()) {
+            return monoError(LOGGER, new IllegalArgumentException("'ruleName' cannot be an empty string."));
+        }
+
+        return connectionProcessor
+            .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
+            .flatMap(managementNode -> managementNode.createRule(ruleName, options));
+    }
+}