Boost URL support

Now require/link Boost::url when using Boost 1.81+; fallback logic keeps compatibility with older Boost versions.
Harden ETP session code: avoid holding locks while doing work, add sleeps in busy-wait loops to prevent spin, early-return on resolver errors.

Author: philippeVerney

CMakeLists.txt

@@ -103,16 +103,25 @@ set (WITH_ETP_SSL ON CACHE BOOL "Build FETPAPI with ETP SSL support.")
 find_package (Threads)
 
 # Boost DEPENDENCY
-find_package(Boost 1.70)
+find_package(Boost 1.81)
 if (NOT Boost_FOUND)
-	# Boost system is required for Beast until version 1.70 : https://www.boost.org/doc/libs/1_69_0/libs/beast/doc/html/beast/introduction.html
-	find_package(Boost 1.66 REQUIRED system)
+	find_package(Boost 1.70 QUIET)
+	if (NOT Boost_FOUND)
+		# Boost system is required for Beast until version 1.70 : https://www.boost.org/doc/libs/1_69_0/libs/beast/doc/html/beast/introduction.html
+		find_package(Boost 1.66 REQUIRED COMPONENTS system)
+	endif()
+else()
+	# Boost URL is mandatory for FETPAPI when using Boost1.81
+	# VCPKG can install Boost 1.81 without URL
+	find_package(Boost 1.81 REQUIRED COMPONENTS url)
 endif()
 if (WIN32 AND (Boost_VERSION_MAJOR EQUAL 1) AND (Boost_VERSION_MINOR LESS 74) AND (Boost_VERSION_MINOR GREATER 71))
 	message(WARNING "You may experience min/max issue with this boost version : See https://github.com/boostorg/beast/issues/1980")
 endif ()
 target_compile_definitions(${PROJECT_NAME} PRIVATE BOOST_ALL_NO_LIB)
-if (DEFINED Boost_SYSTEM_LIBRARY)
+if(TARGET Boost::url)
+	target_link_libraries (${PROJECT_NAME} PRIVATE Boost::url)
+elseif (TARGET Boost::system)
 	target_link_libraries (${PROJECT_NAME} PRIVATE Boost::system)
 else ()
 	target_link_libraries (${PROJECT_NAME} PRIVATE Boost::boost)

cmake/swigEtp1_2Include.i.in

@@ -1937,20 +1937,20 @@ namespace ETP_NS
 		*/
 		std::string startTransaction(std::vector<std::string> dataspaceUris = {}, bool readOnly = false);
 
-		/**
-		* A customer sends to a store to commit and end a transaction. This message implies that the customer 
-		* has received from or sent to the store all the data required for some purpose. The customer asserts that 
-		* the data sent in the scope of this transaction is a consistent unit of work.
-		* It actually sends a message and block the current thread until a response has been received from the store.
+		/*
+		* A customer sends to a store to cancel a transaction. The store MUST disregard any requests or data sent
+		* with that transaction. The current transaction (the one being canceled) MUST NOT change the state of
+		* the store.
+		* It actually sends a message and blocks the current thread until a response has been received from the store.
 		* @return Failure message or empty string if success
 		*/
 		std::string rollbackTransaction();
 
-		/*
-		* A customer sends to a store to cancel a transaction. The store MUST disregard any requests or data sent 
-		* with that transaction. The current transaction (the one being canceled) MUST NOT change the state of 
-		* the store.
-		* It actually sends a message and block the current thread until a response has been received from the store.
+		/**
+		* A customer sends to a store to commit and end a transaction. This message implies that the customer 
+		* has received from or sent to the store all the data required for some purpose. The customer asserts that 
+		* the data sent in the scope of this transaction is a consistent unit of work.
+		* It actually sends a message and blocks the current thread until a response has been received from the store.
 		* @return Failure message or empty string if success
 		*/
 		std::string commitTransaction();

src/fetpapi/etp/AbstractClientSessionCRTP.h

@@ -128,48 +128,47 @@ namespace ETP_NS
 			}
 
 			auto& front = sendingQueue.front();
-			const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
-			bool previousSentMessageCompleted = specificProtocolHandlers.find(std::get<0>(front)->messageHeader.messageId) == specificProtocolHandlers.end();
+			const bool previousSentMessageCompleted = [this, front]() {
+				const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
+				return specificProtocolHandlers.find(std::get<0>(front)->messageHeader.messageId) == specificProtocolHandlers.end();
+			}();
 
 			if (!previousSentMessageCompleted) {
-				fesapi_log("Cannot send Message id :", std::to_string(std::get<0>(front)->messageHeader.messageId), "because the previous message has not finished to be sent.");
+				fesapi_log("Cannot send Message id :", std::to_string(std::get<0>(front)->messageHeader.messageId), " because the previous message has not finished to be sent.");
+				return;
+			}		
+			fesapi_log("Sending Message id :", std::to_string(std::get<0>(front)->messageHeader.messageId));
+			
+			auto avroBytes = std::get<0>(front)->encodeHeaderAndBody();
+			if (avroBytes->size() >= maxWebSocketMessagePayloadSize) {
+				throw std::invalid_argument("You cannot send a message which is too big. Please use message part or chunk or whatever else.");
 			}
-			else {
-				fesapi_log("Sending Message id :", std::to_string(std::get<0>(front)->messageHeader.messageId));
-
-				auto avroBytes = std::get<0>(front)->encodeHeaderAndBody();
-
-				//asio::buffer is a non-owning view. We must keep the underlying storage alive until the I/O completes.
-				if (avroBytes->size() < maxWebSocketMessagePayloadSize) {
-					derived().ws()->async_write(
-						boost::asio::buffer(*avroBytes),
-						[this, self{ this->shared_from_this() }, avroBytes](boost::system::error_code ec, std::size_t)
-						->void
-					{
-
-						if (ec) {
-							std::cerr << "on_write : " << ec.message() << std::endl;
-						}
-						else {
-							// Register the handler to respond to the sent message
-							const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
-							auto& front = sendingQueue.front();
-							auto nextMessage = std::get<0>(front);
-							specificProtocolHandlers[nextMessage->messageHeader.messageId] =
-								std::make_tuple(nextMessage, std::get<1>(front), std::get<2>(front));
-						}
-
-						// Remove the sent message from the queue
-						const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
-						sendingQueue.pop();
-
-						do_write();
-					});
+
+			//asio::buffer is a non-owning view. We must keep the underlying storage alive until the I/O completes.
+			derived().ws()->async_write(
+				boost::asio::buffer(*avroBytes),
+				[this, self{ this->shared_from_this() }, avroBytes](boost::system::error_code ec, std::size_t)
+				->void
+			{
+
+				if (ec) {
+					std::cerr << "on_write : " << ec.message() << std::endl;
 				}
 				else {
-					throw std::invalid_argument("You cannot send a message which is too big. Please use message part or chunk or whatever else.");
+					// Register the handler to respond to the sent message
+					const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
+					auto& front = sendingQueue.front();
+					auto nextMessage = std::get<0>(front);
+					specificProtocolHandlers[nextMessage->messageHeader.messageId] =
+						std::make_tuple(nextMessage, std::get<1>(front), std::get<2>(front));
 				}
-			}
+
+				// Remove the sent message from the queue
+				const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
+				sendingQueue.pop();
+
+				do_write();
+			});
 		}
 	};
 }

src/fetpapi/etp/AbstractSession.cpp

@@ -103,7 +103,7 @@ void AbstractSession::on_read(boost::system::error_code ec, std::size_t bytes_tr
 
 	// Request for Acknowledge
 	if ((receivedMh.messageFlags & 0x10) != 0) {
-		std::shared_ptr < Energistics::Etp::v12::Protocol::Core::Acknowledge> acknowledge;
+		auto acknowledge = std::make_shared<Energistics::Etp::v12::Protocol::Core::Acknowledge>();
 		acknowledge->messageHeader.protocol = receivedMh.protocol;
 		send(acknowledge, receivedMh.messageId, 0x02);
 	}
@@ -117,42 +117,21 @@ void AbstractSession::on_read(boost::system::error_code ec, std::size_t bytes_tr
 			// Receive Protocol Exception
 			protocolHandlers[static_cast<int32_t>(Energistics::Etp::v12::Datatypes::Protocol::Core)]->decodeMessageBody(receivedMh, d);
 			if ((receivedMh.messageFlags & 0x02) != 0) {
-				const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
-				auto specificProtocolHandlerIt = specificProtocolHandlers.find(receivedMh.correlationId);
-				for (int64_t idAlias : std::get<2>(specificProtocolHandlerIt->second)) {
-					auto specificProtocolHandlerIt2 = specificProtocolHandlers.find(idAlias);
-					if (specificProtocolHandlerIt2 != specificProtocolHandlers.end()) {
-						specificProtocolHandlers.erase(specificProtocolHandlerIt2);
-					}
-				}
-				if (specificProtocolHandlerIt != specificProtocolHandlers.end()) {
-					specificProtocolHandlers.erase(specificProtocolHandlerIt);
-				}
+				eraseFromSpecificProtocolHandlers(receivedMh.correlationId);
 			}
 		}
 		else {
-			std::shared_ptr<ETP_NS::ProtocolHandlers> specificProtocolHandler;
-			{
+			std::shared_ptr<ETP_NS::ProtocolHandlers> specificProtocolHandler = [this, &receivedMh]() {
 				const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
 				auto specificProtocolHandlerIt = specificProtocolHandlers.find(receivedMh.correlationId);
-				if (specificProtocolHandlerIt != specificProtocolHandlers.end()) {
-					specificProtocolHandler = std::get<1>(specificProtocolHandlerIt->second);
-				}
-			} // Scope for specificProtocolHandlersLock
+				return specificProtocolHandlerIt != specificProtocolHandlers.end() ? std::get<1>(specificProtocolHandlerIt->second) : nullptr;
+			}();
 
 			if (specificProtocolHandler) {
 				// Receive a message which has been asked to be processed with a specific protocol handler
 				specificProtocolHandler->decodeMessageBody(receivedMh, d);
 				if ((receivedMh.messageFlags & 0x02) != 0) {
-					const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
-					auto specificProtocolHandlerIt = specificProtocolHandlers.find(receivedMh.correlationId);
-					for (int64_t idAlias : std::get<2>(specificProtocolHandlerIt->second)) {
-						auto specificProtocolHandlerIt2 = specificProtocolHandlers.find(idAlias);
-						if (specificProtocolHandlerIt2 != specificProtocolHandlers.end()) {
-							specificProtocolHandlers.erase(specificProtocolHandlerIt2);
-						}
-					}
-					specificProtocolHandlers.erase(specificProtocolHandlerIt);
+					eraseFromSpecificProtocolHandlers(receivedMh.correlationId);
 				}
 			}
 			else {
@@ -174,7 +153,11 @@ void AbstractSession::on_read(boost::system::error_code ec, std::size_t bytes_tr
 		send(ETP_NS::EtpHelpers::buildSingleMessageProtocolException(19, "The agent is unable to de-serialize the body of the message id " + std::to_string(receivedMh.messageId) + " : " + std::string(e.what())), 0, 0x02);
 	}
 
-	if (specificProtocolHandlers.empty() && isCloseRequested_)
+	const bool specificProtocolHandlersIsEmpty = [this]() {
+		const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
+		return specificProtocolHandlers.empty();
+	}();
+	if (specificProtocolHandlersIsEmpty && isCloseRequested_)
 	{
 		etpSessionClosed = true;
 		send(std::make_shared<Energistics::Etp::v12::Protocol::Core::CloseSession>(), 0, 0x02);

src/fetpapi/etp/AbstractSession.h

@@ -187,6 +187,7 @@ namespace ETP_NS
 
 			auto t_start = std::chrono::high_resolution_clock::now();
 			while (isMessageStillProcessing(correlationId == 0 ? msgId : correlationId)) {
+				std::this_thread::sleep_for(std::chrono::milliseconds(100));
 				if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > _timeOut) {
 					throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
 				}
@@ -284,6 +285,7 @@ namespace ETP_NS
 
 			const auto t_start = std::chrono::high_resolution_clock::now();
 			while (isMessageStillProcessing(correlationId == 0 ? msgId : correlationId)) {
+				std::this_thread::sleep_for(std::chrono::milliseconds(100));
 				if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > _timeOut) {
 					throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
 				}
@@ -292,7 +294,9 @@ namespace ETP_NS
 			// If the message has not been answered correctly
 			if (isEtpSessionClosed() && !isCloseRequested()) {
 				// Wait for a reconnection
-				while (isEtpSessionClosed() && !isCloseRequested()) {}
+				while (isEtpSessionClosed() && !isCloseRequested()) {
+					std::this_thread::sleep_for(std::chrono::milliseconds(100));
+				}
 				// Check if reconnection is successfull
 				if (isEtpSessionClosed()) {
 					throw std::runtime_error("The ETP session could not be opened in order to send again the message.");
@@ -364,18 +368,15 @@ namespace ETP_NS
 		*/
 		FETPAPI_DLL_IMPORT_OR_EXPORT void close() {
 			isCloseRequested_ = true;
-			sendingQueueMutex.lock();
-			specificProtocolHandlersMutex.lock();
-			if (specificProtocolHandlers.empty() && sendingQueue.empty()) {
+			const bool shouldSendCloseMsgNow = [this]() {
+				std::scoped_lock lock(sendingQueueMutex, specificProtocolHandlersMutex);
+				return specificProtocolHandlers.empty() && sendingQueue.empty();
+			}();
+			
+			if (shouldSendCloseMsgNow) {
 				etpSessionClosed = true;
-				sendingQueueMutex.unlock();
-				specificProtocolHandlersMutex.unlock();
 				send(std::make_shared<Energistics::Etp::v12::Protocol::Core::CloseSession>(), 0, 0x02);
 			}
-			else {
-				sendingQueueMutex.unlock();
-				specificProtocolHandlersMutex.unlock();
-			}
 		}
 
 		/**
@@ -387,6 +388,7 @@ namespace ETP_NS
 			close();
 			auto t_start = std::chrono::high_resolution_clock::now();
 			while (!webSocketSessionClosed) {
+				std::this_thread::sleep_for(std::chrono::milliseconds(100));
 				if (std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() > _timeOut) {
 					throw std::runtime_error("Time out waiting for closing");
 				}
@@ -594,19 +596,19 @@ namespace ETP_NS
 		*/
 		FETPAPI_DLL_IMPORT_OR_EXPORT std::string startTransaction(std::vector<std::string> dataspaceUris = {}, bool readOnly = false);
 
-		/**
-		* A customer sends to a store to commit and end a transaction. This message implies that the customer 
-		* has received from or sent to the store all the data required for some purpose. The customer asserts that 
-		* the data sent in the scope of this transaction is a consistent unit of work.
+		/*
+		* A customer sends to a store to cancel a transaction. The store MUST disregard any requests or data sent
+		* with that transaction. The current transaction (the one being canceled) MUST NOT change the state of
+		* the store.
 		* It actually sends a message and blocks the current thread until a response has been received from the store.
 		* @return Failure message or empty string if success
 		*/
 		FETPAPI_DLL_IMPORT_OR_EXPORT std::string rollbackTransaction();
 
-		/*
-		* A customer sends to a store to cancel a transaction. The store MUST disregard any requests or data sent 
-		* with that transaction. The current transaction (the one being canceled) MUST NOT change the state of 
-		* the store.
+		/**
+		* A customer sends to a store to commit and end a transaction. This message implies that the customer 
+		* has received from or sent to the store all the data required for some purpose. The customer asserts that 
+		* the data sent in the scope of this transaction is a consistent unit of work.
 		* It actually sends a message and blocks the current thread until a response has been received from the store.
 		* @return Failure message or empty string if success
 		*/
@@ -664,9 +666,10 @@ namespace ETP_NS
 		/// The identifier of the session
 		boost::uuids::uuid identifier{ boost::uuids::nil_uuid() };
 		std::mutex identifierMutex;
-		/// Indicates that the endpoint request to close the websocket session 
-		bool isCloseRequested_{ false };
+		/// Indicates that the endpoint request to close the websocket session
+		std::atomic<bool> isCloseRequested_{ false };
 		size_t reconnectionTryCount_ = 0;
+		static constexpr size_t maxReconnectionTryCount_ = 10;
 
 		AbstractSession() = default;
 
@@ -692,6 +695,21 @@ namespace ETP_NS
 		 */
 		Energistics::Etp::v12::Datatypes::MessageHeader decodeMessageHeader(avro::DecoderPtr decoder);
 
+		/**
+		* Erase all information about a message in specificProtocolHandlers.
+		* The message is identified by its ID and all its aliases.
+		*/
+		void eraseFromSpecificProtocolHandlers(int64_t msgId) {
+			const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
+			auto specificProtocolHandlerIt = specificProtocolHandlers.find(msgId);
+			if (specificProtocolHandlerIt != specificProtocolHandlers.end()) {
+				for (int64_t idAlias : std::get<2>(specificProtocolHandlerIt->second)) {
+					specificProtocolHandlers.erase(idAlias);
+				}
+				specificProtocolHandlers.erase(specificProtocolHandlerIt);
+			}
+		}
+
 		std::shared_ptr<ETP_NS::CoreHandlers> getCoreProtocolHandlers() {
 			auto it = protocolHandlers.find(static_cast<std::underlying_type<Energistics::Etp::v12::Datatypes::Protocol>::type>(Energistics::Etp::v12::Datatypes::Protocol::Core));
 			return it == protocolHandlers.end()

src/fetpapi/etp/ClientSession.h

@@ -64,23 +64,25 @@ namespace ETP_NS
 			// Run will return only when there will no more be any uncomplete operations (such as a reading operation for example)
 			getIoContext().run();
 
-			// Try to reconnect up to 10 times
-			if (!isCloseRequested_ && reconnectionTryCount_ < 10) {
-				++reconnectionTryCount_;
-				std::cerr << "Session " << getIdentifier() << " has been disconnected, trying to reconnect... " << reconnectionTryCount_ << "/10" << std::endl;
-				getIoContext().restart();
-				run();
-			}
-
-			if (!isCloseRequested_ && reconnectionTryCount_ >= 10) {
-				std::cerr << "Could not reconnect after 10 retries... Give up and close" << reconnectionTryCount_ << "/10" << std::endl;
-				isCloseRequested_ = true;
+			if (!isCloseRequested_) {
+				// Try to reconnect up to 10 times
+				if (reconnectionTryCount_ < maxReconnectionTryCount_) {
+					++reconnectionTryCount_;
+					std::cerr << "Session " << getIdentifier() << " has been disconnected, trying to reconnect... " << reconnectionTryCount_ << "/" << maxReconnectionTryCount_ << std::endl;
+					getIoContext().restart();
+					run();
+				}
+				else {
+					std::cerr << "Could not reconnect after " << maxReconnectionTryCount_ << " retries... Give up and close" << reconnectionTryCount_ << "/" << maxReconnectionTryCount_ << std::endl;
+					isCloseRequested_ = true;
+				}
 			}
 		}
 
 		void on_resolve(boost::system::error_code ec, tcp::resolver::results_type results) {
 			if (ec) {
 				std::cerr << "on_resolve : " << ec.message() << std::endl;
+				return;
 			}
 
 			asyncConnect(results);

src/fetpapi/etp/ClientSessionLaunchers.cpp

@@ -141,6 +141,5 @@ std::shared_ptr<ETP_NS::ClientSession> ETP_NS::ClientSessionLaunchers::createCli
 	}
 #endif
 
-	initializationParams->postSessionCreationOperation(result.get());
 	return result;
 }