Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix bad_weak_ptr when close a ClientConnection during construction #350

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 9 additions & 12 deletions lib/ClientConnection.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand All @@ -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();

Expand All @@ -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);
Expand Down
4 changes: 1 addition & 3 deletions lib/ClientConnection.h
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down
4 changes: 4 additions & 0 deletions lib/ConnectionPool.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
16 changes: 16 additions & 0 deletions tests/AuthPluginTest.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}