Fix bad_weak_ptr when close a ClientConnection during construction (#350)

Fixes #348

Fixes #349

### Motivation

When `close` is called in `ClientConnection`'s constructor,
`shared_from_this()` will be called, which results in a
`std::bad_weak_ptr` error. This error does not happen before
#317 because
`shared_from_this()` could only be called when the `producers` or
`consumers` field is not empty.

### Modifications

Throw `ResultAuthenticationError` when there is anything wrong with
authentication in `ClientConnection`'s constructor. Then catch the
result in `ConnectionPool::getConnectionAsync` and returned the failed
future.

In addition, check `authentication_` even for non-TLS URLs. Otherwise,
the null authentication will be used to construct `CommandConnect`.

Add `testInvalidPlugin` and `testTlsConfigError` to verify the changes.

Author: BewareMyPower

lib/ClientConnection.cc

@@ -187,6 +187,12 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
       pool_(pool),
       poolIndex_(poolIndex) {
     LOG_INFO(cnxString_ << "Create ClientConnection, timeout=" << clientConfiguration.getConnectionTimeout());
+    if (!authentication_) {
+        LOG_ERROR("Invalid authentication plugin");
+        throw ResultAuthenticationError;
+        return;
+    }
+
     if (clientConfiguration.isUseTls()) {
 #if BOOST_VERSION >= 105400
         boost::asio::ssl::context ctx(boost::asio::ssl::context::tlsv12_client);
@@ -207,20 +213,13 @@ 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 ResultAuthenticationError;
                 }
             } else {
                 ctx.set_default_verify_paths();
             }
         }
 
-        if (!authentication_) {
-            LOG_ERROR("Invalid authentication plugin");
-            close(ResultAuthenticationError, false);
-            return;
-        }
-
         std::string tlsCertificates = clientConfiguration.getTlsCertificateFilePath();
         std::string tlsPrivateKey = clientConfiguration.getTlsPrivateKeyFilePath();
 
@@ -231,13 +230,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 ResultAuthenticationError;
             }
             if (!file_exists(tlsCertificates)) {
                 LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
-                close(ResultAuthenticationError, false);
-                return;
+                throw ResultAuthenticationError;
             }
             ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
             ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);

lib/ClientConnection.h

@@ -148,9 +148,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
      * @param result all pending futures will complete with this result
      * @param detach remove it from the pool if it's true
      *
-     * `detach` should only be false when:
-     * 1. Before the connection is put into the pool, i.e. during the construction.
-     * 2. When the connection pool is closed
+     * `detach` should only be false when the connection pool is closed.
      */
     void close(Result result = ResultConnectError, bool detach = true);
 

lib/ConnectionPool.cc

@@ -99,6 +99,10 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
         cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
                                        clientConfiguration_, authentication_, clientVersion_, *this,
                                        keySuffix));
+    } catch (Result result) {
+        Promise<Result, ClientConnectionWeakPtr> promise;
+        promise.setFailed(result);
+        return promise.getFuture();
     } catch (const std::runtime_error& e) {
         lock.unlock();
         LOG_ERROR("Failed to create connection: " << e.what())

tests/AuthPluginTest.cc

@@ -581,3 +581,19 @@ TEST(AuthPluginTest, testOauth2Failure) {
     ASSERT_EQ(client5.createProducer(topic, producer), ResultAuthenticationError);
     client5.close();
 }
+
+TEST(AuthPluginTest, testInvalidPlugin) {
+    Client client("pulsar://localhost:6650", ClientConfiguration{}.setAuth(AuthFactory::create("invalid")));
+    Producer producer;
+    ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+    client.close();
+}
+
+TEST(AuthPluginTest, testTlsConfigError) {
+    Client client(serviceUrlTls, ClientConfiguration{}
+                                     .setAuth(AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath))
+                                     .setTlsTrustCertsFilePath("invalid"));
+    Producer producer;
+    ASSERT_EQ(ResultAuthenticationError, client.createProducer("my-topic", producer));
+    client.close();
+}