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 78aeeef0..3e9fe386 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 @@ -9,8 +9,12 @@ #include #include #include +#include +#include "Poco/Net/NetException.h" #include #include +#include +#include #include @@ -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(&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 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(); - handleResponse(req, resp, msg); + if (websocket) { + __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 lock(_mutex); libvariant::Variant retval; + breakPopLoop = false; Net::HTTPServerRequestImpl* reqImpl = static_cast(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) { - this->handler = handler; - this->req = req; - this->resp =resp; - openConnectionCount++; - timer.cancel(); +bool DeviceEventQueue::setHandler(DeviceRepoHandler* handler, Poco::Net::HTTPServerRequest* req, Poco::Net::HTTPServerResponse* resp) { + if (openConnectionCount <= 0) { + this->handler = handler; + this->req = req; + this->resp =resp; + openConnectionCount++; + timer.cancel(); + return true; + } + return false; } void DeviceEventQueue::unsetHandler() { 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 fe60d10e..ab518c10 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 @@ -54,35 +54,52 @@ enum class DeviceAPIMethod { class DeviceEventQueue : public virtual interface::Observer { 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 _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::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,