Fix infinite loop when a disconnection occurs during DataArray reading.

Author: philippeVerney

cmake/swigEtp1_2Include.i.in

@@ -1639,17 +1639,7 @@ namespace ETP_NS
 		void setDataspaceProtocolHandlers(std::shared_ptr<DataspaceHandlers> dataspaceHandlers);
 		void setDataspaceOSDUProtocolHandlers(std::shared_ptr<DataspaceOSDUHandlers> dataspaceOSDUHandlers);
 
-		template<typename T> int64_t sendWithSpecificHandler(const T & mb, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler, int64_t correlationId = 0, int32_t messageFlags = 0)
-		{
-			int64_t msgId = encode(mb, correlationId, messageFlags); // put the message to write in the queue
-
-			if (sendingQueue.size() == 1) {
-				do_write();
-			}
-			specificProtocolHandlers[msgId] = specificHandler;
-
-			return msgId;
-		}		
+		template<typename T> int64_t sendWithSpecificHandler(const T & mb, std::shared_ptr<ETP_NS::ProtocolHandlers> specificHandler, int64_t correlationId = 0, int32_t messageFlags = 0) {}		
 		%template(sendWithSpecificHandler) sendWithSpecificHandler<Energistics::Etp::v12::Protocol::Core::RequestSession>;
 		%template(sendWithSpecificHandler) sendWithSpecificHandler<Energistics::Etp::v12::Protocol::Core::OpenSession>;
 		%template(sendWithSpecificHandler) sendWithSpecificHandler<Energistics::Etp::v12::Protocol::Core::CloseSession>;
@@ -1724,15 +1714,7 @@ namespace ETP_NS
 		%template(sendWithSpecificHandler) sendWithSpecificHandler<Energistics::Etp::v12::Protocol::StoreNotification::SubscribeNotifications>;
 		%template(sendWithSpecificHandler) sendWithSpecificHandler<Energistics::Etp::v12::Protocol::StoreNotification::UnsolicitedStoreNotifications>;
 
-		template<typename T> int64_t send(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
-		{
-			if (protocolHandlers.size() > mb.protocolId) {
-				return sendWithSpecificHandler(mb, protocolHandlers[mb.protocolId], correlationId, messageFlags);
-			}
-			else {
-				throw std::logic_error("The agent has no registered handler at all for the protocol " + std::to_string(mb.protocolId));
-			}
-		}
+		template<typename T> int64_t send(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0) {}
 		%template(send) send<Energistics::Etp::v12::Protocol::Core::RequestSession>;
 		%template(send) send<Energistics::Etp::v12::Protocol::Core::OpenSession>;
 		%template(send) send<Energistics::Etp::v12::Protocol::Core::CloseSession>;
@@ -1807,12 +1789,7 @@ namespace ETP_NS
 		%template(send) send<Energistics::Etp::v12::Protocol::StoreNotification::SubscribeNotifications>;
 		%template(send) send<Energistics::Etp::v12::Protocol::StoreNotification::UnsolicitedStoreNotifications>;
 
-		template<typename T> void sendAndBlock(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
-		{
-			int64_t msgId = send(mb, correlationId, messageFlags);
-			while (isMessageStillProcessing(msgId)) {}
-		}
-		
+		template<typename T> int64_t sendAndBlock(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0) {}
 		%template(sendAndBlock) sendAndBlock<Energistics::Etp::v12::Protocol::Core::RequestSession>;
 		%template(sendAndBlock) sendAndBlock<Energistics::Etp::v12::Protocol::Core::OpenSession>;
 		%template(sendAndBlock) sendAndBlock<Energistics::Etp::v12::Protocol::Core::CloseSession>;

src/etp/AbstractSession.h

@@ -169,7 +169,7 @@ namespace ETP_NS
 		* @param messageFlags	The message flags to be sent within the header
 		* @return The ID of the message that has been put in the sending queue.
 		*/
-		template<typename T> void sendAndBlock(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
+		template<typename T> int64_t sendAndBlock(const T & mb, int64_t correlationId = 0, int32_t messageFlags = 0)
 		{
 			int64_t msgId = send(mb, correlationId, messageFlags);
 			// The correlationId of the first message MUST be set to 0 and the correlationId of all successive
@@ -184,6 +184,8 @@ namespace ETP_NS
 					throw std::runtime_error("Time out waiting for a response of message id " + std::to_string(msgId));
 				}
 			}
+
+			return msgId;
 		}
 
 		/**
@@ -223,6 +225,34 @@ namespace ETP_NS
 			return std::get<0>(queueItem);
 		}
 
+		/**
+		* Send a message to the server and register a specific handler for the response and block the thread until the answer of the server has been processed by the handlers
+		* Please look at setTimeOut if you want to set the default timeout value which is 10 000 ms.
+		*
+		* @param mb				The ETP message body to send
+		* @param correlationId	The ID of the message which this message is answering to.
+		* @param messageFlags	The message flags to be sent within the header
+		* @return The ID of the message that has been put in the sending queue.
+		*/
+		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);
+			// 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();
+			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));
+				}
+			}
+
+			return msgId;
+		}
+
 		/**
 		 * Close the web socket session (without sending any ETP message)
 		 */

src/etp/fesapi/FesapiHdfProxy.cpp

@@ -53,21 +53,11 @@ Energistics::Etp::v12::Datatypes::DataArrayTypes::DataArrayMetadata FesapiHdfPro
 	// We don't care about the template parameter in this particular case
 	auto handlers = std::make_shared<GetFullDataArrayHandlers<int64_t>>(session_, nullptr);
 
-	const int64_t msgId = session_->sendWithSpecificHandler(
+	const int64_t msgId = session_->sendWithSpecificHandlerAndBlock(
 		buildGetDataArrayMetadataMessage(datasetName),
 		handlers,
 		0, 0x02);
 
-	// Blocking loop
-	auto t_start = std::chrono::high_resolution_clock::now();
-	// Use timeOut value for session.
-	auto timeOut = session_->getTimeOut();
-	while (session_->isMessageStillProcessing(msgId)) {
-		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 GetDataArrayMetadata message id " + std::to_string(msgId));
-		}
-	}
-
 	return handlers->getDataArrayMetadata();
 }
 

src/etp/fesapi/FesapiHdfProxy.h

@@ -515,13 +515,10 @@ namespace ETP_NS
 			auto specializedHandler = std::make_shared<GetFullDataArrayHandlers<T>>(session_, values);
 			if (wholeSize + (valueCount + 1) * 8 <= maxAllowedDataArraySize) { // There can be valueCount array block and there is the length of the last array block
 				// Get all values at once
-				const int64_t msgId = session_->sendWithSpecificHandler(
+				const int64_t msgId = session_->sendWithSpecificHandlerAndBlock(
 					buildGetDataArraysMessage(datasetName),
 					specializedHandler,
 					0, 0x02);
-
-				// Blocking loop
-				while (session_->isMessageStillProcessing(msgId)) {}
 			}
 			else {
 				// Get all values using several data subarrays allowing more granular streaming
@@ -581,13 +578,10 @@ namespace ETP_NS
 				}
 
 				// Send message
-				const int64_t msgId = session_->sendWithSpecificHandler(
+				const int64_t msgId = session_->sendWithSpecificHandlerAndBlock(
 					msg,
 					specializedHandler,
 					0, 0x02);
-
-				// Blocking loop
-				while (session_->isMessageStillProcessing(msgId)) {}
 			}
 		}
 	};