Use exception

Author: BewareMyPower

lib/ClientConnection.cc

@@ -189,8 +189,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
     LOG_INFO(cnxString_ << "Create ClientConnection, timeout=" << clientConfiguration.getConnectionTimeout());
     if (!authentication_) {
         LOG_ERROR("Invalid authentication plugin");
-        close(ResultAuthenticationError, false);
-        return;
+        throw ResultException{ResultAuthenticationError};
     }
     if (clientConfiguration.isUseTls()) {
 #if BOOST_VERSION >= 105400
@@ -212,8 +211,7 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
                     ctx.load_verify_file(trustCertFilePath);
                 } else {
                     LOG_ERROR(trustCertFilePath << ": No such trustCertFile");
-                    close(ResultAuthenticationError, false);
-                    return;
+                    throw ResultException{ResultAuthenticationError};
                 }
             } else {
                 ctx.set_default_verify_paths();
@@ -230,13 +228,11 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
             tlsPrivateKey = authData->getTlsPrivateKey();
             if (!file_exists(tlsCertificates)) {
                 LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
-                close(ResultAuthenticationError, false);
-                return;
+                throw ResultException{ResultAuthenticationError};
             }
             if (!file_exists(tlsCertificates)) {
                 LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
-                close(ResultAuthenticationError, false);
-                return;
+                throw ResultException{ResultAuthenticationError};
             }
             ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
             ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);
@@ -1259,13 +1255,7 @@ void ClientConnection::handleConsumerStatsTimeout(const boost::system::error_cod
     startConsumerStatsTimer(consumerStatsRequests);
 }
 
-void ClientConnection::close(Result result, bool constructCompleted) {
-    if (!constructCompleted) {
-        state_ = Disconnected;
-        connectPromise_.setFailed(result);
-        return;
-    }
-
+void ClientConnection::close(Result result) {
     Lock lock(mutex_);
     if (isClosed()) {
         return;

lib/ClientConnection.h

@@ -32,6 +32,7 @@
 #include <boost/asio/strand.hpp>
 #include <boost/optional.hpp>
 #include <deque>
+#include <exception>
 #include <functional>
 #include <memory>
 #include <string>
@@ -146,9 +147,8 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      * Close the connection.
      *
      * @param result all pending futures will complete with this result
-     * @param constructCompleted whether the construction is completed
      */
-    void close(Result result = ResultConnectError, bool constructCompleted = true);
+    void close(Result result = ResultConnectError);
 
     bool isClosed() const;
 

lib/ConnectionPool.cc

@@ -24,6 +24,7 @@
 #include "ClientConnection.h"
 #include "ExecutorService.h"
 #include "LogUtils.h"
+#include "ResultUtils.h"
 
 using boost::asio::ip::tcp;
 namespace ssl = boost::asio::ssl;
@@ -54,20 +55,17 @@ bool ConnectionPool::close() {
         auto& cnx = cnxIt->second;
         if (cnx) {
             // The 2nd argument is false because removing a value during the iteration will cause segfault
-            cnx->close(ResultDisconnected, false);
+            cnx->close(ResultDisconnected);
         }
     }
     pool_.clear();
     return true;
 }
 
-Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const std::string& logicalAddress,
-                                                                           const std::string& physicalAddress,
-                                                                           size_t keySuffix) {
+auto ConnectionPool::getConnectionAsync(const std::string& logicalAddress, const std::string& physicalAddress,
+                                        size_t keySuffix) -> ConnectionFuture {
     if (closed_) {
-        Promise<Result, ClientConnectionWeakPtr> promise;
-        promise.setFailed(ResultAlreadyClosed);
-        return promise.getFuture();
+        return ConnectionFuture::failure(ResultAlreadyClosed);
     }
 
     std::unique_lock<std::recursive_mutex> lock(mutex_);
@@ -99,12 +97,12 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
         cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
                                        clientConfiguration_, authentication_, clientVersion_, *this,
                                        keySuffix));
+    } catch (const ResultException& e) {
+        return ConnectionFuture::failure(e.result());
     } catch (const std::runtime_error& e) {
         lock.unlock();
         LOG_ERROR("Failed to create connection: " << e.what())
-        Promise<Result, ClientConnectionWeakPtr> promise;
-        promise.setFailed(ResultConnectError);
-        return promise.getFuture();
+        return ConnectionFuture::failure(ResultConnectError);
     }
 
     LOG_INFO("Created connection for " << key);

lib/ConnectionPool.h

@@ -53,6 +53,8 @@ class PULSAR_PUBLIC ConnectionPool {
 
     void remove(const std::string& key, ClientConnection* value);
 
+    using ConnectionFuture = Future<Result, ClientConnectionWeakPtr>;
+
     /**
      * Get a connection from the pool.
      * <p>
@@ -73,16 +75,15 @@ class PULSAR_PUBLIC ConnectionPool {
      * @param keySuffix the key suffix to choose which connection on the same broker
      * @return a future that will produce the ClientCnx object
      */
-    Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
-                                                               const std::string& physicalAddress,
-                                                               size_t keySuffix);
+    ConnectionFuture getConnectionAsync(const std::string& logicalAddress, const std::string& physicalAddress,
+                                        size_t keySuffix);
 
-    Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
-                                                               const std::string& physicalAddress) {
+    ConnectionFuture getConnectionAsync(const std::string& logicalAddress,
+                                        const std::string& physicalAddress) {
         return getConnectionAsync(logicalAddress, physicalAddress, generateRandomIndex());
     }
 
-    Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& address) {
+    ConnectionFuture getConnectionAsync(const std::string& address) {
         return getConnectionAsync(address, address);
     }
 

lib/Future.h

@@ -116,6 +116,8 @@ class Future {
 
     Result get(Type &result) { return state_->get(result); }
 
+    static Future<Result, Type> failure(Result result);
+
    private:
     InternalStatePtr<Result, Type> state_;
 
@@ -144,6 +146,13 @@ class Promise {
     const InternalStatePtr<Result, Type> state_;
 };
 
+template <typename Result, typename Type>
+inline Future<Result, Type> Future<Result, Type>::failure(Result result) {
+    Promise<Result, Type> promise;
+    promise.setFailed(result);
+    return promise.getFuture();
+}
+
 }  // namespace pulsar
 
 #endif

lib/ResultUtils.h

@@ -20,10 +20,22 @@
 
 #include <pulsar/Result.h>
 
+#include <exception>
+
 namespace pulsar {
 
 inline bool isResultRetryable(Result result) {
     return result == ResultRetryable || result == ResultDisconnected;
 }
 
+class ResultException : public std::exception {
+   public:
+    ResultException(Result result) : result_(result) {}
+    Result result() const noexcept { return result_; }
+    const char* what() const noexcept { return strResult(result_); }
+
+   private:
+    const Result result_;
+};
+
 }  // namespace pulsar