Adopting latest Beast classes (tcp_stream and asio::ssl::stream)

philippeVerney · philippeVerney · commit 7aca2704755b · 2025-07-15T11:35:06.000+02:00
Store the identifier/UUID of the session
Clean sending queue at disconnection time
Update the Jupyter notebook  to put a dataspace with ACLs and LegalTags
diff --git a/example/withFesapi/etpClient.cpp b/example/withFesapi/etpClient.cpp
@@ -737,6 +737,7 @@ int main(int argc, char **argv)
 
 	std::cout << "Creating a client session..." << std::endl;
 	auto clientSession = ETP_NS::ClientSessionLaunchers::createClientSession(&initializationParams, authorization);
+	clientSession->setVerbose(true);
 
 	repo.setHdfProxyFactory(new ETP_NS::FesapiHdfProxyFactory(clientSession.get()));
 
@@ -752,7 +753,6 @@ int main(int argc, char **argv)
 	}
 
 	clientSession->setTimeOut(60000);
-	clientSession->setVerbose(false);
 	askUser(clientSession, repo);
 
 	sessionThread.join();
diff --git a/python/example/PutHorizon.ipynb b/python/example/PutHorizon.ipynb
@@ -155,18 +155,26 @@
     "dataspace.uri = \"eml:///dataspace('\" + dataspace.path + \"')\"\n",
     "dataspace.customData = fetpapi.MapStringDataValue()\n",
     "dataValue = fetpapi.DataValue()\n",
-    "aos = fetpapi.ArrayOfString\n",
+    "aos = fetpapi.ArrayOfString()\n",
     "dataValue.item = fetpapi.DataValueitem_t()\n",
-    "aos.values = [\"data.default.viewers@osdu.example.com\"]\n",
+    "tmp = fetpapi.StringVector()\n",
+    "tmp.push_back(\"data.default.viewers@osdu.example.com\")\n",
+    "aos.values = tmp\n",
     "dataValue.item.set_ArrayOfString(aos)\n",
     "dataspace.customData[\"viewers\"] = dataValue\n",
-    "aos.values = [\"data.default.owners@osdu.example.com\"]\n",
+    "tmp.clear()\n",
+    "tmp.push_back(\"data.default.owners@osdu.example.com\")\n",
+    "aos.values = tmp\n",
     "dataValue.item.set_ArrayOfString(aos)\n",
     "dataspace.customData[\"owners\"] = dataValue\n",
-    "aos.values = [\"osdu-public-usa-dataset\"]\n",
+    "tmp.clear()\n",
+    "tmp.push_back(\"osdu-public-usa-dataset\")\n",
+    "aos.values = tmp\n",
     "dataValue.item.set_ArrayOfString(aos)\n",
     "dataspace.customData[\"legaltags\"] = dataValue\n",
-    "aos.values = [\"US\"]\n",
+    "tmp.clear()\n",
+    "tmp.push_back(\"US\")\n",
+    "aos.values = tmp\n",
     "dataValue.item.set_ArrayOfString(aos)\n",
     "dataspace.customData[\"otherRelevantDataCountries\"] = dataValue"
    ]
diff --git a/src/etp/AbstractClientSessionCRTP.h b/src/etp/AbstractClientSessionCRTP.h
@@ -30,7 +30,7 @@ namespace ETP_NS
 
 		virtual ~AbstractClientSessionCRTP() = default;
 
-		void on_connect(boost::system::error_code ec) {
+		void on_ssl_handshake(boost::system::error_code ec) {
 			if (ec) {
 				std::cerr << "ERROR at Websocket connection : " << ec.message() << std::endl;
 				return;
diff --git a/src/etp/AbstractSession.cpp b/src/etp/AbstractSession.cpp
@@ -50,6 +50,10 @@ void AbstractSession::on_read(boost::system::error_code ec, std::size_t bytes_tr
 			// This indicates that the web socket (and consequently etp) session was closed
 			std::cerr << "The other endpoint closed the web socket (and consequently etp) connection" << std::endl;
 		}
+		else if (ec == boost::beast::error::timeout) {
+			// This indicates that the web socket (and consequently etp) session was closed
+			std::cerr << "Beast timeout has been reached" << std::endl;
+		}
 		else {
 			// This indicates an unexpected error
 			std::cerr << "on_read : error code number " << ec.value() << std::endl;
@@ -63,6 +67,9 @@ void AbstractSession::on_read(boost::system::error_code ec, std::size_t bytes_tr
 
 		const std::lock_guard<std::mutex> specificProtocolHandlersLock(specificProtocolHandlersMutex);
 		specificProtocolHandlers.clear();
+		const std::lock_guard<std::mutex> sendingQueueLock(sendingQueueMutex);
+		std::queue< std::tuple<int64_t, std::vector<uint8_t>, std::shared_ptr<ETP_NS::ProtocolHandlers>> > empty;
+		std::swap(sendingQueue, empty);
 		webSocketSessionClosed = true;
 		etpSessionClosed = true;
 
diff --git a/src/etp/AbstractSession.h b/src/etp/AbstractSession.h
@@ -57,10 +57,7 @@ namespace ETP_NS
 		* If the ETP session is not set up, it returns the nil UUID.
 		*/
 		const boost::uuids::uuid& getIdentifier() {
-			if (isEtpSessionClosed()) {
-				identifier = boost::uuids::nil_uuid();
-			}
-
+			std::lock_guard<std::mutex> lock(identifierMutex);
 			return identifier;
 		}
 
@@ -246,14 +243,14 @@ namespace ETP_NS
 		*/
 		template<typename T> int64_t sendWithSpecificHandlerAndBlock(const T& mb, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler, int64_t correlationId = 0, int32_t messageFlags = 0)
 		{
-			int64_t msgId = sendWithSpecificHandler(mb, specificHandler, correlationId, messageFlags);
+			const int64_t msgId = sendWithSpecificHandler(mb, specificHandler, correlationId, messageFlags);
 			// The correlationId of the first message MUST be set to 0 and the correlationId of all successive
 			// messages in the same multipart request or notification MUST be set to the messageId of the first
 			// message of the multipart request or notification.
 			// If the request message is itself multipart, the correlationId of each message of the multipart
 			// response MUST be set to the messageId of the FIRST message in the multipart request.
 
-			auto t_start = std::chrono::high_resolution_clock::now();
+			const auto t_start = std::chrono::high_resolution_clock::now();
 			while (isMessageStillProcessing(correlationId == 0 ? msgId : correlationId)) {
 				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));
@@ -633,7 +630,8 @@ namespace ETP_NS
 		/// The next available message id.
 		std::atomic<int64_t> messageId;
 		/// The identifier of the session
-		boost::uuids::uuid identifier;
+		boost::uuids::uuid identifier{ boost::uuids::nil_uuid() };
+		std::mutex identifierMutex;
 		/// Indicates that the endpoint request to close the websocket session 
 		bool isCloseRequested_{ false };
 		size_t reconnectionTryCount_ = 0;
@@ -762,5 +760,7 @@ namespace ETP_NS
 
 			protocolHandlers[protocolId] = coreHandlers;
 		}
+
+		friend void CoreHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::MessageHeader& mh, avro::DecoderPtr d);
 	};
 }
diff --git a/src/etp/ClientSession.h b/src/etp/ClientSession.h
@@ -85,14 +85,10 @@ namespace ETP_NS
 				std::cerr << "on_resolve : " << ec.message() << std::endl;
 			}
 
-			// Reality check: IPv6 is unlikely to be available yet
-			endpoints = std::vector<tcp::endpoint>(results.begin(), results.end());
-			std::stable_partition(endpoints.begin(), endpoints.end(), [](auto entry) {return entry.protocol() == tcp::v4(); });
-
-			asyncConnect();
+			asyncConnect(results);
 		}
 
-		virtual void asyncConnect() = 0;
+		virtual void asyncConnect(const tcp::resolver::results_type& results) = 0;
 
 		virtual bool isTls() const = 0;
 
@@ -129,7 +125,6 @@ namespace ETP_NS
 		std::string proxyAuthorization;
 		std::map<std::string, std::string> additionalHandshakeHeaderFields_;
 		websocket::response_type responseType; // In order to check handshake sec_websocket_protocol
-		std::vector<tcp::endpoint> endpoints; // Store the resolved endpoints to prevent calling resolve once again when reconnecting
 		Energistics::Etp::v12::Protocol::Core::RequestSession requestSession;
 
 		/**
diff --git a/src/etp/PlainClientSession.h b/src/etp/PlainClientSession.h
@@ -35,33 +35,61 @@ namespace ETP_NS
 		virtual ~PlainClientSession() = default;
 
 		// Called by the base class
+#if BOOST_VERSION < 107000
 		FETPAPI_DLL_IMPORT_OR_EXPORT std::unique_ptr<websocket::stream<tcp::socket>>& ws() { return ws_; }
+#else
+		FETPAPI_DLL_IMPORT_OR_EXPORT std::unique_ptr<websocket::stream<boost::beast::tcp_stream>>& ws() { return ws_; }
+#endif
 
 		bool isTls() const final{ return false; }
 
-		void asyncConnect()
+		void asyncConnect(const tcp::resolver::results_type& results)
 		{
-			ws_.reset(new websocket::stream<tcp::socket>(ioc));
-			ws_->binary(true);
+			// Reality check: IPv6 is unlikely to be available yet
+			std::vector<tcp::endpoint> endpoints = std::vector<tcp::endpoint>(results.begin(), results.end());
+			std::stable_partition(endpoints.begin(), endpoints.end(), [](auto entry) {return entry.protocol() == tcp::v4(); });
+
 #if BOOST_VERSION < 107000
+			ws_.reset(new websocket::stream<tcp::socket>(ioc));
 			ws_->write_buffer_size(frameSize_);
-#else
-			ws_->write_buffer_bytes(frameSize_);
-#endif
+			ws_->binary(true);
 
 			// Make the connection on the IP address we get from a lookup
 			boost::asio::async_connect(
 				ws_->next_layer(),
 				endpoints.begin(),
 				endpoints.end(),
 				std::bind(
-					&AbstractClientSessionCRTP::on_connect,
+					&AbstractClientSessionCRTP::on_ssl_handshake,
 					std::static_pointer_cast<AbstractClientSessionCRTP>(shared_from_this()),
 					std::placeholders::_1));
+#else
+			ws_.reset(new websocket::stream<boost::beast::tcp_stream>(ioc));
+			ws_->write_buffer_bytes(frameSize_);
+			ws_->binary(true);
+
+			// Make the connection on the IP address we get from a lookup
+			boost::beast::get_lowest_layer(*ws_).async_connect(
+				endpoints,
+				boost::beast::bind_front_handler(
+					&PlainClientSession::on_connect,
+					std::static_pointer_cast<PlainClientSession>(shared_from_this())));
+#endif
+		}
+
+#if BOOST_VERSION > 106900
+		void on_connect(boost::beast::error_code ec, tcp::resolver::results_type::endpoint_type)
+		{
+			on_ssl_handshake(ec);
 		}
+#endif
 
 	private:
+#if BOOST_VERSION < 107000
 		std::unique_ptr<websocket::stream<tcp::socket>> ws_;
+#else
+		std::unique_ptr<websocket::stream<boost::beast::tcp_stream>> ws_;
+#endif
 		std::size_t frameSize_;
 	};
 }
diff --git a/src/etp/ProtocolHandlers/CoreHandlers.cpp b/src/etp/ProtocolHandlers/CoreHandlers.cpp
@@ -54,6 +54,10 @@ void CoreHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::Mes
 		}
 
 		session->setEtpSessionClosed(false);
+		{
+			std::lock_guard<std::mutex> lock(session->identifierMutex);
+			std::copy(os.sessionId.array.begin(), os.sessionId.array.end(), session->identifier.begin());
+		}
 		on_OpenSession(os, mh.correlationId);
 	}
 	else if (mh.messageType == Energistics::Etp::v12::Protocol::Core::CloseSession::messageTypeId) {
@@ -104,7 +108,7 @@ void CoreHandlers::on_RequestSession(const Energistics::Etp::v12::Protocol::Core
 
 void CoreHandlers::on_OpenSession(const Energistics::Etp::v12::Protocol::Core::OpenSession &, int64_t)
 {
-	session->fesapi_log("The session has been opened with the default core protocol handlers.");
+	session->fesapi_log("The session", session->getIdentifier(), "has been opened with the default core protocol handlers.");
 }
 
 void CoreHandlers::on_CloseSession(const Energistics::Etp::v12::Protocol::Core::CloseSession &, int64_t)
diff --git a/src/etp/ssl/SslClientSession.h b/src/etp/ssl/SslClientSession.h
@@ -25,10 +25,13 @@ under the License.
 #include "ssl_stream.h"
 #elif BOOST_VERSION < 107000
 #include <boost/beast/experimental/core/ssl_stream.hpp>
-#else
+#elif BOOST_VERSION < 108600
 #include <boost/beast/http.hpp>
 #include <boost/beast/ssl.hpp>
 #include <boost/beast/websocket/ssl.hpp>
+#else
+#include <boost/asio/ssl.hpp>
+#include <boost/beast/websocket/ssl.hpp>
 #endif
 
 namespace http = boost::beast::http;    // from <boost/beast/http.hpp>
@@ -48,13 +51,26 @@ namespace ETP_NS
 		virtual ~SslClientSession() {}
 
 		// Called by the base class
+#if BOOST_VERSION < 107000
 		FETPAPI_DLL_IMPORT_OR_EXPORT std::unique_ptr<websocket::stream<boost::beast::ssl_stream<tcp::socket>>>& ws() { return ws_; }
+#elif BOOST_VERSION < 108600
+		FETPAPI_DLL_IMPORT_OR_EXPORT std::unique_ptr<websocket::stream<boost::beast::ssl_stream<boost::beast::tcp_stream>>>& ws() { return ws_; }
+#else
+		FETPAPI_DLL_IMPORT_OR_EXPORT std::unique_ptr< websocket::stream<boost::asio::ssl::stream<boost::beast::tcp_stream>>>& ws() { return ws_; }
+#endif
 
 		bool isTls() const final { return true; }
 
-		void asyncConnect()
+		void asyncConnect(const tcp::resolver::results_type& results)
 		{
+#if BOOST_VERSION < 107000
 			ws_.reset(new websocket::stream<boost::beast::ssl_stream<tcp::socket>>(ioc, sslContext_));
+#elif BOOST_VERSION < 108600
+			ws_.reset(new websocket::stream<boost::beast::ssl_stream<boost::beast::tcp_stream>>(ioc, sslContext_));
+#else
+			ws_.reset(new websocket::stream<boost::asio::ssl::stream<boost::beast::tcp_stream>>(ioc, sslContext_));
+#endif
+
 			ws_->binary(true);
 #if BOOST_VERSION < 107000
 			ws_->write_buffer_size(frameSize_);
@@ -69,7 +85,12 @@ namespace ETP_NS
 				std::cerr << "Websocket on connect (SNI): " << ecSNI.message() << std::endl;
 			}
 
+			// Reality check: IPv6 is unlikely to be available yet
+			std::vector<tcp::endpoint> endpoints = std::vector<tcp::endpoint>(results.begin(), results.end());
+			std::stable_partition(endpoints.begin(), endpoints.end(), [](auto entry) {return entry.protocol() == tcp::v4(); });
+
 			// Make the connection on the IP address we get from a lookup
+#if BOOST_VERSION < 107000
 			boost::asio::async_connect(
 				ws_->next_layer().next_layer(),
 				endpoints.begin(),
@@ -78,9 +99,20 @@ namespace ETP_NS
 					&SslClientSession::on_ssl_connect,
 					std::static_pointer_cast<SslClientSession>(shared_from_this()),
 					std::placeholders::_1));
+#else
+			boost::beast::get_lowest_layer(*ws_).async_connect(
+				endpoints,
+				boost::beast::bind_front_handler(
+					&SslClientSession::on_ssl_connect,
+					std::static_pointer_cast<SslClientSession>(shared_from_this())));
+#endif
 		}
 
+#if BOOST_VERSION < 107000
 		void on_ssl_connect(boost::system::error_code ec) {
+#else
+		void on_ssl_connect(boost::beast::error_code ec, tcp::resolver::results_type::endpoint_type) {
+#endif
 			if (ec) {
 				std::cerr << "on_ssl_connect : " << ec.message() << std::endl;
 			}
@@ -110,7 +142,7 @@ namespace ETP_NS
 				ws_->next_layer().async_handshake(
 					boost::asio::ssl::stream_base::client,
 					std::bind(
-						&AbstractClientSessionCRTP::on_connect,
+						&AbstractClientSessionCRTP::on_ssl_handshake,
 						std::static_pointer_cast<AbstractClientSessionCRTP>(shared_from_this()),
 						std::placeholders::_1));
 			}
@@ -166,14 +198,20 @@ namespace ETP_NS
 			ws_->next_layer().async_handshake(
 				boost::asio::ssl::stream_base::client,
 				std::bind(
-					&AbstractClientSessionCRTP::on_connect,
+					&AbstractClientSessionCRTP::on_ssl_handshake,
 					std::static_pointer_cast<AbstractClientSessionCRTP>(shared_from_this()),
 					std::placeholders::_1));
 		}
 
 	private:
 		boost::asio::ssl::context sslContext_;
+#if BOOST_VERSION < 107000
 		std::unique_ptr<websocket::stream<boost::beast::ssl_stream<tcp::socket>>> ws_;
+#elif BOOST_VERSION < 108600
+		std::unique_ptr<websocket::stream<boost::beast::ssl_stream<boost::beast::tcp_stream>>> ws_;
+#else
+		std::unique_ptr<websocket::stream<boost::asio::ssl::stream<boost::beast::tcp_stream>>> ws_;
+#endif
 		http::request<http::empty_body> proxyHandshake;
 		http::response<http::empty_body> proxyHandshakeResponse;
 		// use own response parser

Original file line number	Diff line number	Diff line change
`@@ -54,6 +54,10 @@ void CoreHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::Mes`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`session->setEtpSessionClosed(false);`
	`57`	`+ {`
	`58`	`+ std::lock_guard<std::mutex> lock(session->identifierMutex);`
	`59`	`+ std::copy(os.sessionId.array.begin(), os.sessionId.array.end(), session->identifier.begin());`
	`60`	`+ }`
`57`	`61`	`on_OpenSession(os, mh.correlationId);`
`58`	`62`	`}`
`59`	`63`	`else if (mh.messageType == Energistics::Etp::v12::Protocol::Core::CloseSession::messageTypeId) {`
`@@ -104,7 +108,7 @@ void CoreHandlers::on_RequestSession(const Energistics::Etp::v12::Protocol::Core`
`104`	`108`
`105`	`109`	`void CoreHandlers::on_OpenSession(const Energistics::Etp::v12::Protocol::Core::OpenSession &, int64_t)`
`106`	`110`	`{`
`107`		`- session->fesapi_log("The session has been opened with the default core protocol handlers.");`
	`111`	`+ session->fesapi_log("The session", session->getIdentifier(), "has been opened with the default core protocol handlers.");`
`108`	`112`	`}`
`109`	`113`
`110`	`114`	`void CoreHandlers::on_CloseSession(const Energistics::Etp::v12::Protocol::Core::CloseSession &, int64_t)`