Enhanced error handling

src/impl/pollservice.cpp

@@ -46,7 +46,7 @@ void PollService::start() {
 
 void PollService::join() {
 	std::unique_lock lock(mMutex);
-	if(std::exchange(mStopped, true))
+	if (std::exchange(mStopped, true))
 		return;
 
 	lock.unlock();
@@ -60,11 +60,11 @@ void PollService::add(socket_t sock, Params params) {
 	std::unique_lock lock(mMutex);
 	assert(mSocks);
 
-	if (!params.callback)
-		throw std::invalid_argument("poll callback is null");
-
 	mSocks->erase(sock);
 
+	if (!params.callback)
+		return;
+
 	PLOG_VERBOSE << "Registering socket in poll service, direction=" << params.direction;
 	auto until = params.timeout ? std::make_optional(clock::now() + *params.timeout) : nullopt;
 	mSocks->emplace(sock, SocketEntry{std::move(params), std::move(until)});
@@ -117,42 +117,48 @@ void PollService::process(std::vector<struct pollfd> &pfds) {
 		if (jt == mSocks->end())
 			continue; // removed
 
-		auto &entry = jt->second;
-		const auto &params = entry.params;
+		try {
+			auto &entry = jt->second;
+			const auto &params = entry.params;
 
-		if (it->revents & POLLNVAL || it->revents & POLLERR) {
-			PLOG_VERBOSE << "Poll error event";
-			auto callback = std::move(params.callback);
-			mSocks->erase(sock);
-			callback(Event::Error);
-			continue;
-		}
+			if (it->revents & POLLNVAL || it->revents & POLLERR) {
+				PLOG_VERBOSE << "Poll error event";
+				auto callback = std::move(params.callback);
+				mSocks->erase(sock);
+				callback(Event::Error);
+				continue;
+			}
 
-		if (it->revents & POLLIN || it->revents & POLLOUT) {
-			entry.until =
-			    params.timeout ? std::make_optional(clock::now() + *params.timeout) : nullopt;
+			if (it->revents & POLLIN || it->revents & POLLOUT) {
+				entry.until =
+				    params.timeout ? std::make_optional(clock::now() + *params.timeout) : nullopt;
 
-			auto callback = params.callback;
+				auto callback = params.callback;
 
-			if (it->revents & POLLIN) {
-				PLOG_VERBOSE << "Poll in event";
-				params.callback(Event::In);
-			}
+				if (it->revents & POLLIN) {
+					PLOG_VERBOSE << "Poll in event";
+					params.callback(Event::In);
+				}
 
-			if (it->revents & POLLOUT) {
-				PLOG_VERBOSE << "Poll out event";
-				params.callback(Event::Out);
+				if (it->revents & POLLOUT) {
+					PLOG_VERBOSE << "Poll out event";
+					params.callback(Event::Out);
+				}
+
+				continue;
 			}
 
-			continue;
-		}
+			if (entry.until && clock::now() >= *entry.until) {
+				PLOG_VERBOSE << "Poll timeout event";
+				auto callback = std::move(params.callback);
+				mSocks->erase(sock);
+				callback(Event::Timeout);
+				continue;
+			}
 
-		if (entry.until && clock::now() >= *entry.until) {
-			PLOG_VERBOSE << "Poll timeout event";
-			auto callback = std::move(params.callback);
+		} catch (const std::exception &e) {
+			PLOG_WARNING << e.what();
 			mSocks->erase(sock);
-			callback(Event::Timeout);
-			continue;
 		}
 	}
 }

src/impl/tcptransport.cpp

@@ -120,7 +120,6 @@ bool TcpTransport::outgoing(message_ptr message) {
 		return true;
 
 	mSendQueue.push(message);
-
 	setPoll(PollService::Direction::Both);
 	return false;
 }
@@ -152,48 +151,50 @@ void TcpTransport::connect() {
 		try {
 			prepare(ai->ai_addr, socklen_t(ai->ai_addrlen));
 
-			auto callback = [this, result, ai, recurse](PollService::Event event) mutable {
-				try {
-					if (event == PollService::Event::Error)
-						throw std::runtime_error("TCP connection failed");
-
-					if (event == PollService::Event::Timeout)
-						throw std::runtime_error("TCP connection timed out");
-
-					if (event != PollService::Event::Out)
-						return;
+		} catch (const std::runtime_error &e) {
+			PLOG_DEBUG << e.what();
+			recurse(ai->ai_next, recurse);
+		}
 
-					int err = 0;
-					socklen_t errlen = sizeof(err);
-					if (::getsockopt(mSock, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen) != 0)
-						throw std::runtime_error("Failed to get socket error code");
+		auto callback = [this, result, ai, recurse](PollService::Event event) mutable {
+			try {
+				if (event == PollService::Event::Error)
+					throw std::runtime_error("TCP connection failed");
 
-					if (err != 0) {
-						std::ostringstream msg;
-						msg << "TCP connection failed, errno=" << err;
-						throw std::runtime_error(msg.str());
-					}
+				if (event == PollService::Event::Timeout)
+					throw std::runtime_error("TCP connection timed out");
 
-					PLOG_INFO << "TCP connected";
-					freeaddrinfo(result);
-					changeState(State::Connected);
+				if (event != PollService::Event::Out)
+					return;
 
-					setPoll(PollService::Direction::In);
+				int err = 0;
+				socklen_t errlen = sizeof(err);
+				if (::getsockopt(mSock, SOL_SOCKET, SO_ERROR, (char *)&err, &errlen) != 0)
+					throw std::runtime_error("Failed to get socket error code");
 
-				} catch (const std::runtime_error &e) {
-					PLOG_DEBUG << e.what();
-					recurse(ai->ai_next, recurse);
+				if (err != 0) {
+					std::ostringstream msg;
+					msg << "TCP connection failed, errno=" << err;
+					throw std::runtime_error(msg.str());
 				}
-			};
 
-			const auto timeout = 10s;
-			PollService::Instance().add(
-			    mSock, {PollService::Direction::Out, timeout, std::move(callback)});
+				PLOG_INFO << "TCP connected";
+				freeaddrinfo(result);
+				changeState(State::Connected);
+				setPoll(PollService::Direction::In);
 
-		} catch (const std::runtime_error &e) {
-			PLOG_DEBUG << e.what();
-			recurse(ai->ai_next, recurse);
-		}
+			} catch (const std::runtime_error &e) {
+				PLOG_DEBUG << e.what();
+				PollService::Instance().remove(mSock);
+				::closesocket(mSock);
+				mSock = INVALID_SOCKET;
+				recurse(ai->ai_next, recurse);
+			}
+		};
+
+		const auto timeout = 10s;
+		PollService::Instance().add(mSock,
+		                            {PollService::Direction::Out, timeout, std::move(callback)});
 	};
 
 	attempt(result, attempt);
@@ -304,8 +305,6 @@ bool TcpTransport::trySendMessage(message_ptr &message) {
 
 void TcpTransport::process(PollService::Event event) {
 	try {
-		PLOG_VERBOSE << "Poll event";
-
 		switch (event) {
 		case PollService::Event::Error: {
 			PLOG_WARNING << "TCP connection terminated";