closeAndBlock now force closing after timeout by default

Add a forceClose(bool) capability: closeAndBlock now accepts an optional forceCloseAfterTimeOut parameter (default true) and will call a new forceClose() when the websocket doesn't close before timeout. Fix #30
Replace usages of std::chrono::high_resolution_clock with std::chrono::steady_clock for timeout/duration calculations to ensure monotonic timing and avoid issues with system clock adjustments.

Author: philippeVerney

README.md

@@ -11,7 +11,7 @@
 # Prepare the dependencies
 Download (build and install if necessary) third party libraries:
 - BOOST : All versions from version 1.66 should be ok but you may experience some [min/max build issues](https://github.com/boostorg/beast/issues/1980) using version 1.72 or 1.73.
-- AVRO : https://avro.apache.org/releases.html#Download (starting from version 1.9.0 [except 1.11.1](https://issues.apache.org/jira/browse/AVRO-3601), build it with the above boost library.)
+- AVRO : https://avro.apache.org/releases.html#Download (starting from version 1.9.0 [except 1.11.1](https://issues.apache.org/jira/browse/AVRO-3601), build it with the above boost library if necessary.)
 - (OPTIONALLY) OpenSSL : version 3.4 is known to work. OpenSSL is mandatory on Unix environement if you want SSL/TLS support. On Windows environment, you can depends on [Wintls](https://github.com/laudrup/boost-wintls/tree/master) instead
 - (OPTIONALLY) [Wintls](https://github.com/laudrup/boost-wintls/tree/master) : Starting from version 0.9.9. Only on Windows and only if you don't want to use OpenSSL to support SSL/TLS. FYI, Wintls uses native Windows API [(SSPI/Schannel)](https://docs.microsoft.com/en-us/windows-server/security/tls/tls-ssl-schannel-ssp-overview) functionality instead of OpenSSL for providing TLS encrypted stream functionality.
 - (OPTIONALLY) [FESAPI](https://github.com/F2I-Consulting/fesapi/releases) : All versions from version 2.7.0.0 should be ok but a minimal version of 2.11.0.0 is recommended to automatically recognize FESAPI CMake Variables using CMake find Module and build silently the EtpClient example.

cmake/swigEtp1_2Include.i.in

@@ -1734,8 +1734,12 @@ namespace ETP_NS
 		* Send a message for closing to the server after having sent all previous messages.
 		* The session would really close only after all messages have been sent and responded.
 		* This method does block.
+		* 
+		* @param	forceCloseAfterTimeOut	If true, the websocket session will be closed
+		*									even if the ETP server did not close the session before the time out.
+		*									By default, it is true.
 		*/
-		void closeAndBlock();
+		void closeAndBlock(bool forceCloseAfterTimeOut = true);
 
 		/**
 		* Check if the ETP session (starting after the Core.OpenSession or Core.RequestSession message) is not opened yet or has been closed.

example/withFesapi/etpClient.cpp

@@ -302,14 +302,14 @@ void askUser(std::shared_ptr<ETP_NS::AbstractSession> session, COMMON_NS::DataOb
 					repo.addOrReplaceGsoapProxy(dataobjectEntry.second.data, ETP_NS::EtpHelpers::getDataObjectType(dataobjectEntry.second.resource.uri), ETP_NS::EtpHelpers::getDataspaceUri(dataobjectEntry.second.resource.uri));
 				}
 				// Parse reps
-				auto global_start = std::chrono::high_resolution_clock::now();
+				auto global_start = std::chrono::steady_clock::now();
 				for (auto* rep : repo.getDataObjects<RESQML2_NS::AbstractRepresentation>()) {
 					std::cout << "Representation " << rep->getTitle() << std::endl;
 					std::unique_ptr<double[]> ijkGridPoints(new double[rep->getXyzPointCountOfAllPatches() * 3]);
-					auto t_start = std::chrono::high_resolution_clock::now();
+					auto t_start = std::chrono::steady_clock::now();
 					try {
 						rep->getXyzPointsOfAllPatches(ijkGridPoints.get());
-						std::cout << "XYZ POINTS IN  " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() << " ms" << std::endl;
+						std::cout << "XYZ POINTS IN  " << std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() << " ms" << std::endl;
 					}
 					catch (...) {
 						std::cerr << "Error reading XYZ points." << std::endl;
@@ -322,20 +322,20 @@ void askUser(std::shared_ptr<ETP_NS::AbstractSession> session, COMMON_NS::DataOb
 						if (dynamic_cast<RESQML2_NS::ContinuousProperty*>(prop) != nullptr) {
 							std::cout << "Continuous Prop " << propIndex++ << "/" << allProps.size() << " : " << prop->getTitle() << std::endl;
 							std::unique_ptr<double[]> propValues(new double[valuesCount]);
-							t_start = std::chrono::high_resolution_clock::now();
+							t_start = std::chrono::steady_clock::now();
 							prop->getDoubleValuesOfPatch(0, propValues.get());
-							std::cout << "Continuous Prop IN  " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() << " ms" << std::endl;
+							std::cout << "Continuous Prop IN  " << std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() << " ms" << std::endl;
 						}
 						else {
 							std::cout << "Non Continuous Prop " << propIndex++ << "/" << allProps.size() << " : " << prop->getTitle() << std::endl;
 							std::unique_ptr<int[]> propValues(new int[valuesCount]);
-							t_start = std::chrono::high_resolution_clock::now();
+							t_start = std::chrono::steady_clock::now();
 							prop->getInt32ValuesOfPatch(0, propValues.get());
-							std::cout << "Non Continuous Prop IN  " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() << " ms" << std::endl;
+							std::cout << "Non Continuous Prop IN  " << std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() << " ms" << std::endl;
 						}
 					}
 				}
-				std::cout << "GLOBALLY DONE IN  " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - global_start).count() << " ms" << std::endl;
+				std::cout << "GLOBALLY DONE IN  " << std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - global_start).count() << " ms" << std::endl;
 			}
 			else {
 				std::cout << "There is no dataobject in this dataspace" << std::endl;
@@ -546,7 +546,7 @@ void askUser(std::shared_ptr<ETP_NS::AbstractSession> session, COMMON_NS::DataOb
 				std::vector< std::string > dataspacesToLock;
 				dataspacesToLock.push_back(dataspace.uri);
 
-				auto transaction_start = std::chrono::high_resolution_clock::now();
+				auto transaction_start = std::chrono::steady_clock::now();
 
 				std::string result = session->startTransaction(dataspacesToLock);
 				if (result.empty()) {
@@ -562,13 +562,13 @@ void askUser(std::shared_ptr<ETP_NS::AbstractSession> session, COMMON_NS::DataOb
 						0.0, 1.0, 0.0, 50.0);
 
 					for (size_t propIndex = 0; propIndex < 100; ++propIndex) {
-						auto t_start = std::chrono::high_resolution_clock::now();
+						auto t_start = std::chrono::steady_clock::now();
 						auto* prop = tmpRepo.createContinuousProperty(horizon_grid_2d_representation, "", "", 1, gsoap_eml2_3::eml23__IndexableElement::nodes, gsoap_resqml2_0_1::resqml20__ResqmlUom::m,
 							gsoap_resqml2_0_1::resqml20__ResqmlPropertyKind::length);
 						std::unique_ptr<double[]> prop_values(new double[ni * nj]);
 						prop->pushBackDoubleHdf5Array2dOfValues(prop_values.get(), ni, nj, hdf_proxy);
-						std::cout << " Pushed prop " << propIndex << " in " << std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count() << " ms" << std::endl;
-						std::cout << " Global time  " << std::chrono::duration<double>(std::chrono::high_resolution_clock::now() - transaction_start).count() << " s" << std::endl;
+						std::cout << " Pushed prop " << propIndex << " in " << std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() << " ms" << std::endl;
+						std::cout << " Global time  " << std::chrono::duration<double>(std::chrono::steady_clock::now() - transaction_start).count() << " s" << std::endl;
 					}
 
 					tmpRepo.setUriSource(dataspace.uri);
@@ -706,7 +706,7 @@ void askUser(std::shared_ptr<ETP_NS::AbstractSession> session, COMMON_NS::DataOb
 	for (auto* hdfProxy : repo.getHdfProxySet()) {
 		hdfProxy->close();
 	}
-	session->close();
+	session->closeAndBlock();
 }
 
 int main(int argc, char **argv)
@@ -750,10 +750,10 @@ int main(int argc, char **argv)
 	std::thread sessionThread(&ETP_NS::ClientSession::run, clientSession);
 
 	// Wait for the ETP session to be opened
-	auto t_start = std::chrono::high_resolution_clock::now();
+	auto t_start = std::chrono::steady_clock::now();
 	while (clientSession->isEtpSessionClosed()) {
-		auto timeOut = std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - t_start).count();
-		if (timeOut > 50000) {
+		auto timeOut = std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count();
+		if (timeOut > 10000) {
 			throw std::invalid_argument("Time out : " + std::to_string(timeOut) + " ms.\n");
 		}
 	}

src/fetpapi/etp/AbstractSession.h

@@ -187,10 +187,10 @@ namespace ETP_NS
 			// 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();
+			auto t_start = std::chrono::steady_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) {
+				if (std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() > _timeOut) {
 					throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
 				}
 			}
@@ -285,10 +285,10 @@ namespace ETP_NS
 			// 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.
 
-			const auto t_start = std::chrono::high_resolution_clock::now();
+			const auto t_start = std::chrono::steady_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) {
+				if (std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() > _timeOut) {
 					throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
 				}
 			}
@@ -385,14 +385,24 @@ namespace ETP_NS
 		* Send a message for closing to the server after having sent all previous messages.
 		* The session would really close only after all messages have been sent and responded.
 		* This method does block.
+		* 
+		* @param	forceCloseAfterTimeOut	If true, the websocket session will be closed
+		*									even if the ETP server did not close the session before the time out.
+		*									By default, it is true.
 		*/
-		FETPAPI_DLL_IMPORT_OR_EXPORT void closeAndBlock() {
+		FETPAPI_DLL_IMPORT_OR_EXPORT void closeAndBlock(bool forceCloseAfterTimeOut = true) {
 			close();
-			auto t_start = std::chrono::high_resolution_clock::now();
+			auto t_start = std::chrono::steady_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");
+				if (std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t_start).count() > _timeOut) {
+					if (forceCloseAfterTimeOut) {
+						fesapi_log("Time out waiting for closing. Forcefully closing.");
+						forceClose();
+					}
+					else {
+						throw std::runtime_error("Time out waiting for closing");
+					}
 				}
 			}
 		}
@@ -691,6 +701,11 @@ namespace ETP_NS
 		 */
 		virtual void do_write() = 0;
 
+		/**
+		 * Force closing of the session
+		 */
+		virtual void forceClose() = 0;
+
 		/**
 		 * Reads the message header currently stored in the decoder.
 		 * @param decoder	Must be initialized with stream containing a coded message header.

src/fetpapi/etp/ClientSession.h

@@ -139,17 +139,16 @@ namespace ETP_NS
 
 		/**
 		 * @param initializationParams  The initialization parameters of the session including IP host, port, requestedProtocols, supportedDataObjects
-		 * @param target				usually "/" but a server can decide to serve etp on a particular target
 		 * @param etpServerAuth			The HTTP authorization attribute to send to the ETP server. It may be empty if not needed.
 		 * @param proxyAuth				The HTTP authorization attribute to send to the proxy server. It may be empty if not needed.
 		 */
 		ClientSession(
-			InitializationParameters const* initializationParams, const std::string& target, const std::string& etpServerAuth, const std::string& proxyAuth = "") :
+			InitializationParameters const* initializationParams, const std::string& etpServerAuth, const std::string& proxyAuth = "") :
 			ioc(),
 			resolver(ioc),
 			etpServerHost(initializationParams->getEtpServerHost()),
 			etpServerPort(std::to_string(initializationParams->getEtpServerPort())),
-			etpServerTarget(target),
+			etpServerTarget(initializationParams->getEtpServerUrlPath()),
 			etpServerAuthorization(etpServerAuth),
 			proxyHost(initializationParams->getProxyHost()),
 			proxyPort(std::to_string(initializationParams->getProxyPort())),

src/fetpapi/etp/ClientSessionLaunchers.cpp

@@ -119,7 +119,7 @@ std::shared_ptr<ETP_NS::ClientSession> ETP_NS::ClientSessionLaunchers::createCli
 
 		std::size_t preferredMaxFrameSize = getNegotiatedMaxWebSocketFramePayloadSize(restClientSession->getResponse().body(), initializationParams->getPreferredMaxFrameSize());
 
-		result = std::make_shared<SslClientSession>(std::move(ctx), initializationParams, "/" + initializationParams->getEtpServerUrlPath(),
+		result = std::make_shared<SslClientSession>(std::move(ctx), initializationParams,
 			authorization, proxyAuthorization,
 			initializationParams->getAdditionalHandshakeHeaderFields(), preferredMaxFrameSize);
 	}
@@ -134,7 +134,7 @@ std::shared_ptr<ETP_NS::ClientSession> ETP_NS::ClientSessionLaunchers::createCli
 
 		std::size_t preferredMaxFrameSize = getNegotiatedMaxWebSocketFramePayloadSize(restClientSession->getResponse().body(), initializationParams->getPreferredMaxFrameSize());
 
-		result = std::make_shared<PlainClientSession>(initializationParams, "/" + initializationParams->getEtpServerUrlPath(),
+		result = std::make_shared<PlainClientSession>(initializationParams,
 			authorization, proxyAuthorization,
 			initializationParams->getAdditionalHandshakeHeaderFields(), preferredMaxFrameSize);
 #if WITH_ETP_SSL

src/fetpapi/etp/InitializationParameters.cpp

@@ -54,7 +54,9 @@ namespace {
 		else {
 			hostEnd = portStart++;
 			portEnd = url.find("/", portStart);
-			uint16_t readPort = static_cast<uint16_t>(stoi(url.substr(portStart, portEnd - portStart)));
+			uint16_t readPort = static_cast<uint16_t>(stoi(portEnd == std::string::npos
+				? url.substr(portStart)
+				: url.substr(portStart, portEnd - portStart)));
 			if (readPort < 1 || readPort > (std::numeric_limits<uint16_t>::max)()) {
 				throw std::out_of_range("The port " + std::to_string(readPort) + " is out of the allowed range for TCP ports (0,2^16)");
 			}
@@ -66,7 +68,7 @@ namespace {
 			std::get<0>(result) = url.substr(hostStart);
 		}
 		else {
-			std::get<2>(result) = portEnd < url.size() - 1 ? url.substr(portEnd + 1) : "";
+			std::get<2>(result) = portEnd == std::string::npos ? "" : url.substr(portEnd);
 			std::get<0>(result) = url.substr(hostStart, hostEnd - hostStart);
 		}
 

src/fetpapi/etp/InitializationParameters.h

@@ -110,12 +110,16 @@ namespace ETP_NS
 		* @param port			The port number to connect to.
 		* @param urlPath		The rest of the locator consists of data specific to the scheme, and is known as the "url-path".
 		*						It supplies the details of how the specified resource can be accessed.
-		*						Note that the "/" between the host (or port) and the url-path is NOT part of the url-path.
+		*						It must start with a slash or be empty.
 		*/
 		InitializationParameters(boost::uuids::uuid instanceUuid,
 			const std::string& host, uint16_t port, const std::string& urlPath = "") :
 			identifier(instanceUuid), etpServerHost(host), etpServerPort(port), etpServerUrlPath(urlPath)
-		{}
+		{
+			if (!etpServerUrlPath.empty() && etpServerUrlPath[0] != '/') {
+				throw std::invalid_argument("urlPath must start with a slash or be empty");
+			}
+		}
 
 		/**
 		* Only to be used for direct connection to the ETP server URL (not whenpassing through a proxy)
@@ -126,14 +130,17 @@ namespace ETP_NS
 		* @param port			The port number to connect to.
 		* @param urlPath		The rest of the locator consists of data specific to the scheme, and is known as the "url-path".
 		*						It supplies the details of how the specified resource can be accessed.
-		*						Note that the "/" between the host (or port) and the url-path is NOT part of the url-path.
+		*						It must start with a slash or be empty.
 		*/
 		InitializationParameters(const std::string & instanceUuid,
 			const std::string& host, uint16_t port, const std::string& urlPath = "") :
 			etpServerHost(host), etpServerPort(port), etpServerUrlPath(urlPath)
 		{
 			std::stringstream ss(instanceUuid);
 			ss >> identifier;
+			if (!etpServerUrlPath.empty() && etpServerUrlPath[0] != '/') {
+				throw std::invalid_argument("urlPath must start with a slash or be empty");
+			}
 		}
 
 		virtual ~InitializationParameters() = default;

src/fetpapi/etp/PlainClientSession.cpp

@@ -22,9 +22,9 @@ under the License.
 using namespace ETP_NS;
 
 PlainClientSession::PlainClientSession(
-	InitializationParameters const* initializationParams, const std::string & target, const std::string & authorization, const std::string& proxyAuthorization,
+	InitializationParameters const* initializationParams,const std::string & authorization, const std::string& proxyAuthorization,
 	const std::map<std::string, std::string>& additionalHandshakeHeaderFields, std::size_t frameSize) :
-		AbstractClientSessionCRTP<PlainClientSession>(initializationParams, target, authorization, proxyAuthorization), frameSize_(frameSize)
+		AbstractClientSessionCRTP<PlainClientSession>(initializationParams, authorization, proxyAuthorization), frameSize_(frameSize)
 {
 	additionalHandshakeHeaderFields_ = additionalHandshakeHeaderFields;
 }

src/fetpapi/etp/PlainClientSession.h

@@ -29,7 +29,7 @@ namespace ETP_NS
 		* @param frameSize				Sets the size of the write buffer used by the implementation to send frames : https://www.boost.org/doc/libs/1_75_0/libs/beast/doc/html/beast/ref/boost__beast__websocket__stream/write_buffer_bytes/overload1.html.
 		*/
 		FETPAPI_DLL_IMPORT_OR_EXPORT PlainClientSession(
-			InitializationParameters const* initializationParams, const std::string & target, const std::string & authorization, const std::string& proxyAuthorization = "",
+			InitializationParameters const* initializationParams, const std::string & authorization, const std::string& proxyAuthorization = "",
 			const std::map<std::string, std::string>& additionalHandshakeHeaderFields = {}, std::size_t frameSize = 4096);
 
 		virtual ~PlainClientSession() = default;
@@ -91,5 +91,13 @@ namespace ETP_NS
 		std::unique_ptr<websocket::stream<boost::beast::tcp_stream>> ws_;
 #endif
 		std::size_t frameSize_;
+
+		/**
+		 * Force closing of the session
+		 */
+		void forceClose() {
+			ws_->next_layer().close();
+			webSocketSessionClosed = true;
+		}
 	};
 }