.

src/hos_cosgi_devicerepository/include/hos_cosgi_devicerepository/hos/cosgi/service/devicerepository/DeviceRepositoryService.h

@@ -39,7 +39,8 @@ enum class DeviceEventType : int {
 	STATUS_UPDATE_RECEIVED = 3,
 	COMMAND_RECEIVED = 4,
 	STATE_CHANGED = 5,
-	ALL_EVENTS = 6
+	ALL_EVENTS = 6,
+	CHANGE_EVENTS = 7
 };
 
 

src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/DeviceRepositoryServiceImpl.cpp

@@ -46,6 +46,7 @@ bool DeviceImpl::handleStatusMsg(const Variant& msg, const timestamp_t& stamp) n
 		DeviceStateChangeEvent event(this, msg, stamp);
 		notifyObservers(DeviceEventType::STATE_CHANGED, &event);
 		notifyObservers(DeviceEventType::ALL_EVENTS, DeviceEventType::STATE_CHANGED, &event);
+		notifyObservers(DeviceEventType::CHANGE_EVENTS, DeviceEventType::STATE_CHANGED, &event);
 	}
 
 	return stateChanged;
@@ -177,6 +178,7 @@ void DeviceRepositoryServiceImpl::addDevice(const std::string& name, XmlRpc::Xml
 	DeviceAddedEvent event(dev.get(), std::chrono::system_clock::now());
 	notifyObservers(DeviceEventType::DEVICE_ADDED, &event);
 	notifyObservers(DeviceEventType::ALL_EVENTS, DeviceEventType::DEVICE_ADDED, &event);
+	notifyObservers(DeviceEventType::CHANGE_EVENTS, DeviceEventType::DEVICE_ADDED, &event);
 }
 
 
@@ -187,8 +189,10 @@ void DeviceRepositoryServiceImpl::removeDevice(const std::string& name) {
 		DeviceRemovedEvent event(dev->second.get(), std::chrono::system_clock::now());
 		notifyObservers(DeviceEventType::DEVICE_REMOVED, &event);
 		notifyObservers(DeviceEventType::ALL_EVENTS, DeviceEventType::DEVICE_REMOVED, &event);
+		notifyObservers(DeviceEventType::CHANGE_EVENTS, DeviceEventType::DEVICE_REMOVED, &event);
 		dev->second->notifyObservers(DeviceEventType::DEVICE_REMOVED, &event);
 		dev->second->notifyObservers(DeviceEventType::ALL_EVENTS, DeviceEventType::DEVICE_REMOVED, &event);
+		dev->second->notifyObservers(DeviceEventType::CHANGE_EVENTS, DeviceEventType::DEVICE_REMOVED, &event);
 		devices.erase(name);
 	}
 }
@@ -246,6 +250,7 @@ void DeviceRepositoryServiceImpl::statusMsgCallback(const hos_core_msgs::DeviceS
 					DeviceStateChangeEvent event(dev->second.get(), m, stamp);
 					notifyObservers(DeviceEventType::STATE_CHANGED, &event);
 					notifyObservers(DeviceEventType::ALL_EVENTS, DeviceEventType::STATE_CHANGED, &event);
+					notifyObservers(DeviceEventType::CHANGE_EVENTS, DeviceEventType::STATE_CHANGED, &event);
 				}
 			}
 

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

@@ -6,6 +6,13 @@
 #include <Poco/URI.h>
 #include <Poco/Timestamp.h>
 #include <boost/algorithm/string.hpp>
+#include <chrono>
+#include <Poco/Net/HTTPServerRequestImpl.h>
+#include <Poco/Net/HTTPServerResponseImpl.h>
+#include <Poco/Net/StreamSocket.h>
+#include <sys/socket.h>
+#include <stropts.h>
+
 
 
 // Service Namespace
@@ -31,23 +38,23 @@ HOS_SERVICE_IMPL_INFO(	DeviceRepoHandlerFactory, // Implementation
 
 DeviceRepoHandlerFactory::DeviceRepoHandlerFactory(devicerepository::DeviceRepositoryService* repoService)
 		: repoService(repoService) {
-	infoCategoryFuncs.insert({"id", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"id", [](devicerepository::Device* device, const std::string& category)->Variant {
 		return  device->getDeviceID();
 	} });
 
-	infoCategoryFuncs.insert({"type", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"type", [](devicerepository::Device* device, const std::string& category)->Variant {
 		return  device->getDeviceType();
 	} });
 
-	infoCategoryFuncs.insert({"status_topic", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"status_topic", [](devicerepository::Device* device, const std::string& category)->Variant {
 		return  device->getStatusTopic();
 	} });
 
-	infoCategoryFuncs.insert({"command_topic", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"command_topic", [](devicerepository::Device* device, const std::string& category)->Variant {
 		return  device->getCmdTopic();
 	} });
 
-	infoCategoryFuncs.insert({"current_state", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"current_state", [](devicerepository::Device* device, const std::string& category)->Variant {
 		auto retval = device->getCurrentState();
 		if (retval.IsMap()) {
 			retval["timestamp"] = utils::timestamp_to_ISO8601(device->getCurrentStateTimestamp());
@@ -55,7 +62,7 @@ DeviceRepoHandlerFactory::DeviceRepoHandlerFactory(devicerepository::DeviceRepos
 		return retval;
 	} });
 
-	infoCategoryFuncs.insert({"last_state", [](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"last_state", [](devicerepository::Device* device, const std::string& category)->Variant {
 		auto retval = device->getLastState();
 		if (retval.IsMap()) {
 			retval["timestamp"] = utils::timestamp_to_ISO8601(device->getLastStateTimestamp());
@@ -63,31 +70,47 @@ DeviceRepoHandlerFactory::DeviceRepoHandlerFactory(devicerepository::DeviceRepos
 		return  retval;
 	} });
 
-	infoCategoryFuncs.insert({"uri", [=](Poco::URI& matchUri, devicerepository::Device* device, Poco::Net::HTTPServerRequest &req, const std::string& category, bool ssl)->Variant {
+	infoCategoryFuncs.insert({"uri", [=](devicerepository::Device* device, const std::string& category)->Variant {
 		std::stringstream ss;
-		if (ssl) {
+		/*if (ssl) {
 			ss << "https://";
 		} else {
 			ss << "http://";
 		}
-		ss << req.getHost() << matchUri.toString() << device->getDeviceID();
+		ss << req.getHost() << matchUri.toString() << device->getDeviceID();*/
+		ss << "<???>";
 		return ss.str();
 	} });
+	io_service_thread.reset(new std::thread([this] {
+		try {
+			this->io_service.run();
+		} catch (std::exception& e) {
+			ROS_ERROR_STREAM("IO-Service thread caught exception: " << e.what());
+		} catch(...) {
+			ROS_ERROR("IO-Service thread caught unknown exception");
+		}
+	}));
 }
 
 
+DeviceRepoHandlerFactory::~DeviceRepoHandlerFactory() {
+	if (io_service_thread) {
+		this->io_service.stop();
+		io_service_thread->join();
+	}
+}
 
 Net::HTTPRequestHandler* DeviceRepoHandlerFactory::createRequestHandler( Poco::URI matchUri, const Net::HTTPServerRequest& request, bool ssl ) noexcept {
-	return new DeviceRepoHandler( this, matchUri, repoService, infoCategoryFuncs, ssl);
+	return new DeviceRepoHandler( this, matchUri, repoService, ssl);
 }
 
 std::shared_ptr<DeviceEventQueue> DeviceRepoHandlerFactory::getDeviceEventQueue(	const std::string& clientID ) {
 	if (clientID.empty()) {
-		return std::make_shared<DeviceEventQueue>(clientID, repoService);
+		return std::make_shared<DeviceEventQueue>(clientID, repoService, this);
 	} else {
 		auto i = deviceEventQueues.find(clientID);
 		if (i == deviceEventQueues.end()) {
-			auto n = std::make_shared<DeviceEventQueue>(clientID, repoService);
+			auto n = std::make_shared<DeviceEventQueue>(clientID, repoService, this);
 			deviceEventQueues.insert({clientID, n});
 			return n;
 		} else {
@@ -97,10 +120,15 @@ std::shared_ptr<DeviceEventQueue> DeviceRepoHandlerFactory::getDeviceEventQueue(
 }
 
 
+void DeviceRepoHandlerFactory::deleteDeviceEventQueue( const std::string& clientID ) {
+	deviceEventQueues.erase(clientID);
+}
+
 
 void DeviceRepoHandler::_handleRequest(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) {
 	if(req.find("Upgrade") != req.end() && Poco::icompare(req["Upgrade"], "websocket") == 0) {
-		// WebSocket
+		throw webserver::webresource::WebResourceHandlerException("ToDo",
+				Net::HTTPResponse::HTTP_NOT_IMPLEMENTED);
 	} else {
 		if (req.getMethod() == Net::HTTPRequest::HTTP_PUT) {
 			// Put Command
@@ -175,9 +203,11 @@ void DeviceRepoHandler::__handleInfo(Poco::Net::HTTPServerRequest &req, Poco::Ne
 
 		// Iterate over devices
 		Variant result = Variant::List();
-		for (size_t t = rangeStart; t <= rangeStop; t++) {
-			auto dev = devices[t].lock();
-			result.Append(getDeviceInfo(dev.get(), req));
+		if (devices.size() > 0) {
+			for (size_t t = rangeStart; t <= rangeStop; t++) {
+				auto dev = devices[t].lock();
+				result.Append(parent->getDeviceInfo(dev.get()));
+			}
 		}
 
 		// There are no devices and a range was given, probably out of range
@@ -220,7 +250,7 @@ void DeviceRepoHandler::__handleInfo(Poco::Net::HTTPServerRequest &req, Poco::Ne
 				path << tmp;
 			}
 			// Get device info
-			Variant result = getDeviceInfo(device.lock().get(), req, category, path.str());
+			Variant result = parent->getDeviceInfo(device.lock().get(), category, path.str());
 			// Return device info
 			handleResponse(req, resp, result);
 		} else {
@@ -321,14 +351,16 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 		DeviceEventType event = DeviceEventType::STATE_CHANGED;
 		if (form->has("onevent")) {
 			eventRaw = form->get("onevent");
-		} else if (req.has("ON-EVENT")) {
-			eventRaw = req.get("ON-EVENT");
-		} else if (req.has("X-ON-EVENT")) {
-			eventRaw = req.get("X-ON-EVENT");
+		} else if (req.has("On-Event")) {
+			eventRaw = req.get("On-Event");
+		} else if (req.has("X-On-Event")) {
+			eventRaw = req.get("X-On-Event");
 		}
 		if (!eventRaw.empty()) {
 			if (eventRaw == "all") {
 				event = DeviceEventType::ALL_EVENTS;
+			} else if (eventRaw == "changes") {
+				event = DeviceEventType::CHANGE_EVENTS;
 			} else if (eventRaw == "command") {
 				event = DeviceEventType::COMMAND_RECEIVED;
 			} else if (eventRaw == "statusupdate") {
@@ -348,10 +380,10 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 		std::string clientID;
 		if (form->has("clientid")) {
 			clientID = form->get("clientid");
-		} else if (req.has("Client-ID")) {
-			clientID = req.get("Client-ID");
-		} else if (req.has("X-Client-ID")) {
-			clientID = req.get("X-Client-ID");
+		} else if (req.has("Client-Id")) {
+			clientID = req.get("Client-Id");
+		} else if (req.has("X-Client-Id")) {
+			clientID = req.get("X-Client-Id");
 		}
 		// Get path from url and append pointer parameter
 		std::stringstream path;
@@ -367,6 +399,8 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 		}
 		// Get event queue
 		auto eventQueue = parent->getDeviceEventQueue(clientID);
+		eventQueue->setHandler(this, &req, &resp);
+		std::cout << "Got EventQueue \"" << eventQueue->clientID << "\"" << std::endl;
 		devicerepository::Device* devptr = nullptr;
 		if (singleDeviceMode) {
 			devptr = device.lock().get();
@@ -374,6 +408,7 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 		eventQueue->checkRegistration(event, devptr, path.str());
 		auto msg = eventQueue->popEvent();
 		handleResponse(req, resp, msg);
+		eventQueue->unsetHandler();
 	} else {
 		throw webserver::webresource::WebResourceHandlerException((boost::format("Could not find device \"%1%\"") % devid).str(),
 				Poco::Net::HTTPResponse::HTTPStatus::HTTP_NOT_FOUND);
@@ -382,27 +417,184 @@ void DeviceRepoHandler::__handleEventLongPolling(Poco::Net::HTTPServerRequest &r
 
 
 
+
+DeviceEventQueue::DeviceEventQueue( 	const std::string& clientID,
+					DeviceRepositoryService* repoService,
+					DeviceRepoHandlerFactory* handlerFactory )
+		: clientID(clientID), repoService(repoService), handlerFactory(handlerFactory), timer(handlerFactory->getIOService()) {
+	std::cout << "DeviceEventQueue \"" << clientID << "\" created" << std::endl;
+}
+
+void DeviceEventQueue::checkRegistration(DeviceEventType eventType, Device* device, const std::string& pointer) {
+	if (!registeredToEvent || eventType != this->eventType || device != this->device) {
+		unregisterEvent();
+		registerEvent(eventType, device);
+	}
+	if (this->pointer != pointer) {
+		this->pointer = pointer;
+	}
+}
+
+void DeviceEventQueue::registerEvent(DeviceEventType eventType, Device* device) {
+    std::lock_guard<std::mutex> lock(_mutex);
+	if (!registeredToEvent) {
+		if (device) {
+			device->addObserver(eventType, this);
+		} else {
+			repoService->addObserver(eventType, this);
+		}
+		this->device = device;
+		this->eventType = eventType;
+		registeredToEvent = true;
+	}
+}
+
+void DeviceEventQueue::unregisterEvent() {
+    std::lock_guard<std::mutex> lock(_mutex);
+	if (registeredToEvent) {
+		if (device) {
+			device->deleteObserver(this);
+		} else {
+			repoService->deleteObserver(this);
+		}
+		registeredToEvent = false;
+	}
+}
+
+void DeviceEventQueue::pushEvent(const libvariant::Variant& event) {
+	{
+		std::lock_guard<std::mutex> lock(_mutex);
+		_eventQueue.push(event);
+	}
+	_cv.notify_one();
+}
+
+libvariant::Variant DeviceEventQueue::popEvent() {
+    std::unique_lock<std::mutex> lock(_mutex);
+	libvariant::Variant retval;
+	Net::HTTPServerRequestImpl* reqImpl = static_cast<Net::HTTPServerRequestImpl*>(req);
+	while (_eventQueue.empty()) {
+		_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;
+
+	}
+	if (!_eventQueue.empty()) {
+		retval = _eventQueue.front();
+		_eventQueue.pop();
+	}
+	return retval;
+}
+
+void DeviceEventQueue::setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp) {
+	this->handler = handler;
+	this->req = req;
+	this->resp =resp;
+	openConnectionCount++;
+	timer.cancel();
+}
+
+void DeviceEventQueue::unsetHandler() {
+	this->handler = nullptr;
+	this->req = nullptr;
+	this->resp =nullptr;
+	openConnectionCount--;
+	if (openConnectionCount <= 0) {
+		std::cout << "DeviceEventQueue \"" << clientID << "\" has no open connections." << std::endl;
+		timer.expires_from_now(boost::posix_time::seconds(10));
+		timer.async_wait([this](const boost::system::error_code& error) {
+			if (error != boost::asio::error::operation_aborted) {
+				handlerFactory->deleteDeviceEventQueue(clientID);
+			}
+		});
+	}
+}
+
 void DeviceEventQueue::notificationReceived( 	interface::Observer<devicerepository::DeviceEventType, devicerepository::DeviceEvent>::observable_type* o,
 									const devicerepository::DeviceEventType& key,
 									const devicerepository::DeviceEvent* e) noexcept {
-	pushEvent(Variant("Test"));
+	Variant event;
+	switch(key) {
+		case DeviceEventType::COMMAND_RECEIVED:
+			{
+				event["eventtype"] = "commandmsg";
+				const devicerepository::DeviceCmdMsgEvent* event2 = static_cast<const devicerepository::DeviceCmdMsgEvent*>(e);
+				event["commandmsg"] = event2->cmdMsg;
+				event["deviceid"] = event2->device->getDeviceID();
+				event["timestamp"] = utils::timestamp_to_ISO8601(event2->timestamp);
+			}
+			break;
+		case DeviceEventType::STATUS_UPDATE_RECEIVED:
+			{
+				event["eventtype"] = "statusmsg";
+				const devicerepository::DeviceStatusMsgEvent* event2 = static_cast<const devicerepository::DeviceStatusMsgEvent*>(e);
+				event["statusmsg"] = event2->statusMsg;
+				event["deviceid"] = event2->device->getDeviceID();
+				event["timestamp"] = utils::timestamp_to_ISO8601(event2->timestamp);
+			}
+			break;
+		case DeviceEventType::STATE_CHANGED:
+			{
+				event["eventtype"] = "statechanged";
+				const devicerepository::DeviceStateChangeEvent* event2 = static_cast<const devicerepository::DeviceStateChangeEvent*>(e);
+				event["statusmsg"] = event2->statusMsg;
+				event["deviceid"] = event2->device->getDeviceID();
+				event["device"] = handlerFactory->getDeviceInfo(event2->device, "");
+				event["timestamp"] = utils::timestamp_to_ISO8601(event2->timestamp);
+			}
+			break;
+		case DeviceEventType::DEVICE_ADDED:
+			{
+				event["eventtype"] = "deviceadded";
+				const devicerepository::DeviceAddedEvent* event2 = static_cast<const devicerepository::DeviceAddedEvent*>(e);
+				event["deviceid"] = event2->device->getDeviceID();
+				event["device"] = handlerFactory->getDeviceInfo(event2->device, "");
+				event["timestamp"] = utils::timestamp_to_ISO8601(event2->timestamp);
+			}
+			break;
+		case DeviceEventType::DEVICE_REMOVED:
+			{
+				event["eventtype"] = "deviceremoved";
+				const devicerepository::DeviceRemovedEvent* event2 = static_cast<const devicerepository::DeviceRemovedEvent*>(e);
+				event["deviceid"] = event2->device->getDeviceID();
+				event["timestamp"] = utils::timestamp_to_ISO8601(event2->timestamp);
+			}
+			break;
+		default:
+			break;
+	}
+	if (!event.IsNull()) {
+		pushEvent(event);
+	}
 }
 
 
-Variant DeviceRepoHandler::getDeviceInfo(devicerepository::Device* device, Net::HTTPServerRequest &req, const std::string& category, const std::string& path) {
+Variant DeviceRepoHandlerFactory::getDeviceInfo(devicerepository::Device* device, const std::string& category, const std::string& path) {
 	Variant retval;
 
 	// We want all info categories
 	if (category.empty()) {
 		for (auto c : infoCategoryFuncs) {
-			retval[c.first] = c.second(matchUri, device, req, category, ssl);
+			retval[c.first] = c.second(device, category);
 		}
 
 	// We want a single info category
 	} else {
 		auto c = infoCategoryFuncs.find(category);
 		if (c != infoCategoryFuncs.end()) {
-			retval = c->second(matchUri, device, req, category, ssl);
+			retval = c->second(device, category);
 			if (!path.empty()) {
 				if (retval.HasPath(path)) {
 					retval = retval.GetPath(path);
@@ -471,7 +663,7 @@ DeviceAPIMethod DeviceRepoHandler::getDeviceAPIMethod(Net::HTTPServerRequest &re
 					(boost::format("Unknown device api method \"%1%\"") % methodRaw).str(),
 					Net::HTTPResponse::HTTPStatus::HTTP_BAD_REQUEST );
 		}
-	} else if (form->has("onevent")) {
+	} else if (form->has("onevent") || req.has("On-Event") || req.has("X-On-Event")) {
 		retval = DeviceAPIMethod::EVENT;
 	}
 	return retval;

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

@@ -9,7 +9,6 @@
 #include <Poco/Net/HTTPRequestHandler.h>
 #include <Variant/Variant.h>
 #include <ros/ros.h>
-#pragma once
 
 
 #include <hos/cosgi/service/Implementation.h>
@@ -23,7 +22,10 @@
 #include <queue>
 #include <condition_variable>
 #include <memory>
+#include <thread>
 #include <functional>
+#include <boost/asio.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
 
 
 
@@ -35,6 +37,9 @@ namespace devicerepository {
 namespace webresource {
 
 
+class DeviceRepoHandler;
+class DeviceRepoHandlerFactory;
+
 
 enum class DeviceAPIMethod {
 	GET,
@@ -56,65 +61,27 @@ public:
 	std::string pointer;
 	bool registeredToEvent = false;
 	size_t openConnectionCount = 0;
+	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 )
-		: clientID(clientID), repoService(repoService) {}
-	void pushEvent(const libvariant::Variant& event) {
-		{
-			std::lock_guard<std::mutex> lock(_mutex);
-			_eventQueue.push(event);
-		}
-		_cv.notify_one();
-	}
-	libvariant::Variant popEvent() {
-	    std::unique_lock<std::mutex> lock(_mutex);
-		libvariant::Variant retval;
-		while (_eventQueue.empty()) {
-			_cv.wait(lock);
-		}
-		retval = _eventQueue.front();
-		_eventQueue.pop();
-		return retval;
-	}
-	void checkRegistration(DeviceEventType eventType, Device* device, const std::string& pointer) {
-		if (!registeredToEvent || eventType != this->eventType || device != this->device) {
-			unregisterEvent();
-			registerEvent(eventType, device);
-		}
-		if (this->pointer != pointer) {
-			this->pointer = pointer;
-		}
-	}
-	void registerEvent(DeviceEventType eventType, Device* device = nullptr) {
-	    std::lock_guard<std::mutex> lock(_mutex);
-		if (!registeredToEvent) {
-			if (device) {
-				device->addObserver(eventType, this);
-			} else {
-				repoService->addObserver(eventType, this);
-			}
-			this->device = device;
-			this->eventType = eventType;
-			registeredToEvent = true;
-		}
-	}
-	void unregisterEvent() {
-	    std::lock_guard<std::mutex> lock(_mutex);
-		if (registeredToEvent) {
-			if (device) {
-				device->deleteObserver(this);
-			} else {
-				repoService->deleteObserver(this);
-			}
-			registeredToEvent = false;
-		}
-	}
+						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 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,
@@ -128,19 +95,18 @@ class DeviceRepoHandlerFactory
 		: public virtual webserver::WebResourceHandlerFactory,
 		public implementation::Service<DeviceRepoHandlerFactory> {
 public:
-	typedef std::function<libvariant::Variant( 	Poco::URI& matchUri,
-												devicerepository::Device* device,
-												Poco::Net::HTTPServerRequest &req,
-												const std::string& category,
-												bool ssl )> infoCategoryFunc_t;
+	typedef std::function<libvariant::Variant( 	devicerepository::Device* device, const std::string& category )> infoCategoryFunc_t;
 private:
 	webserver::WebServerService* service = nullptr;
 	DeviceRepositoryService* repoService;
 	std::map<std::string, infoCategoryFunc_t> infoCategoryFuncs;
 	std::map<std::string, std::shared_ptr<DeviceEventQueue>> deviceEventQueues;
+	boost::asio::io_service io_service;
+	std::unique_ptr<std::thread> io_service_thread;
 
 public:
 	DeviceRepoHandlerFactory(devicerepository::DeviceRepositoryService* repoService);
+	~DeviceRepoHandlerFactory();
 
 	virtual Poco::Net::HTTPRequestHandler* createRequestHandler(Poco::URI matchUri,  const Poco::Net::HTTPServerRequest& request, bool ssl ) noexcept override;
 
@@ -152,8 +118,18 @@ public:
 		return false;
 	}
 
+	boost::asio::io_service& getIOService() {
+		return io_service;
+	}
+
 	std::shared_ptr<DeviceEventQueue> getDeviceEventQueue(	const std::string& clientID );
 
+	void deleteDeviceEventQueue( const std::string& clientID );
+
+	libvariant::Variant getDeviceInfo(	devicerepository::Device* device,
+										const std::string& category = std::string(),
+										const std::string& path = std::string() );
+
 };
 
 
@@ -164,7 +140,6 @@ protected:
 	DeviceRepoHandlerFactory* parent;
 	Poco::URI matchUri;
 	devicerepository::DeviceRepositoryService* repoService;
-	const std::map<std::string, DeviceRepoHandlerFactory::infoCategoryFunc_t>& infoCategoryFuncs;
 	bool ssl;
 
 
@@ -173,9 +148,8 @@ public:
 	DeviceRepoHandler( 	DeviceRepoHandlerFactory* parent,
 						Poco::URI matchUri,
 						devicerepository::DeviceRepositoryService* repoService,
-						const std::map<std::string, DeviceRepoHandlerFactory::infoCategoryFunc_t>& infoCategoryFuncs,
 						bool ssl = false )
-		: parent(parent), matchUri(matchUri), repoService(repoService), infoCategoryFuncs(infoCategoryFuncs), ssl(ssl) {}
+		: parent(parent), matchUri(matchUri), repoService(repoService), ssl(ssl) {}
 
 	virtual void _handleRequest(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp) override;
 
@@ -185,11 +159,6 @@ public:
 
 	void __handleEventLongPolling(Poco::Net::HTTPServerRequest &req, Poco::Net::HTTPServerResponse &resp);
 
-	libvariant::Variant getDeviceInfo(	devicerepository::Device* device,
-										Poco::Net::HTTPServerRequest &req,
-										const std::string& category = std::string(),
-										const std::string& path = std::string() );
-
 	void sendDeviceCommand( devicerepository::Device* device,
 							Poco::Net::HTTPServerRequest &req,
 							const std::string& cmd,

src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/dojox/socket.js

@@ -1,111 +1,228 @@
-//>>built
-define("dojox/socket",["dojo/_base/array","dojo/_base/lang","dojo/_base/xhr","dojo/aspect","dojo/on","dojo/Evented","dojo/_base/url"],function(_1,_2,_3,_4,on,_5,_6){
-var _7=window.WebSocket;
-var _8=function(_9){
-if(typeof _9=="string"){
-_9={url:_9};
-}
-return _7?_8.WebSocket(_9,true):_8.LongPoll(_9);
+define("dojox/socket", [
+	"dojo/_base/array",
+	"dojo/_base/lang",
+	"dojo/_base/xhr",
+	"dojo/aspect",
+	"dojo/on",
+	"dojo/Evented",
+	"dojo/_base/url"
+], function(array, lang, xhr, aspect, on, Evented, dBaseUrl) {
+
+var WebSocket = window.WebSocket;
+
+var Socket = function(/*dojo.__XhrArgs*/ argsOrUrl){
+	// summary:
+	//		Provides a simple socket connection using WebSocket, or alternate
+	//		communication mechanisms in legacy browsers for comet-style communication. This is based
+	//		on the WebSocket API and returns an object that implements the WebSocket interface:
+	//		http://dev.w3.org/html5/websockets/#websocket
+	// description:
+	//		Provides socket connections. This can be used with virtually any Comet protocol.
+	// argsOrUrl:
+	//		This uses the same arguments as the other I/O functions in Dojo, or a
+	//		URL to connect to. The URL should be a relative URL in order to properly
+	//		work with WebSockets (it can still be host relative, like //other-site.org/endpoint)
+	// returns:
+	//		An object that implements the WebSocket API
+	// example:
+	//		| require(["dojox/socket", "dojo/aspect"], function(socket, aspect) {
+	//		|    var sock = socket({"url://comet-server/comet");
+	//		|    // we could also add auto-reconnect support
+	//		|    // now we can connect to standard HTML5 WebSocket-style events
+	//		|    aspect.after(socket, "onmessage", function(event){
+	//		|       var message = event.data;
+	//		|       // do something with the message
+	//		|    });
+	//		|    // send something
+	//		|    sock.send("hi there");
+	//		|    ...
+	//		| });
+	//		You can also use the Reconnect module:
+	//		| require["dojox/socket", "dojox/socket/Reconnect"], function(dxSocket, reconnect){
+	//		|    var socket = dxSocket({url:"/comet"});
+	//		|    // add auto-reconnect support
+	//		|    socket = reconnect(socket);
+	if(typeof argsOrUrl == "string"){
+		argsOrUrl = {url: argsOrUrl};
+	}
+	return WebSocket ? Socket.WebSocket(argsOrUrl, true) : Socket.LongPoll(argsOrUrl);
 };
-_8.WebSocket=function(_a,_b){
-var _c=document.baseURI||window.location.href;
-var ws=new _7(new _6(_c.replace(/^http/i,"ws"),_a.url));
-ws.on=function(_d,_e){
-ws.addEventListener(_d,_e,true);
+
+Socket.WebSocket = function(args, fallback){
+	// summary:
+	//		A wrapper for WebSocket, than handles standard args and relative URLs
+	var baseURI = document.baseURI || window.location.href;
+	var ws = new WebSocket(new dBaseUrl(baseURI.replace(/^http/i,'ws'), args.url));
+	ws.on = function(type, listener){
+		ws.addEventListener(type, listener, true);
+	};
+	var opened;
+	aspect.after(ws, "onopen", function(event){
+		opened = true;
+	}, true);
+	aspect.after(ws, "onclose", function(event){
+		if(opened){
+			return;
+		}
+		if(fallback){
+			Socket.replace(ws, Socket.LongPoll(args), true);
+		}
+	}, true);
+	return ws;
 };
-var _f;
-_4.after(ws,"onopen",function(_10){
-_f=true;
-},true);
-_4.after(ws,"onclose",function(_11){
-if(_f){
-return;
-}
-if(_b){
-_8.replace(ws,_8.LongPoll(_a),true);
-}
-},true);
-return ws;
+
+Socket.replace = function(socket, newSocket, listenForOpen){
+	// make the original socket a proxy for the new socket
+	socket.send = lang.hitch(newSocket, "send");
+	socket.close = lang.hitch(newSocket, "close");
+	var proxyEvent = function(type){
+		(newSocket.addEventListener || newSocket.on).call(newSocket, type, function(event){
+			on.emit(socket, event.type, event);
+		}, true);
+	};
+
+	if(listenForOpen){
+		proxyEvent("open");
+	}
+	// redirect the events as well
+	array.forEach(["message", "close", "error"], proxyEvent);
 };
-_8.replace=function(_12,_13,_14){
-_12.send=_2.hitch(_13,"send");
-_12.close=_2.hitch(_13,"close");
-var _15=function(_16){
-(_13.addEventListener||_13.on).call(_13,_16,function(_17){
-on.emit(_12,_17.type,_17);
-},true);
+
+Socket.LongPoll = function(/*dojo.__XhrArgs*/ args){
+	// summary:
+	//		Provides a simple long-poll based comet-style socket/connection to a server and returns an
+	//		object implementing the WebSocket interface:
+	//		http://dev.w3.org/html5/websockets/#websocket
+	// args:
+	//		This uses the same arguments as the other I/O functions in Dojo, with this addition:
+	//	args.interval:
+	//		Indicates the amount of time (in milliseconds) after a response was received
+	//		before another request is made. By default, a request is made immediately
+	//		after getting a response. The interval can be increased to reduce load on the
+	//		server or to do simple time-based polling where the server always responds
+	//		immediately.
+	//	args.transport:
+	//		Provide an alternate transport like dojo.io.script.get
+	// returns:
+	//		An object that implements the WebSocket API
+	// example:
+	//		| dojo.require("dojox.socket.LongPoll");
+	//		| var socket = dojox.socket.LongPoll({url:"/comet"});
+	//		or:
+	//		| dojo.require("dojox.socket.LongPoll");
+	//		| dojox.socket.LongPoll.add();
+	//		| var socket = dojox.socket({url:"/comet"});
+
+	var cancelled = false,
+		first = true,
+		timeoutId,
+		connections = [];
+
+	// create the socket object
+	var fire, connect;
+	var socket = {
+		send: function(data){
+			// summary:
+			//		Send some data using XHR or provided transport
+			var sendArgs = lang.delegate(args);
+			sendArgs.rawBody = data;
+			clearTimeout(timeoutId);
+			var deferred = first ? (first = false) || socket.firstRequest(sendArgs) :
+				socket.transport(sendArgs);
+			connections.push(deferred);
+			deferred.then(function(response){
+				// got a response
+				socket.readyState = 1;
+				// remove the current connection
+				connections.splice(array.indexOf(connections, deferred), 1);
+				// reconnect to listen for the next message if there are no active connections,
+				// we queue it up in case one of the onmessage handlers has a message to send
+				if(!connections.length){
+					timeoutId = setTimeout(connect, args.interval);
+				}
+				if(response){
+					// now send the message along to listeners
+					fire("message", {data: response}, deferred);
+				}
+			}, function(error){
+				connections.splice(array.indexOf(connections, deferred), 1);
+				// an error occurred, fire the appropriate event listeners
+				if(!cancelled){
+					fire("error", {error:error}, deferred);
+					if(!connections.length){
+						socket.readyState = 3;
+						fire("close", {wasClean:false}, deferred);
+					}
+				}
+			});
+			return deferred;
+		},
+		close: function(){
+			// summary:
+			//		Close the connection
+			socket.readyState = 2;
+			cancelled = true;
+			var i;
+			for(i = 0; i < connections.length; i++){
+				connections[i].cancel();
+			}
+			socket.readyState = 3;
+			fire("close", {wasClean:true});
+		},
+		transport: args.transport || xhr.post,
+		args: args,
+		url: args.url,
+		readyState: 0,
+		CONNECTING: 0,
+		OPEN: 1,
+		CLOSING: 2,
+		CLOSED: 3,
+		on: Evented.prototype.on,
+		firstRequest: function(args){
+			// summary:
+			//		This allows for special handling for the first request. This is useful for
+			//		providing information to disambiguate between the first request and
+			//		subsequent long-poll requests so the server can properly setup a
+			//		connection on the first connection or reject a request for an expired
+			//		connection if the request is not expecting to be the first for a connection.
+			//		This method can be overriden. The default behavior is to include a Pragma
+			//		header with a value of "start-long-poll"
+			var headers = (args.headers || (args.headers = {}));
+			headers.Pragma = "start-long-poll";
+			try{
+				return this.transport(args);
+			}finally{
+				// cleanup the header so it is not used on subsequent requests
+				delete headers.Pragma;
+			}
+		}
+	};
+	fire = function(type, object, deferred){
+		if(socket["on" + type]){
+			object.ioArgs = deferred && deferred.ioArgs;
+			object.type = type;
+			on.emit(socket, type, object);
+		}
+	};
+	connect = function(){
+		if(socket.readyState == 0){
+			// we fire the open event now because we really don't know when the "socket"
+			// is truly open, and this gives us a to do a send() and get it included in the
+			// HTTP request
+			fire("open",{});
+		}
+		// make the long-poll connection, to wait for response from the server
+		if(!connections.length){
+			socket.send();
+		}
+	};
+	// provide an alias for Dojo's connect method
+	socket.connect = socket.on;
+	// do the initial connection
+	setTimeout(connect);
+	return socket;
 };
-if(_14){
-_15("open");
-}
-_1.forEach(["message","close","error"],_15);
-};
-_8.LongPoll=function(_18){
-var _19=false,_1a=true,_1b,_1c=[];
-var _1d,_1e;
-var _1f={send:function(_20){
-var _21=_2.delegate(_18);
-_21.rawBody=_20;
-clearTimeout(_1b);
-var _22=_1a?(_1a=false)||_1f.firstRequest(_21):_1f.transport(_21);
-_1c.push(_22);
-_22.then(function(_23){
-_1f.readyState=1;
-_1c.splice(_1.indexOf(_1c,_22),1);
-if(!_1c.length){
-_1b=setTimeout(_1e,_18.interval);
-}
-if(_23){
-_1d("message",{data:_23},_22);
-}
-},function(_24){
-_1c.splice(_1.indexOf(_1c,_22),1);
-if(!_19){
-_1d("error",{error:_24},_22);
-if(!_1c.length){
-_1f.readyState=3;
-_1d("close",{wasClean:false},_22);
-}
-}
-});
-return _22;
-},close:function(){
-_1f.readyState=2;
-_19=true;
-var i;
-for(i=0;i<_1c.length;i++){
-_1c[i].cancel();
-}
-_1f.readyState=3;
-_1d("close",{wasClean:true});
-},transport:_18.transport||_3.post,args:_18,url:_18.url,readyState:0,CONNECTING:0,OPEN:1,CLOSING:2,CLOSED:3,on:_5.prototype.on,firstRequest:function(_25){
-var _26=(_25.headers||(_25.headers={}));
-_26.Pragma="start-long-poll";
-try{
-return this.transport(_25);
-}
-finally{
-delete _26.Pragma;
-}
-}};
-_1d=function(_27,_28,_29){
-if(_1f["on"+_27]){
-_28.ioArgs=_29&&_29.ioArgs;
-_28.type=_27;
-on.emit(_1f,_27,_28);
-}
-};
-_1e=function(){
-if(_1f.readyState==0){
-_1d("open",{});
-}
-if(!_1c.length){
-_1f.send();
-}
-};
-_1f.connect=_1f.on;
-setTimeout(_1e);
-return _1f;
-};
-return _8;
+
+return Socket;
+
 });

src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/hos/DeviceRepo.js

@@ -9,8 +9,13 @@ define([
 	});
 	var _updateSocketOpened = false;
 	var _updateSocket = new Socket({ 
-			url: config.hos.deviceRepoUrl + "?onevent=statechanged",
-			transport: xhr.post });
+			url: config.hos.deviceRepoUrl,
+			headers: {
+				"On-Event": "changes",
+				"Connection-Timeout": 60000,
+				"Client-Id": 1234
+			},
+			transport: xhr.get });
 	_updateSocket.on("open", function(event){
 			_socketOpened = true;
 			console.log("Socket Opened ", event.message);