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

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

Author: philippeVerney

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();

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"
    ]

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;

src/etp/AbstractSession.cpp

@@ -50,6 +50,12 @@ 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;
 		}
+#if BOOST_VERSION > 106900
+		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;
+		}
+#endif
 		else {
 			// This indicates an unexpected error
 			std::cerr << "on_read : error code number " << ec.value() << std::endl;
@@ -63,6 +69,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;
 

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;
 		}
 
@@ -78,11 +75,6 @@ namespace ETP_NS
 			return _timeOut;
 		}
 
-		/**
-		* The list of subscriptions recorded by customers on this session.
-		*/
-		std::unordered_map<int64_t, Energistics::Etp::v12::Datatypes::Object::SubscriptionInfo> subscriptions;
-
 		/**
 		 * Set the Core protocol handlers
 		 */
@@ -246,14 +238,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));
@@ -370,11 +362,6 @@ namespace ETP_NS
 		*/
 		FETPAPI_DLL_IMPORT_OR_EXPORT bool isEtpSessionClosed() const { return webSocketSessionClosed || etpSessionClosed; }
 
-		void setEtpSessionClosed(bool etpSessionClosed_) {
-			etpSessionClosed = etpSessionClosed_;
-			reconnectionTryCount_ = 0;
-		}
-
 		/****************
 		*** DATASPACE ***
 		****************/
@@ -633,7 +620,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;
@@ -646,6 +634,11 @@ namespace ETP_NS
 			receivedBuffer.consume(receivedBuffer.size());
 		}
 
+		void setEtpSessionClosed(bool etpSessionClosed_) {
+			etpSessionClosed = etpSessionClosed_;
+			reconnectionTryCount_ = 0;
+		}
+
 		/**
 		 * Write the current buffer on the web socket
 		 */
@@ -762,5 +755,7 @@ namespace ETP_NS
 
 			protocolHandlers[protocolId] = coreHandlers;
 		}
+
+		friend void CoreHandlers::decodeMessageBody(const Energistics::Etp::v12::Datatypes::MessageHeader& mh, avro::DecoderPtr d);
 	};
 }

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;
 
 		/**

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_;
 	};
 }

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)

src/etp/ProtocolHandlers/StoreNotificationHandlers.cpp

@@ -103,20 +103,7 @@ void StoreNotificationHandlers::on_UnsubscribeNotifications(const Energistics::E
 {
 	session->fesapi_log("on_UnsubscribeNotifications");
 
-	int64_t toRemove = (std::numeric_limits<int64_t>::max)();
-	for (const auto& pair : session->subscriptions) {
-		if (pair.second.requestUuid.array == msg.requestUuid.array) {
-			toRemove = pair.first;
-			break;
-		}
-	}
-
-	if (toRemove != (std::numeric_limits<int64_t>::max)()) {
-		session->subscriptions.erase(toRemove);
-	}
-	else {
-		session->send(ETP_NS::EtpHelpers::buildSingleMessageProtocolException(5, "The subscription request UUID is unknown by the store."), messageId, 0x02);
-	}
+	session->send(ETP_NS::EtpHelpers::buildSingleMessageProtocolException(7, "The StoreHandlers::on_UnsubscribeNotifications method has not been overriden by the agent."), 0x02);
 }
 
 void StoreNotificationHandlers::on_UnsolicitedStoreNotifications(const Energistics::Etp::v12::Protocol::StoreNotification::UnsolicitedStoreNotifications &, int64_t)