merge changes from develop

platform/OSP/Web/bundle/bundle.properties

@@ -1,6 +1,6 @@
 # Bundle properties for the Web bundle.
 #
-# Copyright (c) 2007-2021 by Applied Informatics.
+# Copyright (c) 2007-2022 by Applied Informatics.
 # All rights reserved.
 
 # The name of the auth service to use.

platform/OSP/WebEvent/include/Poco/OSP/WebEvent/WebEventRequestHandler.h

@@ -48,6 +48,9 @@ class OSPWebEvent_API WebEventRequestHandler: public Poco::Net::HTTPRequestHandl
 		/// as the SUBSCRIBE message.
 
 private:
+	bool authenticate() const;
+	bool authorize() const;
+
 	WebEventServiceImpl::Ptr _pWebEventServiceImpl;
 	Poco::OSP::BundleContext::Ptr _pContext;
 };

platform/OSP/WebEvent/include/Poco/OSP/WebEvent/WebEventService.h

@@ -114,6 +114,18 @@ class OSPWebEvent_API WebEventService: public Poco::OSP::Service
 		/// to anything other than 200 (OK), the WebSocket connection will not
 		/// be established.
 
+	Poco::BasicEvent<const std::string> subscriptionRequestReceived;
+		/// Fired when a subscription request for a new subject arrives.
+		/// This event is fired once per subject subscription on a
+		/// socket. To find out the number of clients subscribed to a subject,
+		/// use the subscriberCount() call.
+
+	Poco::BasicEvent<const std::string> unsubscriptionRequestReceived;
+		/// Fired when an unsubscription request for a subject arrives.
+		/// This event is fired once per unsubscription on a socket.
+		/// To find out the number of clients subscribed to a subject,
+		/// use the subscriberCount() call.
+
 	Poco::BasicEvent<const NotificationEvent> notificationSent;
 		/// Fired when a notification has been sent, either by calling
 		/// notify(), or by a client sending a NOTIFY message to
@@ -137,6 +149,9 @@ class OSPWebEvent_API WebEventService: public Poco::OSP::Service
 		/// delivered to a subscriber due to a network issue, the subscriber will be removed
 		/// and its WebSocket closed.
 
+	virtual int subscriberCount(const std::string& subject) = 0;
+		/// Returns the number of subscribers to the subject.
+
 	// Service
 	const std::type_info& type() const;
 	bool isA(const std::type_info& otherType) const;

platform/OSP/WebEvent/include/Poco/OSP/WebEvent/WebEventServiceImpl.h

@@ -51,6 +51,7 @@ class OSPWebEvent_API WebEventServiceImpl: public WebEventService
 	// WebEventService
 	Poco::BasicEvent<const NotificationEvent>& subjectNotified(const std::string& subject);
 	void notify(const std::string& subjectName, const std::string& data);
+	int subscriberCount(const std::string& subject);
 
 	// Service
 	const std::type_info& type() const;

platform/OSP/WebEvent/src/WebEventRequestHandler.cpp

@@ -40,55 +40,77 @@ WebEventRequestHandler::~WebEventRequestHandler()
 
 void WebEventRequestHandler::handleRequest(Poco::Net::HTTPServerRequest& request, Poco::Net::HTTPServerResponse& response)
 {
-	try
+	if (authenticate() && authorize())
 	{
-		_pWebEventServiceImpl->preflightRequest(request);
-		if (response.getStatus() == Poco::Net::HTTPResponse::HTTP_OK)
+		try
 		{
-			const Poco::Net::HTTPServerRequest* pRequest = &request;
-			_pWebEventServiceImpl->requestReceived(pRequest);
+			_pWebEventServiceImpl->preflightRequest(request);
 			if (response.getStatus() == Poco::Net::HTTPResponse::HTTP_OK)
 			{
-				Poco::SharedPtr<Poco::Net::WebSocket> pWS = new Poco::Net::WebSocket(request, response);
-				_pContext->logger().information(Poco::format("WebSocket connection established with %s.", request.clientAddress().toString()));
-				_pWebEventServiceImpl->addSubscriber(pWS);
+				const Poco::Net::HTTPServerRequest* pRequest = &request;
+				_pWebEventServiceImpl->requestReceived(pRequest);
+				if (response.getStatus() == Poco::Net::HTTPResponse::HTTP_OK)
+				{
+					Poco::SharedPtr<Poco::Net::WebSocket> pWS = new Poco::Net::WebSocket(request, response);
+					_pContext->logger().information(Poco::format("WebSocket connection established with %s.", request.clientAddress().toString()));
+					_pWebEventServiceImpl->addSubscriber(pWS);
+				}
+				else
+				{
+					_pContext->logger().notice(Poco::format("WebSocket connection from %s rejected by delegate (status=%d).",
+						request.clientAddress().toString(),
+						static_cast<int>(response.getStatus())));
+					response.setContentLength(0);
+					response.send();
+				}
 			}
 			else
 			{
-				_pContext->logger().notice(Poco::format("WebSocket connection from %s rejected by delegate (status=%d).", 
+				_pContext->logger().notice(Poco::format("WebSocket connection from %s rejected by preflight (status=%d).",
 					request.clientAddress().toString(),
 					static_cast<int>(response.getStatus())));
 				response.setContentLength(0);
 				response.send();
 			}
 		}
-		else
+		catch (Poco::Net::WebSocketException& exc)
 		{
-			_pContext->logger().notice(Poco::format("WebSocket connection from %s rejected by preflight (status=%d).", 
-				request.clientAddress().toString(),
-				static_cast<int>(response.getStatus())));
-			response.setContentLength(0);
-			response.send();
+			_pContext->logger().log(exc);
+			switch (exc.code())
+			{
+			case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_UNSUPPORTED_VERSION:
+				response.set("Sec-WebSocket-Version", Poco::Net::WebSocket::WEBSOCKET_VERSION);
+				// fallthrough
+			case Poco::Net::WebSocket::WS_ERR_NO_HANDSHAKE:
+			case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_VERSION:
+			case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_KEY:
+				response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_BAD_REQUEST);
+				response.setContentLength(0);
+				response.send();
+				break;
+			}
 		}
 	}
-	catch (Poco::Net::WebSocketException& exc)
+	else
 	{
-		_pContext->logger().log(exc);
-		switch (exc.code())
-		{
-		case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_UNSUPPORTED_VERSION:
-			response.set("Sec-WebSocket-Version", Poco::Net::WebSocket::WEBSOCKET_VERSION);
-			// fallthrough
-		case Poco::Net::WebSocket::WS_ERR_NO_HANDSHAKE:
-		case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_VERSION:
-		case Poco::Net::WebSocket::WS_ERR_HANDSHAKE_NO_KEY:
-			response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_BAD_REQUEST);
-			response.setContentLength(0);
-			response.send();
-			break;
-		}
+		response.setStatusAndReason(Poco::Net::HTTPResponse::HTTP_UNAUTHORIZED);
+		response.setContentLength(0);
+		response.send();
 	}
 }
 
+bool WebEventRequestHandler::authenticate() const
+{
+	// TODO: actually authenticate
+	return true;
+}
+
+
+bool WebEventRequestHandler::authorize() const
+{
+	// TODO: actually authorize
+	return true;
+}
+
 
-} } } // namespace Poco::OSP::WebEvent
+} } } // namespace Poco::OSP::WebEvent

platform/OSP/WebEvent/src/WebEventServiceImpl.cpp

@@ -411,6 +411,15 @@ void WebEventServiceImpl::unsubscribe(Poco::SharedPtr<Poco::Net::WebSocket> pWS,
 }
 
 
+int WebEventServiceImpl::subscriberCount(const std::string& subject)
+{
+	int cnt = 0;
+	auto er = _subjectMap.equal_range(subject);
+	for (auto it = er.first; it != er.second; ++it) ++cnt;
+	return cnt;
+}
+
+
 void WebEventServiceImpl::send(Poco::SharedPtr<Poco::Net::WebSocket> pWS, const std::string& message)
 {
 	_workerQueue.enqueueNotification(new SendNotification(*this, pWS, message));
@@ -510,6 +519,7 @@ void WebEventServiceImpl::subscribeImpl(Poco::SharedPtr<Poco::Net::WebSocket> pW
 			{
 				it->second->subjectNames.insert(*itSub);
 				_subjectMap.insert(SubjectMap::value_type(*itSub, it->second));
+				subscriptionRequestReceived.notify(this, *itSub);
 			}
 		}
 	}
@@ -542,6 +552,7 @@ void WebEventServiceImpl::unsubscribeImpl(Poco::SharedPtr<Poco::Net::WebSocket>
 					if (itSM->second == it->second)
 					{
 						_subjectMap.erase(itSM);
+						unsubscriptionRequestReceived.notify(this, *itSub);
 						break;
 					}
 					++itSM;
@@ -558,7 +569,7 @@ void WebEventServiceImpl::sendImpl(Poco::SharedPtr<Poco::Net::WebSocket> pWS, co
 	{
 		Poco::FastMutex::ScopedLock lock(_sendMutex);
 
-		pWS->sendFrame(message.data(), message.size());
+		pWS->sendFrame(message.data(), static_cast<int>(message.size()));
 	}
 	catch (Poco::Exception& exc)
 	{
@@ -574,7 +585,7 @@ void WebEventServiceImpl::receiveImpl(Poco::SharedPtr<Poco::Net::WebSocket> pWS)
 	{
 		Poco::Buffer<char> buffer(4096);
 		int flags;
-		int n = pWS->receiveFrame(buffer.begin(), buffer.size(), flags);
+		int n = pWS->receiveFrame(buffer.begin(), static_cast<int>(buffer.size()), flags);
 		_pContext->logger().debug(Poco::format("Frame received (length=%d, flags=0x%x).", n, unsigned(flags)));
 
 		if (flags & Poco::Net::WebSocket::FRAME_OP_PONG)
@@ -678,7 +689,7 @@ void WebEventServiceImpl::shutdownImpl(Poco::SharedPtr<Poco::Net::WebSocket> pWS
 		{
 			Poco::Buffer<char> buffer(4096);
 			int flags;
-			pWS->receiveFrame(buffer.begin(), buffer.size(), flags);
+			pWS->receiveFrame(buffer.begin(), static_cast<int>(buffer.size()), flags);
 			if (flags & Poco::Net::WebSocket::FRAME_OP_CLOSE)
 			{
 				pWS->close();

protocols/MQTT/MQTT.bndlspec

@@ -4,7 +4,7 @@
 		<symbolicName>io.macchina.mqtt.client</symbolicName>
 		<version>2.0.0</version>
 		<vendor>Applied Informatics</vendor>
-		<copyright>(c) 2015-2020, Applied Informatics Software Engineering GmbH</copyright>
+		<copyright>(c) 2015-2022, Applied Informatics Software Engineering GmbH</copyright>
 		<activator>
 			<class>IoT::MQTT::BundleActivator</class>
 			<library>io.macchina.mqtt.client</library>

protocols/Modbus/RTU/src/RTUPort.h

@@ -108,6 +108,9 @@ class RTUPort
 	void reset();
 		/// Resets the connection to the bus.
 
+	std::string address() const;
+		/// Returns the device name.
+
 protected:
 	Poco::UInt16 updateCRC16(Poco::UInt16 crc, Poco::UInt8 byte);
 	bool checkFrame(std::size_t size);
@@ -145,6 +148,12 @@ inline int RTUPort::maxSimultaneousTransactions() const
 }
 
 
+inline std::string RTUPort::address() const
+{
+       return _pSerialPort->device();
+}
+
+
 } } } // namespace IoT::Modbus::RTU
 
 

protocols/Modbus/TCP/src/TCPPort.cpp

@@ -34,6 +34,7 @@ TCPPort::TCPPort(const Poco::Net::SocketAddress& serverAddress):
 	try
 	{
 		connect();
+		_socket.setNoDelay(true);
 	}
 	catch (...)
 	{

protocols/Modbus/TCP/src/TCPPort.h

@@ -146,6 +146,9 @@ class TCPPort
 	void reset();
 		/// Closes and re-opens the connection.
 
+	std::string address() const;
+		/// Returns server address.
+
 private:
 	enum
 	{
@@ -199,6 +202,12 @@ inline int TCPPort::maxSimultaneousTransactions() const
 }
 
 
+inline std::string TCPPort::address() const
+{
+	return _serverAddress.toString();
+}
+
+
 } } } // namespace IoT::Modbus::TCP
 
 

protocols/Modbus/include/IoT/Modbus/IModbusMaster.h

@@ -66,6 +66,9 @@ class IModbusMaster: public Poco::OSP::Service
 	virtual ~IModbusMaster();
 		/// Destroys the IModbusMaster.
 
+	virtual std::string address() const = 0;
+		/// ReturnsModbus master address as a string.
+
 	bool isA(const std::type_info& otherType) const;
 		/// Returns true if the class is a subclass of the class given by otherType.
 
@@ -111,7 +114,7 @@ class IModbusMaster: public Poco::OSP::Service
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
 		/// Throws a ModbusException if the device responds with an exception message.
 
-	virtual std::vector < Poco::UInt16 > readWriteMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 writeStartingAddress, std::vector < Poco::UInt16 > writeValues, Poco::UInt16 readStartingAddress, Poco::UInt8 nOfReadRegisters) = 0;
+	virtual std::vector < Poco::UInt16 > readWriteMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 writeStartingAddress, const std::vector < Poco::UInt16 >& writeValues, Poco::UInt16 readStartingAddress, Poco::UInt8 nOfReadRegisters) = 0;
 		/// Sends a Read/Write Multiple registers request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
@@ -245,13 +248,13 @@ class IModbusMaster: public Poco::OSP::Service
 	const std::type_info& type() const;
 		/// Returns the type information for the object's class.
 
-	virtual void writeMultipleCoils(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, std::vector < bool > values) = 0;
+	virtual void writeMultipleCoils(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, const std::vector < bool >& values) = 0;
 		/// Sends a Write Multiple Coils request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
 		/// Throws a ModbusException if the device responds with an exception message.
 
-	virtual void writeMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, std::vector < Poco::UInt16 > values) = 0;
+	virtual void writeMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, const std::vector < Poco::UInt16 >& values) = 0;
 		/// Sends a Write Multiple Registers request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.

protocols/Modbus/include/IoT/Modbus/ModbusMaster.h

@@ -604,6 +604,9 @@ class IoTModbus_API ModbusMaster
 	virtual ~ModbusMaster();
 		/// Destroys the ModbusMaster.
 
+	virtual std::string address() const = 0;
+		/// ReturnsModbus master address as a string.
+
 	virtual Poco::UInt16 sendRequest(const GenericMessage& message) = 0;
 		/// Sends a generic Modbus message.
 		///
@@ -756,13 +759,13 @@ class IoTModbus_API ModbusMaster
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
 		/// Throws a ModbusException if the device responds with an exception message.
 
-	virtual void writeMultipleCoils(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, std::vector<bool> values) = 0;
+	virtual void writeMultipleCoils(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, const std::vector<bool>& values) = 0;
 		/// Sends a Write Multiple Coils request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
 		/// Throws a ModbusException if the device responds with an exception message.
 
-	virtual void writeMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, std::vector<Poco::UInt16> values) = 0;
+	virtual void writeMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 outputAddress, const std::vector<Poco::UInt16>& values) = 0;
 		/// Sends a Write Multiple Registers request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
@@ -774,7 +777,8 @@ class IoTModbus_API ModbusMaster
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.
 		/// Throws a ModbusException if the device responds with an exception message.
 
-	virtual std::vector<Poco::UInt16> readWriteMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 writeStartingAddress, std::vector<Poco::UInt16> writeValues, Poco::UInt16 readStartingAddress, Poco::UInt8 nOfReadRegisters) = 0;
+	virtual std::vector<Poco::UInt16> readWriteMultipleRegisters(Poco::UInt8 slaveAddress, Poco::UInt16 writeStartingAddress,
+		const std::vector<Poco::UInt16>& writeValues, Poco::UInt16 readStartingAddress, Poco::UInt8 nOfReadRegisters) = 0;
 		/// Sends a Read/Write Multiple registers request to the device and waits for the response.
 		///
 		/// Throws a Poco::TimeoutException if the device does not respond within the specified timeout.