add tests to verify message is correctly set

Author: BewareMyPower

include/pulsar/Result.h

@@ -24,6 +24,7 @@
 #include <cstdint>
 #include <iosfwd>
 #include <string>
+#include <utility>
 
 namespace pulsar {
 
@@ -104,7 +105,13 @@ PULSAR_PUBLIC const char* strResult(Result result);
 PULSAR_PUBLIC std::ostream& operator<<(std::ostream& s, pulsar::Result result);
 
 struct Error {
-    Result result;
+    Error() = default;
+    Error(Result result) : result(result) {}
+    Error(Result result, std::string message) : result(result), message(std::move(message)) {}
+
+    operator Result() const { return result; }
+
+    Result result = ResultOk;
     std::string message;
 };
 

lib/BinaryProtoLookupService.cc

@@ -37,7 +37,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
     -> LookupResultFuture {
     LOG_DEBUG("find broker from " << address << ", authoritative: " << authoritative << ", topic: " << topic
                                   << ", redirect count: " << redirectCount);
-    auto promise = std::make_shared<Promise<Result, LookupResult>>();
+    auto promise = std::make_shared<Promise<Error, LookupResult>>();
     if (maxLookupRedirects_ > 0 && redirectCount > maxLookupRedirects_) {
         LOG_ERROR("Too many lookup request redirects on topic " << topic << ", configured limit is "
                                                                 << maxLookupRedirects_);
@@ -62,7 +62,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
         auto lookupPromise = std::make_shared<LookupDataResultPromise>();
         cnx->newTopicLookup(topic, authoritative, listenerName_, newRequestId(), lookupPromise);
         lookupPromise->getFuture().addListener([this, cnx, promise, topic, address, redirectCount](
-                                                   Result result, const LookupDataResultPtr& data) {
+                                                   Error result, const LookupDataResultPtr& data) {
             if (result != ResultOk || !data) {
                 LOG_ERROR("Lookup failed for " << topic << ", result " << result);
                 promise->setFailed(result);
@@ -74,7 +74,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
             if (data->isRedirect()) {
                 LOG_DEBUG("Lookup request is for " << topic << " redirected to " << responseBrokerAddress);
                 findBroker(responseBrokerAddress, data->isAuthoritative(), topic, redirectCount + 1)
-                    .addListener([promise](Result result, const LookupResult& value) {
+                    .addListener([promise](Error result, const LookupResult& value) {
                         if (result == ResultOk) {
                             promise->setValue(value);
                         } else {
@@ -100,7 +100,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
  * @param    topicName topic to get number of partitions.
  *
  */
-Future<Result, LookupDataResultPtr> BinaryProtoLookupService::getPartitionMetadataAsync(
+Future<Error, LookupDataResultPtr> BinaryProtoLookupService::getPartitionMetadataAsync(
     const TopicNamePtr& topicName) {
     LookupDataResultPromisePtr promise = std::make_shared<LookupDataResultPromise>();
     if (!topicName) {
@@ -135,7 +135,7 @@ void BinaryProtoLookupService::sendPartitionMetadataLookupRequest(const std::str
                                                      std::placeholders::_2, clientCnx, promise));
 }
 
-void BinaryProtoLookupService::handlePartitionMetadataLookup(const std::string& topicName, Result result,
+void BinaryProtoLookupService::handlePartitionMetadataLookup(const std::string& topicName, Error result,
                                                              const LookupDataResultPtr& data,
                                                              const ClientConnectionWeakPtr& clientCnx,
                                                              const LookupDataResultPromisePtr& promise) {
@@ -154,9 +154,9 @@ uint64_t BinaryProtoLookupService::newRequestId() {
     return ++requestIdGenerator_;
 }
 
-Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespaceAsync(
+Future<Error, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespaceAsync(
     const NamespaceNamePtr& nsName, CommandGetTopicsOfNamespace_Mode mode) {
-    NamespaceTopicsPromisePtr promise = std::make_shared<Promise<Result, NamespaceTopicsPtr>>();
+    NamespaceTopicsPromisePtr promise = std::make_shared<Promise<Error, NamespaceTopicsPtr>>();
     if (!nsName) {
         promise->setFailed(ResultInvalidTopicName);
         return promise->getFuture();
@@ -168,9 +168,9 @@ Future<Result, NamespaceTopicsPtr> BinaryProtoLookupService::getTopicsOfNamespac
     return promise->getFuture();
 }
 
-Future<Result, SchemaInfo> BinaryProtoLookupService::getSchema(const TopicNamePtr& topicName,
-                                                               const std::string& version) {
-    GetSchemaPromisePtr promise = std::make_shared<Promise<Result, SchemaInfo>>();
+Future<Error, SchemaInfo> BinaryProtoLookupService::getSchema(const TopicNamePtr& topicName,
+                                                              const std::string& version) {
+    GetSchemaPromisePtr promise = std::make_shared<Promise<Error, SchemaInfo>>();
 
     if (!topicName) {
         promise->setFailed(ResultInvalidTopicName);
@@ -197,7 +197,7 @@ void BinaryProtoLookupService::sendGetSchemaRequest(const std::string& topicName
                                                   << " version: " << version);
 
     conn->newGetSchema(topicName, version, requestId)
-        .addListener([promise](Result result, const SchemaInfo& schemaInfo) {
+        .addListener([promise](Error result, const SchemaInfo& schemaInfo) {
             if (result != ResultOk) {
                 promise->setFailed(result);
                 return;
@@ -228,11 +228,10 @@ void BinaryProtoLookupService::sendGetTopicsOfNamespaceRequest(const std::string
                                std::placeholders::_1, std::placeholders::_2, promise));
 }
 
-void BinaryProtoLookupService::getTopicsOfNamespaceListener(Result result,
-                                                            const NamespaceTopicsPtr& topicsPtr,
+void BinaryProtoLookupService::getTopicsOfNamespaceListener(Error result, const NamespaceTopicsPtr& topicsPtr,
                                                             const NamespaceTopicsPromisePtr& promise) {
     if (result != ResultOk) {
-        promise->setFailed(ResultLookupError);
+        promise->setFailed(Error{ResultLookupError, result.message});
         return;
     }
 

lib/BinaryProtoLookupService.h

@@ -34,8 +34,8 @@ using ClientConnectionWeakPtr = std::weak_ptr<ClientConnection>;
 class ConnectionPool;
 class LookupDataResult;
 class ServiceNameResolver;
-using NamespaceTopicsPromisePtr = std::shared_ptr<Promise<Result, NamespaceTopicsPtr>>;
-using GetSchemaPromisePtr = std::shared_ptr<Promise<Result, SchemaInfo>>;
+using NamespaceTopicsPromisePtr = std::shared_ptr<Promise<Error, NamespaceTopicsPtr>>;
+using GetSchemaPromisePtr = std::shared_ptr<Promise<Error, SchemaInfo>>;
 
 class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
    public:
@@ -48,12 +48,12 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
 
     LookupResultFuture getBroker(const TopicName& topicName) override;
 
-    Future<Result, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) override;
+    Future<Error, LookupDataResultPtr> getPartitionMetadataAsync(const TopicNamePtr& topicName) override;
 
-    Future<Result, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(
+    Future<Error, NamespaceTopicsPtr> getTopicsOfNamespaceAsync(
         const NamespaceNamePtr& nsName, CommandGetTopicsOfNamespace_Mode mode) override;
 
-    Future<Result, SchemaInfo> getSchema(const TopicNamePtr& topicName, const std::string& version) override;
+    Future<Error, SchemaInfo> getSchema(const TopicNamePtr& topicName, const std::string& version) override;
 
     ServiceNameResolver& getServiceNameResolver() override { return serviceNameResolver_; }
 
@@ -75,7 +75,7 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
                                             const ClientConnectionWeakPtr& clientCnx,
                                             const LookupDataResultPromisePtr& promise);
 
-    void handlePartitionMetadataLookup(const std::string& topicName, Result result,
+    void handlePartitionMetadataLookup(const std::string& topicName, Error result,
                                        const LookupDataResultPtr& data,
                                        const ClientConnectionWeakPtr& clientCnx,
                                        const LookupDataResultPromisePtr& promise);
@@ -87,7 +87,7 @@ class PULSAR_PUBLIC BinaryProtoLookupService : public LookupService {
     void sendGetSchemaRequest(const std::string& topicName, const std::string& version, Result result,
                               const ClientConnectionWeakPtr& clientCnx, const GetSchemaPromisePtr& promise);
 
-    void getTopicsOfNamespaceListener(Result result, const NamespaceTopicsPtr& topicsPtr,
+    void getTopicsOfNamespaceListener(Error result, const NamespaceTopicsPtr& topicsPtr,
                                       const NamespaceTopicsPromisePtr& promise);
 
     uint64_t newRequestId();

lib/Client.cc

@@ -70,7 +70,9 @@ std::variant<Producer, Error> Client::createProducerV2(const std::string& topic)
 std::variant<Producer, Error> Client::createProducerV2(const std::string& topic,
                                                        const ProducerConfiguration& conf) {
     Promise<bool, std::variant<Producer, Error> > promise;
-    createProducerAsyncV2(topic, conf, [promise](const auto& result) { promise.setValue(result); });
+    createProducerAsyncV2(topic, conf, [promise](std::variant<Producer, Error>&& result) {
+        promise.setValue(std::move(result));
+    });
     Future<bool, std::variant<Producer, Error> > future = promise.getFuture();
 
     std::variant<Producer, Error> result;
@@ -114,8 +116,9 @@ std::variant<Consumer, Error> Client::subscribeV2(const std::string& topic,
                                                   const std::string& subscriptionName,
                                                   const ConsumerConfiguration& conf) {
     Promise<bool, std::variant<Consumer, Error> > promise;
-    subscribeAsyncV2(topic, subscriptionName, conf,
-                     [promise](const auto& result) { promise.setValue(result); });
+    subscribeAsyncV2(topic, subscriptionName, conf, [promise](std::variant<Consumer, Error>&& result) {
+        promise.setValue(std::move(result));
+    });
     Future<bool, std::variant<Consumer, Error> > future = promise.getFuture();
 
     std::variant<Consumer, Error> result;
@@ -168,8 +171,9 @@ std::variant<Consumer, Error> Client::subscribeV2(const std::vector<std::string>
                                                   const std::string& subscriptionName,
                                                   const ConsumerConfiguration& conf) {
     Promise<bool, std::variant<Consumer, Error> > promise;
-    subscribeAsyncV2(topics, subscriptionName, conf,
-                     [promise](const auto& result) { promise.setValue(result); });
+    subscribeAsyncV2(topics, subscriptionName, conf, [promise](std::variant<Consumer, Error>&& result) {
+        promise.setValue(std::move(result));
+    });
     Future<bool, std::variant<Consumer, Error> > future = promise.getFuture();
 
     std::variant<Consumer, Error> result;
@@ -233,8 +237,9 @@ Result Client::createReader(const std::string& topic, const MessageId& startMess
 std::variant<Reader, Error> Client::createReaderV2(const std::string& topic, const MessageId& startMessageId,
                                                    const ReaderConfiguration& conf) {
     Promise<bool, std::variant<Reader, Error> > promise;
-    createReaderAsyncV2(topic, startMessageId, conf,
-                        [promise](const auto& result) { promise.setValue(result); });
+    createReaderAsyncV2(topic, startMessageId, conf, [promise](std::variant<Reader, Error>&& result) {
+        promise.setValue(std::move(result));
+    });
     Future<bool, std::variant<Reader, Error> > future = promise.getFuture();
 
     std::variant<Reader, Error> result;

lib/ClientConnection.cc

@@ -1048,7 +1048,7 @@ void ClientConnection::newLookup(const SharedBuffer& cmd, uint64_t requestId, co
                 self->numOfPendingLookupRequest_--;
             }
         });
-    request->getFuture().addListener([promise](Result result, const LookupDataResultPtr& lookupDataResult) {
+    request->getFuture().addListener([promise](Error result, const LookupDataResultPtr& lookupDataResult) {
         if (result == ResultOk) {
             promise->setValue(lookupDataResult);
         } else {
@@ -1413,13 +1413,13 @@ Future<Result, GetLastMessageIdResponse> ClientConnection::newGetLastMessageId(u
     return request->getFuture();
 }
 
-Future<Result, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(
+Future<Error, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(
     const std::string& nsName, CommandGetTopicsOfNamespace_Mode mode, uint64_t requestId) {
     Lock lock(mutex_);
     if (isClosed()) {
         lock.unlock();
         LOG_ERROR(cnxString() << "Client is not connected to the broker");
-        Promise<Result, NamespaceTopicsPtr> promise;
+        Promise<Error, NamespaceTopicsPtr> promise;
         promise.setFailed(ResultNotConnected);
         return promise.getFuture();
     }
@@ -1437,14 +1437,14 @@ Future<Result, NamespaceTopicsPtr> ClientConnection::newGetTopicsOfNamespace(
     return request->getFuture();
 }
 
-Future<Result, SchemaInfo> ClientConnection::newGetSchema(const std::string& topicName,
-                                                          const std::string& version, uint64_t requestId) {
+Future<Error, SchemaInfo> ClientConnection::newGetSchema(const std::string& topicName,
+                                                         const std::string& version, uint64_t requestId) {
     Lock lock(mutex_);
 
     if (isClosed()) {
         lock.unlock();
         LOG_ERROR(cnxString() << "Client is not connected to the broker");
-        Promise<Result, SchemaInfo> promise;
+        Promise<Error, SchemaInfo> promise;
         promise.setFailed(ResultNotConnected);
         return promise.getFuture();
     }
@@ -1556,7 +1556,8 @@ void ClientConnection::handlePartitionedMetadataResponse(
                                       << " msg: " << partitionMetadataResponse.message());
                 checkServerError(partitionMetadataResponse.error(), partitionMetadataResponse.message());
                 request->fail(
-                    getResult(partitionMetadataResponse.error(), partitionMetadataResponse.message()));
+                    Error{getResult(partitionMetadataResponse.error(), partitionMetadataResponse.message()),
+                          partitionMetadataResponse.message()});
             } else {
                 LOG_ERROR(cnxString() << "Failed partition-metadata lookup req_id: "
                                       << partitionMetadataResponse.request_id() << " with empty response: ");
@@ -1628,7 +1629,8 @@ void ClientConnection::handleLookupTopicRespose(
                                       << " error: " << lookupTopicResponse.error()
                                       << " msg: " << lookupTopicResponse.message());
                 checkServerError(lookupTopicResponse.error(), lookupTopicResponse.message());
-                request->fail(getResult(lookupTopicResponse.error(), lookupTopicResponse.message()));
+                request->fail(Error{getResult(lookupTopicResponse.error(), lookupTopicResponse.message()),
+                                    lookupTopicResponse.message()});
             } else {
                 LOG_ERROR(cnxString() << "Failed lookup req_id: " << lookupTopicResponse.request_id()
                                       << " with empty response: ");
@@ -1699,6 +1701,7 @@ void ClientConnection::handleProducerSuccess(const proto::CommandProducerSuccess
 
 void ClientConnection::handleError(const proto::CommandError& error) {
     Result result = getResult(error.error(), error.message());
+    Error errorResult{result, error.has_message() ? error.message() : ""};
     LOG_WARN(cnxString() << "Received error response from server: " << result
                          << (error.has_message() ? (" (" + error.message() + ")") : "")
                          << " -- req_id: " << error.request_id());
@@ -1716,27 +1719,51 @@ void ClientConnection::handleError(const proto::CommandError& error) {
             data.errorMessage = error.message();
         }
         request->fail(result, data);
-    } else {
-        auto it = pendingGetLastMessageIdRequests_.find(error.request_id());
-        if (it != pendingGetLastMessageIdRequests_.end()) {
-            auto request = std::move(it->second);
-            pendingGetLastMessageIdRequests_.erase(it);
-            lock.unlock();
+        return;
+    }
 
-            request->fail(result);
-        } else {
-            auto it = pendingGetNamespaceTopicsRequests_.find(error.request_id());
-            if (it != pendingGetNamespaceTopicsRequests_.end()) {
-                auto request = std::move(it->second);
-                pendingGetNamespaceTopicsRequests_.erase(it);
-                lock.unlock();
+    auto lookupIt = pendingLookupRequests_.find(error.request_id());
+    if (lookupIt != pendingLookupRequests_.end()) {
+        auto request = std::move(lookupIt->second);
+        pendingLookupRequests_.erase(lookupIt);
+        numOfPendingLookupRequest_--;
+        lock.unlock();
 
-                request->fail(result);
-            } else {
-                lock.unlock();
-            }
-        }
+        request->fail(errorResult);
+        return;
+    }
+
+    auto lastMessageIdIt = pendingGetLastMessageIdRequests_.find(error.request_id());
+    if (lastMessageIdIt != pendingGetLastMessageIdRequests_.end()) {
+        auto request = std::move(lastMessageIdIt->second);
+        pendingGetLastMessageIdRequests_.erase(lastMessageIdIt);
+        lock.unlock();
+
+        request->fail(result);
+        return;
+    }
+
+    auto topicsIt = pendingGetNamespaceTopicsRequests_.find(error.request_id());
+    if (topicsIt != pendingGetNamespaceTopicsRequests_.end()) {
+        auto request = std::move(topicsIt->second);
+        pendingGetNamespaceTopicsRequests_.erase(topicsIt);
+        lock.unlock();
+
+        request->fail(errorResult);
+        return;
     }
+
+    auto schemaIt = pendingGetSchemaRequests_.find(error.request_id());
+    if (schemaIt != pendingGetSchemaRequests_.end()) {
+        auto request = std::move(schemaIt->second);
+        pendingGetSchemaRequests_.erase(schemaIt);
+        lock.unlock();
+
+        request->fail(errorResult);
+        return;
+    }
+
+    lock.unlock();
 }
 
 std::string ClientConnection::getMigratedBrokerServiceUrl(
@@ -1959,7 +1986,7 @@ void ClientConnection::handleGetSchemaResponse(const proto::CommandGetSchemaResp
                                              : "")
                                      << " -- req_id: " << response.request_id());
             }
-            request->fail(result);
+            request->fail(Error{result, response.has_error_message() ? response.error_message() : ""});
             return;
         }