DEVOPS-10175 smartptr connection pointers

src/Connection.cpp

@@ -22,6 +22,7 @@ Connection::Connection(const AMQP::Address& address, int timeout): timeout(timeo
 
 Connection::~Connection() {
 	delete pimpl;
+	pimpl = nullptr;
 }
 
 void Connection::connect() {

src/linux/ConnectionImpl.cpp

@@ -13,8 +13,8 @@ ConnectionImpl::ConnectionImpl(const AMQP::Address& address) :
         sslInited = true;
     } 
     eventLoop = event_base_new();
-    handler = new TCPHandler(eventLoop);
-    connection = new AMQP::TcpConnection(handler, address);
+    handler.reset(new TCPHandler(eventLoop));
+    connection.reset(new AMQP::TcpConnection(handler.get(), address));
     thread = std::thread(ConnectionImpl::loopThread, this);
 }
 
@@ -25,16 +25,20 @@ ConnectionImpl::~ConnectionImpl() {
     }
     event_base_loopbreak(eventLoop);
     thread.join();
-    delete connection;
-    delete handler;
+    connection.reset(nullptr);
+    handler.reset(nullptr);
 
     event_base_free(eventLoop);
 }
 
 void ConnectionImpl::loopThread(ConnectionImpl* thiz) {
     event_base* loop = thiz->eventLoop;
     while(!thiz->connection->closed()) {
-        event_base_loop(loop, EVLOOP_NONBLOCK);
+        try{
+            event_base_loop(loop, EVLOOP_NONBLOCK);
+        }catch(std::exception& ex){
+            Biterp::Logging::error("Channel loop error: " + std::string(ex.what()));
+        }
     }
 }
 
@@ -50,7 +54,7 @@ void ConnectionImpl::openChannel(std::unique_ptr<AMQP::TcpChannel>& channel) {
     std::condition_variable cv;
     bool ready = false;
 
-    channel.reset(new AMQP::TcpChannel(connection));
+    channel.reset(new AMQP::TcpChannel(connection.get()));
     channel->onReady([&]() {
         std::unique_lock<std::mutex> lock(m);
         ready = true;

src/linux/ConnectionImpl.h

@@ -21,8 +21,8 @@ class ConnectionImpl{
 
 private:
     event_base* eventLoop;
-    TCPHandler* handler;
-    AMQP::TcpConnection* connection;
+    std::unique_ptr<TCPHandler> handler;
+    std::unique_ptr<AMQP::TcpConnection> connection;
 
     std::unique_ptr<AMQP::TcpChannel> trChannel;
     std::thread thread;

src/windows/ConnectionImpl.cpp

@@ -4,8 +4,8 @@ ConnectionImpl::ConnectionImpl(const AMQP::Address& address) :
 	handler(address.hostname(), address.port(), address.secure()),
 	trChannel(nullptr)
 {
-	connection = new AMQP::Connection(&handler, address.login(), address.vhost());
-	handler.setConnection(connection);
+	connection.reset(new AMQP::Connection(&handler, address.login(), address.vhost()));
+	handler.setConnection(connection.get());
 	thread = std::thread(SimplePocoHandler::loopThread, &handler);
 }
 
@@ -16,7 +16,7 @@ ConnectionImpl::~ConnectionImpl() {
 		connection->close();
 	}
 	thread.join();
-	delete connection;
+	connection.reset(nullptr);
 }
 
 void ConnectionImpl::openChannel(std::unique_ptr<AMQP::Channel>& channel) {
@@ -29,7 +29,7 @@ void ConnectionImpl::openChannel(std::unique_ptr<AMQP::Channel>& channel) {
 	std::mutex m;
 	std::condition_variable cv;
 	bool ready = false;
-	channel.reset(new AMQP::Channel(connection));
+	channel.reset(new AMQP::Channel(connection.get()));
 	channel->onReady([&]() {
 		std::unique_lock<std::mutex> lock(m);
 		ready = true;

src/windows/ConnectionImpl.h

@@ -18,7 +18,7 @@ class ConnectionImpl{
 
 private:
 	SimplePocoHandler handler;
-	AMQP::Connection* connection;
+	std::unique_ptr<AMQP::Connection> connection;
 	std::unique_ptr<AMQP::Channel> trChannel;
 	std::unique_ptr<AMQP::Channel> rcChannel;
 	std::thread thread;