[feat] Support pattern subscribe non-persistent topic. (#207)

* [feat] Support pattern subscribe non-persistent topic.

* Fix code reviews.

Author: shibd

include/pulsar/ConsumerConfiguration.h

@@ -27,6 +27,7 @@
 #include <pulsar/InitialPosition.h>
 #include <pulsar/KeySharedPolicy.h>
 #include <pulsar/Message.h>
+#include <pulsar/RegexSubscriptionMode.h>
 #include <pulsar/Result.h>
 #include <pulsar/Schema.h>
 #include <pulsar/TypedMessage.h>
@@ -383,6 +384,19 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     int getPatternAutoDiscoveryPeriod() const;
 
+    /**
+     * Determines which topics this consumer should be subscribed to - Persistent, Non-Persistent, or
+     * AllTopics. Only used with pattern subscriptions.
+     *
+     * @param regexSubscriptionMode The default value is `PersistentOnly`.
+     */
+    ConsumerConfiguration& setRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode);
+
+    /**
+     * @return the regex subscription mode for the pattern consumer.
+     */
+    RegexSubscriptionMode getRegexSubscriptionMode() const;
+
     /**
      * The default value is `InitialPositionLatest`.
      *

include/pulsar/RegexSubscriptionMode.h

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#ifndef PULSAR_CPP_REGEX_SUB_MODE_H
+#define PULSAR_CPP_REGEX_SUB_MODE_H
+
+namespace pulsar {
+enum RegexSubscriptionMode
+{
+    /**
+     * Only subscribe to persistent topics.
+     */
+    PersistentOnly = 0,
+
+    /**
+     * Only subscribe to non-persistent topics.
+     */
+    NonPersistentOnly = 1,
+
+    /**
+     * Subscribe to both persistent and non-persistent topics.
+     */
+    AllTopics = 2
+};
+}
+
+#endif  // PULSAR_CPP_REGEX_SUB_MODE_H

lib/BinaryProtoLookupService.cc

@@ -151,7 +151,7 @@ uint64_t BinaryProtoLookupService::newRequestId() {
 }
 
 Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespaceAsync(
-    const NamespaceNamePtr& nsName) {
+    const NamespaceNamePtr& nsName, CommandGetTopicsOfNamespace_Mode mode) {
     NamespaceTopicsPromisePtr promise = std::make_shared<Promise<Result, NamespaceTopicsPtr>>();
     if (!nsName) {
         promise->setFailed(ResultInvalidTopicName);
@@ -160,7 +160,7 @@ Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespac
     std::string namespaceName = nsName->toString();
     cnxPool_.getConnectionAsync(serviceNameResolver_.resolveHost())
         .addListener(std::bind(&BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest, this,
-                               namespaceName, std::placeholders::_1, std::placeholders::_2, promise));
+                               namespaceName, mode, std::placeholders::_1, std::placeholders::_2, promise));
     return promise->getFuture();
 }
 
@@ -201,7 +201,9 @@ void BinaryProtoLookupService::sendGetSchemaRequest(const std::string& topicName
         });
 }
 
-void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string& nsName, Result result,
+void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string& nsName,
+                                                               CommandGetTopicsOfNamespace_Mode mode,
+                                                               Result result,
                                                                const ClientConnectionWeakPtr& clientCnx,
                                                                NamespaceTopicsPromisePtr promise) {
     if (result != ResultOk) {
@@ -212,8 +214,7 @@ void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string
     ClientConnectionPtr conn = clientCnx.lock();
     uint64_t requestId = newRequestId();
     LOG_DEBUG("sendGetTopicsOfNamespaceRequest. requestId: " << requestId << " nsName: " << nsName);
-
-    conn->newGetTopicsOfNamespace(nsName, requestId)
+    conn->newGetTopicsOfNamespace(nsName, mode, requestId)
         .addListener(std::bind(&BinaryProtoLookupService::getTopicsOfNamespaceListener, this,
                                std::placeholders::_1, std::placeholders::_2, promise));
 }

lib/BinaryProtoLookupService.h

@@ -49,7 +49,8 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
 
     Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) override;
 
-    Future<Result, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(const NamespaceNamePtr& nsName) override;
+    Future<Result, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(
+        const NamespaceNamePtr& nsName, CommandGetTopicsOfNamespace_Mode mode) override;
 
     Future<Result, boost::optional<SchemaInfo>> getSchema(const TopicNamePtr& topicName) override;
 
@@ -75,8 +76,8 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
                                        const ClientConnectionWeakPtr& clientCnx,
                                        LookupDataResultPromisePtr promise);
 
-    void sendGetTopicsOfNamespaceRequest(const std::string& nsName, Result result,
-                                         const ClientConnectionWeakPtr& clientCnx,
+    void sendGetTopicsOfNamespaceRequest(const std::string& nsName, CommandGetTopicsOfNamespace_Mode mode,
+                                         Result result, const ClientConnectionWeakPtr& clientCnx,
                                          NamespaceTopicsPromisePtr promise);
 
     void sendGetSchemaRequest(const std::string& topiName, Result result,

lib/ClientConnection.cc

@@ -1293,8 +1293,8 @@ Future<Result, GetLastMessageIdResponse> ClientConnection::newGetLastMessageId(u
     return promise.getFuture();
 }
 
-Future<Result, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(const std::string& nsName,
-                                                                             uint64_t requestId) {
+Future<Result, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(
+    const std::string& nsName, CommandGetTopicsOfNamespace_Mode mode, uint64_t requestId) {
     Lock lock(mutex_);
     Promise<Result, NamespaceTopicsPtr> promise;
     if (isClosed()) {
@@ -1306,7 +1306,7 @@ Future<Result, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(con
 
     pendingGetNamespaceTopicsRequests_.insert(std::make_pair(requestId, promise));
     lock.unlock();
-    sendCommand(Commands::newGetTopicsOfNamespace(nsName, requestId));
+    sendCommand(Commands::newGetTopicsOfNamespace(nsName, mode, requestId));
     return promise.getFuture();
 }
 

lib/ClientConnection.h

@@ -181,7 +181,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
 
     Future<Result, GetLastMessageIdResponse> newGetLastMessageId(uint64_t consumerId, uint64_t requestId);
 
-    Future<Result, NamespaceTopicsPtr> newGetTopicsOfNamespace(const std::string& nsName, uint64_t requestId);
+    Future<Result, NamespaceTopicsPtr> newGetTopicsOfNamespace(const std::string& nsName,
+                                                               CommandGetTopicsOfNamespace_Mode mode,
+                                                               uint64_t requestId);
 
     Future<Result, boost::optional<SchemaInfo>> newGetSchema(const std::string& topicName,
                                                              uint64_t requestId);

lib/ClientImpl.cc

@@ -338,29 +338,53 @@ void ClientImpl::subscribeWithRegexAsync(const std::string& regexPattern, const
         }
     }
 
-    NamespaceNamePtr nsName = topicNamePtr->getNamespaceName();
+    if (TopicName::containsDomain(regexPattern)) {
+        LOG_WARN("Ignore invalid domain: "
+                 << topicNamePtr->getDomain()
+                 << ", use the RegexSubscriptionMode parameter to set the topic type");
+    }
+
+    CommandGetTopicsOfNamespace_Mode mode;
+    auto regexSubscriptionMode = conf.getRegexSubscriptionMode();
+    switch (regexSubscriptionMode) {
+        case PersistentOnly:
+            mode = CommandGetTopicsOfNamespace_Mode_PERSISTENT;
+            break;
+        case NonPersistentOnly:
+            mode = CommandGetTopicsOfNamespace_Mode_NON_PERSISTENT;
+            break;
+        case AllTopics:
+            mode = CommandGetTopicsOfNamespace_Mode_ALL;
+            break;
+        default:
+            LOG_ERROR("RegexSubscriptionMode not valid: " << regexSubscriptionMode);
+            callback(ResultInvalidConfiguration, Consumer());
+            return;
+    }
 
-    lookupServicePtr_->getTopicsOfNamespaceAsync(nsName).addListener(
-        std::bind(&ClientImpl::createPatternMultiTopicsConsumer, shared_from_this(), std::placeholders::_1,
-                  std::placeholders::_2, regexPattern, subscriptionName, conf, callback));
+    lookupServicePtr_->getTopicsOfNamespaceAsync(topicNamePtr->getNamespaceName(), mode)
+        .addListener(std::bind(&ClientImpl::createPatternMultiTopicsConsumer, shared_from_this(),
+                               std::placeholders::_1, std::placeholders::_2, regexPattern, mode,
+                               subscriptionName, conf, callback));
 }
 
 void ClientImpl::createPatternMultiTopicsConsumer(const Result result, const NamespaceTopicsPtr topics,
                                                   const std::string& regexPattern,
+                                                  CommandGetTopicsOfNamespace_Mode mode,
                                                   const std::string& subscriptionName,
                                                   const ConsumerConfiguration& conf,
                                                   SubscribeCallback callback) {
     if (result == ResultOk) {
         ConsumerImplBasePtr consumer;
 
-        PULSAR_REGEX_NAMESPACE::regex pattern(regexPattern);
+        PULSAR_REGEX_NAMESPACE::regex pattern(TopicName::removeDomain(regexPattern));
 
         NamespaceTopicsPtr matchTopics =
             PatternMultiTopicsConsumerImpl::topicsPatternFilter(*topics, pattern);
 
         auto interceptors = std::make_shared<ConsumerInterceptors>(conf.getInterceptors());
 
-        consumer = std::make_shared<PatternMultiTopicsConsumerImpl>(shared_from_this(), regexPattern,
+        consumer = std::make_shared<PatternMultiTopicsConsumerImpl>(shared_from_this(), regexPattern, mode,
                                                                     *matchTopics, subscriptionName, conf,
                                                                     lookupServicePtr_, interceptors);
 

lib/ClientImpl.h

@@ -28,6 +28,7 @@
 #include "Future.h"
 #include "LookupDataResult.h"
 #include "MemoryLimitController.h"
+#include "ProtoApiEnums.h"
 #include "ServiceNameResolver.h"
 #include "SynchronizedHashMap.h"
 
@@ -151,8 +152,10 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
     void handleClose(Result result, SharedInt remaining, ResultCallback callback);
 
     void createPatternMultiTopicsConsumer(const Result result, const NamespaceTopicsPtr topics,
-                                          const std::string& regexPattern, const std::string& consumerName,
-                                          const ConsumerConfiguration& conf, SubscribeCallback callback);
+                                          const std::string& regexPattern,
+                                          CommandGetTopicsOfNamespace_Mode mode,
+                                          const std::string& consumerName, const ConsumerConfiguration& conf,
+                                          SubscribeCallback callback);
 
     enum State
     {

lib/Commands.cc

@@ -609,12 +609,14 @@ SharedBuffer Commands::newGetLastMessageId(uint64_t consumerId, uint64_t request
     return buffer;
 }
 
-SharedBuffer Commands::newGetTopicsOfNamespace(const std::string& nsName, uint64_t requestId) {
+SharedBuffer Commands::newGetTopicsOfNamespace(const std::string& nsName,
+                                               CommandGetTopicsOfNamespace_Mode mode, uint64_t requestId) {
     BaseCommand cmd;
     cmd.set_type(BaseCommand::GET_TOPICS_OF_NAMESPACE);
     CommandGetTopicsOfNamespace* getTopics = cmd.mutable_gettopicsofnamespace();
     getTopics->set_request_id(requestId);
     getTopics->set_namespace_(nsName);
+    getTopics->set_mode(static_cast<proto::CommandGetTopicsOfNamespace_Mode>(mode));
 
     const SharedBuffer buffer = writeMessageWithSize(cmd);
     cmd.clear_gettopicsofnamespace();

lib/Commands.h

@@ -156,7 +156,8 @@ class Commands {
     static SharedBuffer newSeek(uint64_t consumerId, uint64_t requestId, const MessageId& messageId);
     static SharedBuffer newSeek(uint64_t consumerId, uint64_t requestId, uint64_t timestamp);
     static SharedBuffer newGetLastMessageId(uint64_t consumerId, uint64_t requestId);
-    static SharedBuffer newGetTopicsOfNamespace(const std::string& nsName, uint64_t requestId);
+    static SharedBuffer newGetTopicsOfNamespace(const std::string& nsName,
+                                                CommandGetTopicsOfNamespace_Mode mode, uint64_t requestId);
 
     static bool peerSupportsGetLastMessageId(int32_t peerVersion);
     static bool peerSupportsActiveConsumerListener(int32_t peerVersion);