From e7451b1c5931e1637979bf6a5a5741762a0b5ae4 Mon Sep 17 00:00:00 2001 From: Matthias Janetschek Date: Tue, 2 Jun 2015 17:17:09 +0200 Subject: [PATCH] . --- .../DeviceRepositoryService.h | 3 +- .../DeviceRepositoryServiceImpl.cpp | 5 + .../webresource/DeviceRepoHandler.cpp | 252 ++++++++++++-- .../webresource/DeviceRepoHandler.h | 103 ++---- .../webroot/DojoApp/dojo/dojox/socket.js | 329 ++++++++++++------ .../webroot/DojoApp/dojo/hos/DeviceRepo.js | 9 +- 6 files changed, 495 insertions(+), 206 deletions(-) diff --git a/src/hos_cosgi_devicerepository/include/hos_cosgi_devicerepository/hos/cosgi/service/devicerepository/DeviceRepositoryService.h b/src/hos_cosgi_devicerepository/include/hos_cosgi_devicerepository/hos/cosgi/service/devicerepository/DeviceRepositoryService.h index 0b1ae91c..5c2d8969 100644 --- a/src/hos_cosgi_devicerepository/include/hos_cosgi_devicerepository/hos/cosgi/service/devicerepository/DeviceRepositoryService.h +++ b/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 }; diff --git a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/DeviceRepositoryServiceImpl.cpp b/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/DeviceRepositoryServiceImpl.cpp index 9cc8f44b..e688b330 100644 --- a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/DeviceRepositoryServiceImpl.cpp +++ b/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); } } diff --git a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.cpp b/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.cpp index de1851cb..78aeeef0 100644 --- a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.cpp +++ b/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.cpp @@ -6,6 +6,13 @@ #include #include #include +#include +#include +#include +#include +#include +#include + // 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 DeviceRepoHandlerFactory::getDeviceEventQueue( const std::string& clientID ) { if (clientID.empty()) { - return std::make_shared(clientID, repoService); + return std::make_shared(clientID, repoService, this); } else { auto i = deviceEventQueues.find(clientID); if (i == deviceEventQueues.end()) { - auto n = std::make_shared(clientID, repoService); + auto n = std::make_shared(clientID, repoService, this); deviceEventQueues.insert({clientID, n}); return n; } else { @@ -97,10 +120,15 @@ std::shared_ptr 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 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 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 lock(_mutex); + _eventQueue.push(event); + } + _cv.notify_one(); +} + +libvariant::Variant DeviceEventQueue::popEvent() { + std::unique_lock lock(_mutex); + libvariant::Variant retval; + Net::HTTPServerRequestImpl* reqImpl = static_cast(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::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(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(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(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(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(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; diff --git a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.h b/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.h index fba45c62..fe60d10e 100644 --- a/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.h +++ b/src/hos_cosgi_devicerepository/src/hos/cosgi/service/devicerepository/webresource/DeviceRepoHandler.h @@ -9,7 +9,6 @@ #include #include #include -#pragma once #include @@ -23,7 +22,10 @@ #include #include #include +#include #include +#include +#include @@ -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 _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 lock(_mutex); - _eventQueue.push(event); - } - _cv.notify_one(); - } - libvariant::Variant popEvent() { - std::unique_lock 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 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 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::observable_type* o, const devicerepository::DeviceEventType& key, @@ -128,19 +95,18 @@ class DeviceRepoHandlerFactory : public virtual webserver::WebResourceHandlerFactory, public implementation::Service { public: - typedef std::function infoCategoryFunc_t; + typedef std::function infoCategoryFunc_t; private: webserver::WebServerService* service = nullptr; DeviceRepositoryService* repoService; std::map infoCategoryFuncs; std::map> deviceEventQueues; + boost::asio::io_service io_service; + std::unique_ptr 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 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& infoCategoryFuncs; bool ssl; @@ -173,9 +148,8 @@ public: DeviceRepoHandler( DeviceRepoHandlerFactory* parent, Poco::URI matchUri, devicerepository::DeviceRepositoryService* repoService, - const std::map& 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, diff --git a/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/dojox/socket.js b/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/dojox/socket.js index 5f781d0c..675adf7c 100644 --- a/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/dojox/socket.js +++ b/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; + }); diff --git a/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/hos/DeviceRepo.js b/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/hos/DeviceRepo.js index d6642172..184146eb 100644 --- a/src/hos_cosgi_testwebapp/webroot/DojoApp/dojo/hos/DeviceRepo.js +++ b/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);