.

src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.cpp

@@ -9,8 +9,12 @@
 #include <chrono>
 #include <Poco/Net/HTTPServerRequestImpl.h>
 #include <Poco/Net/HTTPServerResponseImpl.h>
+#include <Poco/Net/WebSocket.h>
+#include "Poco/Net/NetException.h"
 #include <Poco/Net/StreamSocket.h>
 #include <sys/socket.h>
+#include <sys/select.h>
+#include <sys/time.h>
 #include <stropts.h>
 
 
@@ -83,7 +87,10 @@ DeviceRepoHandlerFactory::DeviceRepoHandlerFactory(devicerepository::DeviceRepos
 	} });
 	io_service_thread.reset(new std::thread([this] {
 		try {
+			std::cout << "Boost IO service starting" << std::endl;
+			boost::asio::io_service::work work(io_service);
 			this->io_service.run();
+			std::cout << "Boost IO service stopped" << std::endl;
 		} catch (std::exception& e) {
 			ROS_ERROR_STREAM("IO-Service thread caught exception: " << e.what());
 		} catch(...) {
@@ -127,6 +134,7 @@ void DeviceRepoHandlerFactory::deleteDeviceEventQueue( const std::string& client
 
 void DeviceRepoHandler::_handleRequest(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) {
 	if(req.find("Upgrade") != req.end() && Poco::icompare(req["Upgrade"], "websocket") == 0) {
+		//__handleEvent(req, resp, true);
 		throw webserver::webresource::WebResourceHandlerException("ToDo",
 				Net::HTTPResponse::HTTP_NOT_IMPLEMENTED);
 	} else {
@@ -141,7 +149,7 @@ void DeviceRepoHandler::_handleRequest(Poco::Net::HTTPServerRequest &req, Poco::
 			} else if (method == DeviceAPIMethod::COMMAND || method == DeviceAPIMethod::STATUSUPDATE) {
 				__handleCommand(req, resp, method);
 			} else if (method == DeviceAPIMethod::EVENT) {
-				__handleEventLongPolling(req, resp);
+				__handleEvent(req, resp, false);
 			} else {
 				throw webserver::webresource::WebResourceHandlerException("Could not understood request",
 						Net::HTTPResponse::HTTP_BAD_REQUEST);
@@ -319,8 +327,85 @@ void DeviceRepoHandler::__handleCommand(Poco::Net::HTTPServerRequest &req, Poco:
 }
 
 
+void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue) {
+	Variant msg;
+	bool breakLoop = false;
+	parent->getIOService().post([&]() {
+		Net::HTTPServerRequestImpl* reqImpl = static_cast<Net::HTTPServerRequestImpl*>(&req);
+		// I don't like the content of this loop at all.
+		while (!breakLoop) {
+			// How to find out whether the client disconnected: Monitor the socket with Posix "select":
+			// If select says that there is data, available() returns 0 then apparently the client has disconnected.
+			// WTF, is this really the only way to notice that a client properly disconnected? Internet says so.
+		auto sockfs = reqImpl->socket().impl()->sockfd();
+		fd_set fds;
+		FD_ZERO(&fds);
+		FD_SET(sockfs, &fds);
+			struct timeval timeout = {0, 1000};
+			int sel = select(sockfs + 1, &fds, 0, 0, &timeout);
+			if (sel < 0 || (sel > 0 && reqImpl->socket().available() == 0)) {
+				ROS_ERROR_STREAM("Apparently we have a disconnect in device event queue \"" << eventQueue->clientID << "\"");
+				breakLoop = true;
+				eventQueue->cancelPopEvent();
+			}
+		}
+	});
+	try {
+		msg = eventQueue->popEvent();
+	} catch (DeviceEventQueue::DisconnectedException& e) {
+	}
+	breakLoop = true;
+	handleResponse(req, resp, msg);
+};
 
-void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) {
+void DeviceRepoHandler::__handleEventWebSocket(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue) {
+	try {
+		Net::WebSocket ws(req, resp);
+		std::cout << "WebSocket connection established." << std::endl;
+		bool error = false;
+		/*std::thread revcThread([&]() {
+			char buffer[1024];
+			int flags;
+			int n;
+			do {
+				n = ws.receiveFrame(buffer, sizeof(buffer), flags);
+				std::cout << Poco::format("Frame received (length=%d, flags=0x%x).", n, unsigned(flags)) << std::endl;
+			} while (!error && n > 0 && (flags & Net::WebSocket::FRAME_OP_BITMASK) != Net::WebSocket::FRAME_OP_CLOSE);
+			error= true;
+		});*/
+		char buffer[1024];
+		int flags;
+		int n;
+		auto format = getResponseDataFormat(req);
+		while (!error) {
+			Variant msg;
+			try {
+				msg = eventQueue->popEvent();
+				std::string encoded = encodeVariant(msg, req, format);
+				ws.sendFrame(encoded.c_str(), encoded.size(), Net::WebSocket::FRAME_FLAG_FIN | Net::WebSocket::FRAME_OP_TEXT);
+			} catch (DeviceEventQueue::DisconnectedException& e) {
+				error = true;
+			}
+		}
+		std::cout << "WebSocket connection closed." << std::endl;
+	} catch (Net::WebSocketException& exc) {
+		std::cout << "WebSocketException caught: " << exc.what() << std::endl;
+		switch (exc.code()) {
+			case Net::WebSocket::WS_ERR_HANDSHAKE_UNSUPPORTED_VERSION:
+			resp.set("Sec-WebSocket-Version", Net::WebSocket::WEBSOCKET_VERSION);
+			// fallthrough
+			case Net::WebSocket::WS_ERR_NO_HANDSHAKE:
+			case Net::WebSocket::WS_ERR_HANDSHAKE_NO_VERSION:
+			case Net::WebSocket::WS_ERR_HANDSHAKE_NO_KEY:
+			resp.setStatusAndReason(Net::HTTPResponse::HTTP_BAD_REQUEST);
+			resp.setContentLength(0);
+			resp.send();
+			break;
+		}
+	}
+};
+
+void DeviceRepoHandler::__handleEvent(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, bool websocket) {
 	std::vector<std::string> pathPieces, matchPieces;
 	matchUri.getPathSegments(matchPieces);
 	Poco::URI uri(req.getURI());
@@ -399,15 +484,21 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 		}
 		// Get event queue
 		auto eventQueue = parent->getDeviceEventQueue(clientID);
-		eventQueue->setHandler(this, &req, &resp);
+		if (!eventQueue->setHandler(this, &req, &resp)) {
+			throw webserver::webresource::WebResourceHandlerException((boost::format("Client id \"%1%\" already in use.") % devid).str(),
+					Poco::Net::HTTPResponse::HTTPStatus::HTTP_BAD_REQUEST);
+		}
 		std::cout << "Got EventQueue \"" << eventQueue->clientID << "\"" << std::endl;
 		devicerepository::Device* devptr = nullptr;
 		if (singleDeviceMode) {
 			devptr = device.lock().get();
 		}
 		eventQueue->checkRegistration(event, devptr, path.str());
-		auto msg = eventQueue->popEvent();
+		if (websocket) {
-		handleResponse(req, resp, msg);
+			__handleEventWebSocket(req, resp, eventQueue.get());
+		} else {
+			__handleEventLongPolling(req, resp, eventQueue.get());
+		}
 		eventQueue->unsetHandler();
 	} else {
 		throw webserver::webresource::WebResourceHandlerException((boost::format("Could not find device \"%1%\"") % devid).str(),
@@ -418,6 +509,7 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 
 
 
+
 DeviceEventQueue::DeviceEventQueue( 	const std::string& clientID,
 					DeviceRepositoryService* repoService,
 					DeviceRepoHandlerFactory* handlerFactory )
@@ -472,38 +564,43 @@ void DeviceEventQueue::pushEvent(const libvariant::Variant& event) {
 libvariant::Variant DeviceEventQueue::popEvent() {
     std::unique_lock<std::mutex> lock(_mutex);
 	libvariant::Variant retval;
+	breakPopLoop = false;
 	Net::HTTPServerRequestImpl* reqImpl = static_cast<Net::HTTPServerRequestImpl*>(req);
-	while (_eventQueue.empty()) {
+	// I don't like the content of this loop at all.
+	while (_eventQueue.empty() && !breakPopLoop) {
 		_cv.wait_for(lock, std::chrono::seconds(1));
-		/*
-		auto addr = reqImpl->socket().peerAddress();
-		int error_code;
-		socklen_t error_code_size = sizeof(error_code);
-		auto error = getsockopt(reqImpl->socket().impl()->sockfd(), SOL_SOCKET, SO_ERROR, &error_code, &error_code_size);
-		int tmp = FLUSHRW;
-		try {
-			reqImpl->socket().impl()->ioctl(I_GWROPT, tmp);
-		} catch (std::exception& e) {
-			std::cout << "Debug01 Exception: " << e.what() << std::endl;
-		}
-		std::cout << "debug01: " << error << " - " << addr.toString() << " - " << reqImpl->socket().impl()->available() << " - " << errno << std::endl;
-		*/
-		std::cout << "debug01: " << reqImpl->socket().impl()->available() << " - " << errno << std::endl;
 
+		// Monitor the socket for a very small time with Posix "select".
+		// If select says that there is data, available() returns 0 then apparently the client has disconnected.
+		// WTF, is this really the only way to notice that a client properly disconnected? Internet says so.
+		/*fd_set fds;
+		auto sockfs = reqImpl->socket().impl()->sockfd();
+		FD_ZERO(&fds);
+		FD_SET(sockfs, &fds);
+		struct timeval timeout = {0, 1};
+		int sel = select(sockfs + 1, &fds, 0, 0, &timeout);
+		if (sel > 0 && reqImpl->socket().available() == 0) {
+			ROS_ERROR_STREAM("Apparently we have a disconnect in device event queue \"" << clientID << "\"");
+			throw DisconnectedException();
+		}*/
 	}
-	if (!_eventQueue.empty()) {
+	if (!_eventQueue.empty() && !breakPopLoop) {
 		retval = _eventQueue.front();
 		_eventQueue.pop();
 	}
 	return retval;
 }
 
-void DeviceEventQueue::setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp) {
+bool DeviceEventQueue::setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp) {
-	this->handler = handler;
+	if (openConnectionCount <= 0) {
-	this->req = req;
+		this->handler = handler;
-	this->resp =resp;
+		this->req = req;
-	openConnectionCount++;
+		this->resp =resp;
-	timer.cancel();
+		openConnectionCount++;
+		timer.cancel();
+		return true;
+	}
+	return false;
 }
 
 void DeviceEventQueue::unsetHandler() {

src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.h

@@ -54,35 +54,52 @@ enum class DeviceAPIMethod {
 class DeviceEventQueue
 	: public virtual interface::Observer<devicerepository::DeviceEventType, devicerepository::DeviceEvent> {
 public:
+	class DisconnectedException : public std::exception {};
+
 	const std::string clientID;
 	DeviceEventType eventType = DeviceEventType::ALL_EVENTS;
 	DeviceRepositoryService* const repoService;
 	Device* device = nullptr;
 	std::string pointer;
+
 	bool registeredToEvent = false;
 	size_t openConnectionCount = 0;
+	bool breakPopLoop = false;
+
 	DeviceRepoHandler* handler = nullptr;
 	DeviceRepoHandlerFactory* handlerFactory;
 	Poco::Net::HTTPServerRequest* req = nullptr;
 	Poco::Net::HTTPServerResponse* resp = nullptr;
+
 	std::queue<libvariant::Variant> _eventQueue;
 	std::mutex _mutex;
 	std::condition_variable _cv;
+
 	boost::asio::deadline_timer timer;
+
 	DeviceEventQueue( 	const std::string& clientID,
 						DeviceRepositoryService* repoService,
 						DeviceRepoHandlerFactory* handlerFactory );
+
 	void pushEvent(const libvariant::Variant& event);
 	libvariant::Variant popEvent();
-	void setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp);
+	void cancelPopEvent() {
+		breakPopLoop = true;
+		_cv.notify_all();
+	}
+
+	bool setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp);
 	void unsetHandler();
+
 	void checkRegistration(DeviceEventType eventType, Device* device, const std::string& pointer);
 	void registerEvent(DeviceEventType eventType, Device* device = nullptr);
 	void unregisterEvent();
+
 	~DeviceEventQueue() {
 		unregisterEvent();
 		std::cout << "DeviceEventQueue \"" << clientID << "\" destroyed" << std::endl;
 	}
+
 	virtual void notificationReceived( 	interface::Observer<devicerepository::DeviceEventType, devicerepository::DeviceEvent>::observable_type* o,
 										const devicerepository::DeviceEventType& key,
 										const devicerepository::DeviceEvent* e) noexcept override;
@@ -157,7 +174,11 @@ public:
 
 	void __handleCommand(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceAPIMethod method);
 
-	void __handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp);
+	void __handleEvent(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, bool websocket);
+
+	void __handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue);
+
+	void __handleEventWebSocket(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp, DeviceEventQueue* eventQueue);
 
 	void sendDeviceCommand( devicerepository::Device* device,
 							Poco::Net::HTTPServerRequest &req,