net/Socket.cpp | 57 +++++++++++++ net/Socket.hpp | 53 +----------- wsd/Admin.cpp | 222 +++++++++++++++++++++++++++-------------------------- wsd/Admin.hpp | 92 ++++++--------------- wsd/AdminModel.cpp | 6 - wsd/AdminModel.hpp | 10 +- wsd/Exceptions.hpp | 9 -- wsd/FileServer.cpp | 2 wsd/LOOLWSD.cpp | 18 ++-- 9 files changed, 225 insertions(+), 244 deletions(-)
New commits: commit f392d9e6f0103360dbcaaf99fc4768dfd9624eba Author: Michael Meeks <[email protected]> Date: Wed Mar 15 18:21:59 2017 +0000 Move http serving into socket impl. Avoid caching headers with parameter, and add Date: parameter. diff --git a/net/Socket.cpp b/net/Socket.cpp index 5f13202..7af8760 100644 --- a/net/Socket.cpp +++ b/net/Socket.cpp @@ -12,6 +12,10 @@ #include <stdio.h> #include <ctype.h> +#include <Poco/DateTime.h> +#include <Poco/DateTimeFormat.h> +#include <Poco/DateTimeFormatter.h> + #include "SigUtil.hpp" #include "Socket.hpp" #include "ServerSocket.hpp" @@ -147,4 +151,57 @@ void SocketPoll::dumpState(std::ostream& os) i->dumpState(os); } +namespace HttpHelper +{ + void sendFile(const std::shared_ptr<StreamSocket>& socket, const std::string& path, + Poco::Net::HTTPResponse& response, bool noCache) + { + struct stat st; + if (stat(path.c_str(), &st) != 0) + { + LOG_WRN("#" << socket->getFD() << ": Failed to stat [" << path << "]. File will not be sent."); + throw Poco::FileNotFoundException("Failed to stat [" + path + "]. File will not be sent."); + return; + } + + int bufferSize = std::min(st.st_size, (off_t)Socket::MaximumSendBufferSize); + if (st.st_size >= socket->getSendBufferSize()) + { + socket->setSocketBufferSize(bufferSize); + bufferSize = socket->getSendBufferSize(); + } + + response.setContentLength(st.st_size); + response.set("User-Agent", HTTP_AGENT_STRING); + response.set("Date", Poco::DateTimeFormatter::format(Poco::Timestamp(), Poco::DateTimeFormat::HTTP_FORMAT)); + if (!noCache) + { + // 60 * 60 * 24 * 128 (days) = 11059200 + response.set("Cache-Control", "max-age=11059200"); + response.set("ETag", "\"" LOOLWSD_VERSION_HASH "\""); + } + + std::ostringstream oss; + response.write(oss); + const std::string header = oss.str(); + LOG_TRC("#" << socket->getFD() << ": Sending file [" << path << "]: " << header); + socket->send(header); + + std::ifstream file(path, std::ios::binary); + bool flush = true; + do + { + char buf[bufferSize]; + file.read(buf, sizeof(buf)); + const int size = file.gcount(); + if (size > 0) + socket->send(buf, size, flush); + else + break; + flush = false; + } + while (file); + } +} + /* vim:set shiftwidth=4 softtabstop=4 expandtab: */ diff --git a/net/Socket.hpp b/net/Socket.hpp index 9460c45..708ae39 100644 --- a/net/Socket.hpp +++ b/net/Socket.hpp @@ -820,58 +820,15 @@ protected: namespace HttpHelper { - inline void sendFile(const std::shared_ptr<StreamSocket>& socket, const std::string& path, - Poco::Net::HTTPResponse& response) - { - struct stat st; - if (stat(path.c_str(), &st) != 0) - { - LOG_WRN("#" << socket->getFD() << ": Failed to stat [" << path << "]. File will not be sent."); - throw Poco::FileNotFoundException("Failed to stat [" + path + "]. File will not be sent."); - return; - } - - int bufferSize = std::min(st.st_size, (off_t)Socket::MaximumSendBufferSize); - if (st.st_size >= socket->getSendBufferSize()) - { - socket->setSocketBufferSize(bufferSize); - bufferSize = socket->getSendBufferSize(); - } - - response.setContentLength(st.st_size); - response.set("User-Agent", HTTP_AGENT_STRING); - // 60 * 60 * 24 * 128 (days) = 11059200 - response.set("Cache-Control", "max-age=11059200"); - response.set("ETag", "\"" LOOLWSD_VERSION_HASH "\""); - - std::ostringstream oss; - response.write(oss); - const std::string header = oss.str(); - LOG_TRC("#" << socket->getFD() << ": Sending file [" << path << "]: " << header); - socket->send(header); - - std::ifstream file(path, std::ios::binary); - bool flush = true; - do - { - char buf[bufferSize]; - file.read(buf, sizeof(buf)); - const int size = file.gcount(); - if (size > 0) - socket->send(buf, size, flush); - else - break; - flush = false; - } - while (file); - } + void sendFile(const std::shared_ptr<StreamSocket>& socket, const std::string& path, + Poco::Net::HTTPResponse& response, bool noCache = false); inline void sendFile(const std::shared_ptr<StreamSocket>& socket, const std::string& path, - const std::string& mediaType) + const std::string& mediaType, bool noCache = false) { Poco::Net::HTTPResponse response; response.setContentType(mediaType); - sendFile(socket, path, response); + sendFile(socket, path, response, noCache); } }; diff --git a/wsd/FileServer.cpp b/wsd/FileServer.cpp index 2189205..bba77d5 100644 --- a/wsd/FileServer.cpp +++ b/wsd/FileServer.cpp @@ -197,7 +197,7 @@ void FileServerRequestHandler::handleRequest(const HTTPRequest& request, Poco::M } response.setContentType(mimeType); - HttpHelper::sendFile(socket, filepath, response); + HttpHelper::sendFile(socket, filepath, response, noCache); } } catch (const Poco::Net::NotAuthenticatedException& exc) commit d19b6eb3512931bbb2dc45004ca75aef682d9709 Author: Michael Meeks <[email protected]> Date: Wed Mar 15 16:39:13 2017 +0000 Move memstats & cpustats into the main polling thread. We can calculate the timeout ourselves easily and add it to the polling loop simplifying life. Also ensure we never send messages to a non-authenticated thread. diff --git a/wsd/Admin.cpp b/wsd/Admin.cpp index e4b616b..a3b7035 100644 --- a/wsd/Admin.cpp +++ b/wsd/Admin.cpp @@ -32,7 +32,6 @@ #include "FileServer.hpp" #include "IoUtil.hpp" #include "Protocol.hpp" -#include "LOOLWebSocket.hpp" // FIXME: remove. #include "LOOLWSD.hpp" #include "Log.hpp" #include "Storage.hpp" @@ -87,16 +86,18 @@ void AdminRequestHandler::handleMessage(bool /* fin */, WSOpCode /* code */, std } else { - sendTextFrame("InvalidAuthToken"); + sendFrame("InvalidAuthToken"); LOG_TRC("Invalid auth token"); + shutdown(); return; } } if (!_isAuthenticated) { - sendTextFrame("NotAuthenticated"); + sendFrame("NotAuthenticated"); LOG_TRC("Not authenticated"); + shutdown(); return; } else if (tokens[0] == "documents" || @@ -229,7 +230,11 @@ AdminRequestHandler::AdminRequestHandler(Admin* adminManager, void AdminRequestHandler::sendTextFrame(const std::string& message) { UnitWSD::get().onAdminQueryMessage(message); - sendFrame(message); + std::cerr << "Admin: send text frame '" << message << "'\n"; + if (_isAuthenticated) + sendFrame(message); + else + LOG_TRC("Skip sending message to non-authenticated client: '" << message << "'"); } bool AdminRequestHandler::handleInitialRequest( @@ -273,7 +278,10 @@ bool AdminRequestHandler::handleInitialRequest( Admin::Admin() : SocketPoll("admin"), _model(AdminModel()), - _forKitPid(-1) + _forKitPid(-1), + _lastTotalMemory(0), + _memStatsTaskIntervalMs(5000), + _cpuStatsTaskIntervalMs(5000) { LOG_INF("Admin ctor."); @@ -281,20 +289,54 @@ Admin::Admin() : const auto totalMem = getTotalMemoryUsage(); LOG_TRC("Total memory used: " << totalMem); _model.addMemStats(totalMem); - - _memStatsTask = new MemoryStatsTask(this); - _memStatsTimer.schedule(_memStatsTask, _memStatsTaskInterval, _memStatsTaskInterval); - - _cpuStatsTask = new CpuStats(this); - _cpuStatsTimer.schedule(_cpuStatsTask, _cpuStatsTaskInterval, _cpuStatsTaskInterval); } Admin::~Admin() { LOG_INF("~Admin dtor."); +} + +void Admin::pollingThread() +{ + std::chrono::steady_clock::time_point lastCPU, lastMem; + + lastCPU = std::chrono::steady_clock::now(); + lastMem = lastCPU; + + while (!_stop && !TerminationFlag && !ShutdownRequestFlag) + { + std::chrono::steady_clock::time_point now = std::chrono::steady_clock::now(); + int cpuWait = _cpuStatsTaskIntervalMs - + std::chrono::duration_cast<std::chrono::milliseconds>(now - lastCPU).count(); + if (cpuWait < 0) + { + // TODO: implement me ... + lastCPU = now; + cpuWait += _cpuStatsTaskIntervalMs; + } + int memWait = _memStatsTaskIntervalMs - + std::chrono::duration_cast<std::chrono::milliseconds>(now - lastCPU).count(); + if (memWait < 0) + { + std::unique_lock<std::mutex> modelLock(getLock()); + const auto totalMem = getTotalMemoryUsage(); + if (totalMem != _lastTotalMemory) + { + LOG_TRC("Total memory used: " << totalMem); + _lastTotalMemory = totalMem; + } + + _model.addMemStats(totalMem); - _memStatsTask->cancel(); - _cpuStatsTask->cancel(); + lastMem = now; + memWait += _memStatsTaskIntervalMs; + } + + // Handle websockets & other work. + int timeout = std::min(cpuWait, memWait); + LOG_TRC("Admin poll for " << timeout << "ms"); + poll(timeout); + } } void Admin::addDoc(const std::string& docKey, Poco::Process::PID pid, const std::string& filename, const std::string& sessionId) @@ -316,43 +358,18 @@ void Admin::rmDoc(const std::string& docKey) _model.removeDocument(docKey); } -void MemoryStatsTask::run() -{ - std::unique_lock<std::mutex> modelLock(_admin->getLock()); - const auto totalMem = _admin->getTotalMemoryUsage(); - - if (totalMem != _lastTotalMemory) - { - LOG_TRC("Total memory used: " << totalMem); - _lastTotalMemory = totalMem; - } - - _admin->getModel().addMemStats(totalMem); -} - -void CpuStats::run() -{ - //TODO: Implement me - //std::unique_lock<std::mutex> modelLock(_admin->getLock()); - //model.addCpuStats(totalMem); -} - void Admin::rescheduleMemTimer(unsigned interval) { - _memStatsTask->cancel(); - _memStatsTaskInterval = interval; - _memStatsTask = new MemoryStatsTask(this); - _memStatsTimer.schedule(_memStatsTask, _memStatsTaskInterval, _memStatsTaskInterval); + _memStatsTaskIntervalMs = interval; LOG_INF("Memory stats interval changed - New interval: " << interval); + wakeup(); } void Admin::rescheduleCpuTimer(unsigned interval) { - _cpuStatsTask->cancel(); - _cpuStatsTaskInterval = interval; - _cpuStatsTask = new CpuStats(this); - _cpuStatsTimer.schedule(_cpuStatsTask, _cpuStatsTaskInterval, _cpuStatsTaskInterval); + _cpuStatsTaskIntervalMs = interval; LOG_INF("CPU stats interval changed - New interval: " << interval); + wakeup(); } unsigned Admin::getTotalMemoryUsage() @@ -373,12 +390,12 @@ unsigned Admin::getTotalMemoryUsage() unsigned Admin::getMemStatsInterval() { - return _memStatsTaskInterval; + return _memStatsTaskIntervalMs; } unsigned Admin::getCpuStatsInterval() { - return _cpuStatsTaskInterval; + return _cpuStatsTaskIntervalMs; } AdminModel& Admin::getModel() diff --git a/wsd/Admin.hpp b/wsd/Admin.hpp index 2b8d4f0..da51cdc 100644 --- a/wsd/Admin.hpp +++ b/wsd/Admin.hpp @@ -22,7 +22,6 @@ #include "AdminModel.hpp" #include "Log.hpp" -#include <LOOLWebSocket.hpp> #include "net/WebSocketHandler.hpp" @@ -76,6 +75,9 @@ public: startThread(); } + /// Custom poll thread function + void pollingThread() override; + unsigned getTotalMemoryUsage(); /// Update the Admin Model. @@ -114,58 +116,10 @@ private: AdminModel _model; std::mutex _modelMutex; int _forKitPid; - - Poco::Util::Timer _memStatsTimer; - Poco::AutoPtr<MemoryStatsTask> _memStatsTask; - unsigned _memStatsTaskInterval = 5000; - - Poco::Util::Timer _cpuStatsTimer; - Poco::Util::TimerTask::Ptr _cpuStatsTask; - unsigned _cpuStatsTaskInterval = 5000; -}; - -/// Memory statistics. -class MemoryStatsTask : public Poco::Util::TimerTask -{ -public: - MemoryStatsTask(Admin* admin) - : _admin(admin), - _lastTotalMemory(0) - { - LOG_DBG("Memory stat ctor"); - } - - ~MemoryStatsTask() - { - LOG_DBG("Memory stat dtor"); - } - - long getLastTotalMemory() { return _lastTotalMemory; } - - void run() override; - -private: - Admin* _admin; long _lastTotalMemory; -}; - -/// CPU statistics. -class CpuStats : public Poco::Util::TimerTask -{ -public: - CpuStats(Admin* /*admin*/) - { - LOG_DBG("Cpu stat ctor"); - } - ~CpuStats() - { - LOG_DBG("Cpu stat dtor"); - } - - void run() override; - -private: + std::atomic<int> _memStatsTaskIntervalMs; + std::atomic<int> _cpuStatsTaskIntervalMs; }; #endif diff --git a/wsd/Exceptions.hpp b/wsd/Exceptions.hpp index b05d98d..b97263e 100644 --- a/wsd/Exceptions.hpp +++ b/wsd/Exceptions.hpp @@ -58,15 +58,6 @@ public: using LoolException::LoolException; }; -/// An generic error-message exception meant to -/// propagate via a valid LOOLWebSocket to the client. -/// The contents of what() will be displayed on screen. -class WebSocketErrorMessageException : public LoolException -{ -public: - using LoolException::LoolException; -}; - #endif /* vim:set shiftwidth=4 softtabstop=4 expandtab: */ commit 0089723f693a8d9549b9abb7236c2f5bae4d633a Author: Michael Meeks <[email protected]> Date: Wed Mar 15 16:25:29 2017 +0000 Admin: review error handling on auth. failure. diff --git a/wsd/Admin.cpp b/wsd/Admin.cpp index d3e684b..e4b616b 100644 --- a/wsd/Admin.cpp +++ b/wsd/Admin.cpp @@ -232,10 +232,10 @@ void AdminRequestHandler::sendTextFrame(const std::string& message) sendFrame(message); } -bool AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> &socketWeak, - const Poco::Net::HTTPRequest& request) +bool AdminRequestHandler::handleInitialRequest( + const std::weak_ptr<StreamSocket> &socketWeak, + const Poco::Net::HTTPRequest& request) { - HTTPResponse response; auto socket = socketWeak.lock(); // Different session id pool for admin sessions (?) @@ -260,28 +260,12 @@ bool AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> return true; } -// FIXME: ... should we move ourselves to an Admin poll [!?] ... -// Probably [!] ... and use this termination thing ? ... -// Perhaps that should be the 'Admin' instance / sub-class etc. -// FIXME: have name 'admin' etc. -// []() { return TerminationFlag.load(); }); -#if 0 - catch(const Poco::Net::NotAuthenticatedException& exc) - { - LOG_INF("Admin::NotAuthenticated"); - response.set("WWW-Authenticate", "Basic realm=\"online\""); - response.setStatusAndReason(HTTPResponse::HTTP_UNAUTHORIZED); - response.setContentLength(0); - socket->send(response); - } - catch (const std::exception& exc) - { - LOG_INF("Admin::handleRequest: Exception: " << exc.what()); - response.setStatusAndReason(HTTPResponse::HTTP_BAD_REQUEST); - response.setContentLength(0); - socket->send(response); - } -#endif + HTTPResponse response; + response.setStatusAndReason(HTTPResponse::HTTP_BAD_REQUEST); + response.setContentLength(0); + LOG_INF("Admin::handleInitialRequest bad request"); + socket->send(response); + return false; } diff --git a/wsd/Admin.hpp b/wsd/Admin.hpp index 4332e27..2b8d4f0 100644 --- a/wsd/Admin.hpp +++ b/wsd/Admin.hpp @@ -51,7 +51,6 @@ private: private: Admin* _admin; -// std::shared_ptr<LOOLWebSocket> _adminWs; FIXME - this is us now ! int _sessionId; bool _isAuthenticated; }; commit 909b5f8ac3aa57c4bc786fa03c5d999805e6d7ca Author: Michael Meeks <[email protected]> Date: Wed Mar 15 16:13:13 2017 +0000 Admin: should be its own socket-poll goodness. diff --git a/net/Socket.hpp b/net/Socket.hpp index 1d67bb6..9460c45 100644 --- a/net/Socket.hpp +++ b/net/Socket.hpp @@ -412,7 +412,7 @@ public: wakeup(); } - void dumpState(std::ostream& os); + virtual void dumpState(std::ostream& os); /// Removes a socket from this poller. /// NB. this must be called from the socket poll that diff --git a/wsd/Admin.cpp b/wsd/Admin.cpp index 6f9f7cb..d3e684b 100644 --- a/wsd/Admin.cpp +++ b/wsd/Admin.cpp @@ -232,7 +232,7 @@ void AdminRequestHandler::sendTextFrame(const std::string& message) sendFrame(message); } -void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> &socketWeak, +bool AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> &socketWeak, const Poco::Net::HTTPRequest& request) { HTTPResponse response; @@ -257,6 +257,7 @@ void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> handler->_sessionId = sessionId; model.subscribe(sessionId, handler); } + return true; } // FIXME: ... should we move ourselves to an Admin poll [!?] ... @@ -281,10 +282,12 @@ void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> socket->send(response); } #endif + return false; } /// An admin command processor. Admin::Admin() : + SocketPoll("admin"), _model(AdminModel()), _forKitPid(-1) { @@ -411,4 +414,10 @@ void Admin::updateMemoryDirty(const std::string& docKey, int dirty) _model.updateMemoryDirty(docKey, dirty); } +void Admin::dumpState(std::ostream& os) +{ + // FIXME: be more helpful ... + SocketPoll::dumpState(os); +} + /* vim:set shiftwidth=4 softtabstop=4 expandtab: */ diff --git a/wsd/Admin.hpp b/wsd/Admin.hpp index c15cbbb..4332e27 100644 --- a/wsd/Admin.hpp +++ b/wsd/Admin.hpp @@ -36,7 +36,9 @@ public: const std::weak_ptr<StreamSocket>& socket, const Poco::Net::HTTPRequest& request); - static void handleInitialRequest(const std::weak_ptr<StreamSocket> &socket, + /// Handle the initial Admin WS upgrade request. + /// @returns true if we should give this socket to the Admin poll. + static bool handleInitialRequest(const std::weak_ptr<StreamSocket> &socket, const Poco::Net::HTTPRequest& request); private: @@ -57,7 +59,7 @@ private: class MemoryStatsTask; /// An admin command processor. -class Admin +class Admin : public SocketPoll { Admin(); public: @@ -69,6 +71,12 @@ public: return admin; } + void start() + { + // FIXME: not if admin console is not enabled ? + startThread(); + } + unsigned getTotalMemoryUsage(); /// Update the Admin Model. @@ -101,6 +109,8 @@ public: void updateLastActivityTime(const std::string& docKey); void updateMemoryDirty(const std::string& docKey, int dirty); + void dumpState(std::ostream& os) override; + private: AdminModel _model; std::mutex _modelMutex; diff --git a/wsd/LOOLWSD.cpp b/wsd/LOOLWSD.cpp index 3aee14d..81608b6 100644 --- a/wsd/LOOLWSD.cpp +++ b/wsd/LOOLWSD.cpp @@ -1808,8 +1808,12 @@ private: else if (reqPathSegs.size() >= 2 && reqPathSegs[0] == "lool" && reqPathSegs[1] == "adminws") { LOG_ERR("Admin request: " << request.getURI()); - AdminRequestHandler::handleInitialRequest(_socket, request); - + if (AdminRequestHandler::handleInitialRequest(_socket, request)) + { + // Hand the socket over to the Admin poll. + WebServerPoll.releaseSocket(socket); + Admin::instance().insertNewSocket(socket); + } } // Client post and websocket connections else if ((request.getMethod() == HTTPRequest::HTTP_GET || @@ -2336,6 +2340,7 @@ public: _acceptPoll.insertNewSocket(findServerPort(port)); _acceptPoll.startThread(); WebServerPoll.startThread(); + Admin::instance().start(); } void stop() @@ -2364,6 +2369,9 @@ public: os << "Prisoner poll:\n"; PrisonerPoll.dumpState(os); + os << "Admin poll:\n"; + Admin::instance().dumpState(os); + os << "Document Broker polls " << "[ " << DocBrokers.size() << " ]:\n"; for (auto &i : DocBrokers) commit e1f576bdb38f9cf63f1c74cde55afdd25d7b96ed Author: Michael Meeks <[email protected]> Date: Wed Mar 15 14:58:55 2017 +0000 Switch to using websocket here. diff --git a/wsd/Admin.cpp b/wsd/Admin.cpp index 07723df..6f9f7cb 100644 --- a/wsd/Admin.cpp +++ b/wsd/Admin.cpp @@ -54,14 +54,19 @@ using Poco::Net::HTTPServerRequest; using Poco::Net::HTTPServerResponse; using Poco::Util::Application; -bool AdminRequestHandler::adminCommandHandler(const std::vector<char>& payload) + /// Process incoming websocket messages +void AdminRequestHandler::handleMessage(bool /* fin */, WSOpCode /* code */, std::vector<char> &payload) { + // FIXME: check fin, code etc. const std::string firstLine = getFirstLine(payload.data(), payload.size()); StringTokenizer tokens(firstLine, " ", StringTokenizer::TOK_IGNORE_EMPTY | StringTokenizer::TOK_TRIM); - LOG_TRC("Recv: " << firstLine); + LOG_TRC("Recv: " << firstLine << " tokens " << tokens.count()); if (tokens.count() < 1) - return false; + { + LOG_TRC("too few tokens"); + return; + } std::unique_lock<std::mutex> modelLock(_admin->getLock()); AdminModel& model = _admin->getModel(); @@ -83,14 +88,16 @@ bool AdminRequestHandler::adminCommandHandler(const std::vector<char>& payload) else { sendTextFrame("InvalidAuthToken"); - return false; + LOG_TRC("Invalid auth token"); + return; } } if (!_isAuthenticated) { sendTextFrame("NotAuthenticated"); - return false; + LOG_TRC("Not authenticated"); + return; } else if (tokens[0] == "documents" || tokens[0] == "active_users_count" || @@ -154,7 +161,7 @@ bool AdminRequestHandler::adminCommandHandler(const std::vector<char>& payload) LOG_INF("Shutdown requested by admin."); ShutdownRequestFlag = true; SocketPoll::wakeupWorld(); - return false; + return; } else if (tokens[0] == "set" && tokens.count() > 1) { @@ -170,7 +177,7 @@ bool AdminRequestHandler::adminCommandHandler(const std::vector<char>& payload) { LOG_WRN("Invalid setting value: " << setting[1] << " for " << setting[0]); - return false; + return; } if (setting[0] == "mem_stats_size") @@ -207,36 +214,13 @@ bool AdminRequestHandler::adminCommandHandler(const std::vector<char>& payload) } } } - - return true; -} - -/// Handle admin requests. -void AdminRequestHandler::handleWSRequests(HTTPServerRequest& request, HTTPServerResponse& response, int sessionId) -{ - _adminWs = std::make_shared<LOOLWebSocket>(request, response); - - { - std::unique_lock<std::mutex> modelLock(_admin->getLock()); - // Subscribe the websocket of any AdminModel updates - AdminModel& model = _admin->getModel(); - _sessionId = sessionId; - model.subscribe(_sessionId, _adminWs); - } - - IoUtil::SocketProcessor(_adminWs, "admin", - [this](const std::vector<char>& payload) - { - return adminCommandHandler(payload); - }, - []() { }, - []() { return TerminationFlag.load(); }); - - LOG_DBG("Finishing Admin Session " << Util::encodeId(sessionId)); } -AdminRequestHandler::AdminRequestHandler(Admin* adminManager) - : _admin(adminManager), +AdminRequestHandler::AdminRequestHandler(Admin* adminManager, + const std::weak_ptr<StreamSocket>& socket, + const Poco::Net::HTTPRequest& request) + : WebSocketHandler(socket, request), + _admin(adminManager), _sessionId(0), _isAuthenticated(false) { @@ -245,7 +229,7 @@ AdminRequestHandler::AdminRequestHandler(Admin* adminManager) void AdminRequestHandler::sendTextFrame(const std::string& message) { UnitWSD::get().onAdminQueryMessage(message); - _adminWs->sendFrame(message.data(), message.size()); + sendFrame(message); } void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> &socketWeak, @@ -255,21 +239,32 @@ void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> auto socket = socketWeak.lock(); // Different session id pool for admin sessions (?) -// const auto sessionId = Util::decodeId(LOOLWSD::GenSessionId()); - - try - { - std::string requestURI = request.getURI(); - StringTokenizer pathTokens(requestURI, "/", StringTokenizer::TOK_IGNORE_EMPTY | StringTokenizer::TOK_TRIM); + const auto sessionId = Util::decodeId(LOOLWSD::GenSessionId()); - if (request.find("Upgrade") != request.end() && Poco::icompare(request["Upgrade"], "websocket") == 0) - { - // First Upgrade. - WebSocketHandler ws(socketWeak, request); + std::string requestURI = request.getURI(); + StringTokenizer pathTokens(requestURI, "/", StringTokenizer::TOK_IGNORE_EMPTY | StringTokenizer::TOK_TRIM); -// handleWSRequests(request, response, sessionId); + if (request.find("Upgrade") != request.end() && Poco::icompare(request["Upgrade"], "websocket") == 0) + { + Admin &admin = Admin::instance(); + auto handler = std::make_shared<AdminRequestHandler>(&admin, socketWeak, request); + socket->setHandler(handler); + + { // FIXME: weird locking around subscribe ... + std::unique_lock<std::mutex> modelLock(admin.getLock()); + // Subscribe the websocket of any AdminModel updates + AdminModel& model = admin.getModel(); + handler->_sessionId = sessionId; + model.subscribe(sessionId, handler); } } + +// FIXME: ... should we move ourselves to an Admin poll [!?] ... +// Probably [!] ... and use this termination thing ? ... +// Perhaps that should be the 'Admin' instance / sub-class etc. +// FIXME: have name 'admin' etc. +// []() { return TerminationFlag.load(); }); +#if 0 catch(const Poco::Net::NotAuthenticatedException& exc) { LOG_INF("Admin::NotAuthenticated"); @@ -285,6 +280,7 @@ void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> response.setContentLength(0); socket->send(response); } +#endif } /// An admin command processor. diff --git a/wsd/Admin.hpp b/wsd/Admin.hpp index 82b5a5e..c15cbbb 100644 --- a/wsd/Admin.hpp +++ b/wsd/Admin.hpp @@ -24,14 +24,17 @@ #include "Log.hpp" #include <LOOLWebSocket.hpp> +#include "net/WebSocketHandler.hpp" + class Admin; -class StreamSocket; -/// Admin requests over HTTP(S) handler. -class AdminRequestHandler /// public Poco::Net::HTTPRequestHandler +/// Handle admin client's Websocket requests & replies. +class AdminRequestHandler : public WebSocketHandler { public: - AdminRequestHandler(Admin* adminManager); + AdminRequestHandler(Admin* adminManager, + const std::weak_ptr<StreamSocket>& socket, + const Poco::Net::HTTPRequest& request); static void handleInitialRequest(const std::weak_ptr<StreamSocket> &socket, const Poco::Net::HTTPRequest& request); @@ -41,11 +44,12 @@ private: void sendTextFrame(const std::string& message); - bool adminCommandHandler(const std::vector<char>& payload); + /// Process incoming websocket messages + void handleMessage(bool fin, WSOpCode code, std::vector<char> &data); private: Admin* _admin; - std::shared_ptr<LOOLWebSocket> _adminWs; +// std::shared_ptr<LOOLWebSocket> _adminWs; FIXME - this is us now ! int _sessionId; bool _isAuthenticated; }; @@ -55,6 +59,7 @@ class MemoryStatsTask; /// An admin command processor. class Admin { + Admin(); public: virtual ~Admin(); @@ -91,20 +96,12 @@ public: void rescheduleCpuTimer(unsigned interval); - static AdminRequestHandler* createRequestHandler() - { - return new AdminRequestHandler(&instance()); - } - std::unique_lock<std::mutex> getLock() { return std::unique_lock<std::mutex>(_modelMutex); } void updateLastActivityTime(const std::string& docKey); void updateMemoryDirty(const std::string& docKey, int dirty); private: - Admin(); - -private: AdminModel _model; std::mutex _modelMutex; int _forKitPid; diff --git a/wsd/AdminModel.cpp b/wsd/AdminModel.cpp index 3b81ff3..c23641c 100644 --- a/wsd/AdminModel.cpp +++ b/wsd/AdminModel.cpp @@ -21,7 +21,7 @@ #include <Poco/URI.h> #include "Protocol.hpp" -#include <LOOLWebSocket.hpp> +#include "net/WebSocketHandler.hpp" #include "Log.hpp" #include "Unit.hpp" #include "Util.hpp" @@ -70,7 +70,7 @@ bool Subscriber::notify(const std::string& message) try { UnitWSD::get().onAdminNotifyMessage(message); - webSocket->sendFrame(message.data(), message.length()); + webSocket->sendFrame(message); return true; } catch (const std::exception& ex) @@ -156,7 +156,7 @@ unsigned AdminModel::getKitsMemoryUsage() return totalMem; } -void AdminModel::subscribe(int sessionId, std::shared_ptr<LOOLWebSocket>& ws) +void AdminModel::subscribe(int sessionId, const std::weak_ptr<WebSocketHandler>& ws) { const auto ret = _subscribers.emplace(sessionId, Subscriber(sessionId, ws)); if (!ret.second) diff --git a/wsd/AdminModel.hpp b/wsd/AdminModel.hpp index 8bfb719..8250687 100644 --- a/wsd/AdminModel.hpp +++ b/wsd/AdminModel.hpp @@ -17,7 +17,7 @@ #include <Poco/Process.h> #include "Log.hpp" -#include <LOOLWebSocket.hpp> +#include "net/WebSocketHandler.hpp" #include "Util.hpp" /// A client view in Admin controller. @@ -98,7 +98,7 @@ private: class Subscriber { public: - Subscriber(int sessionId, std::shared_ptr<LOOLWebSocket>& ws) + Subscriber(int sessionId, const std::weak_ptr<WebSocketHandler>& ws) : _sessionId(sessionId), _ws(ws), _start(std::time(nullptr)) @@ -125,8 +125,8 @@ private: /// Admin session Id int _sessionId; - /// LOOLWebSocket to use to send messages to session - std::weak_ptr<LOOLWebSocket> _ws; + /// The underlying AdminRequestHandler + std::weak_ptr<WebSocketHandler> _ws; std::set<std::string> _subscriptions; @@ -153,7 +153,7 @@ public: /// Returns memory consumed by all active loolkit processes unsigned getKitsMemoryUsage(); - void subscribe(int sessionId, std::shared_ptr<LOOLWebSocket>& ws); + void subscribe(int sessionId, const std::weak_ptr<WebSocketHandler>& ws); void subscribe(int sessionId, const std::string& command); void unsubscribe(int sessionId, const std::string& command); commit d9980e3392f04ba8f764ca561a300e36f7a130e8 Author: Michael Meeks <[email protected]> Date: Wed Mar 15 14:40:24 2017 +0000 Adjust initial AdminRequestHandler entry point. diff --git a/wsd/Admin.cpp b/wsd/Admin.cpp index 636c403..07723df 100644 --- a/wsd/Admin.cpp +++ b/wsd/Admin.cpp @@ -32,7 +32,7 @@ #include "FileServer.hpp" #include "IoUtil.hpp" #include "Protocol.hpp" -#include "LOOLWebSocket.hpp" +#include "LOOLWebSocket.hpp" // FIXME: remove. #include "LOOLWSD.hpp" #include "Log.hpp" #include "Storage.hpp" @@ -40,6 +40,9 @@ #include "Unit.hpp" #include "Util.hpp" +#include "net/Socket.hpp" +#include "net/WebSocketHandler.hpp" + #include "common/SigUtil.hpp" using namespace LOOLProtocol; @@ -245,14 +248,14 @@ void AdminRequestHandler::sendTextFrame(const std::string& message) _adminWs->sendFrame(message.data(), message.size()); } -void AdminRequestHandler::handleRequest(HTTPServerRequest& request, HTTPServerResponse& response) +void AdminRequestHandler::handleInitialRequest(const std::weak_ptr<StreamSocket> &socketWeak, + const Poco::Net::HTTPRequest& request) { - // Different session id pool for admin sessions (?) - const auto sessionId = Util::decodeId(LOOLWSD::GenSessionId()); - - Util::setThreadName("admin_ws_" + std::to_string(sessionId)); + HTTPResponse response; + auto socket = socketWeak.lock(); - LOG_DBG("Thread started."); + // Different session id pool for admin sessions (?) +// const auto sessionId = Util::decodeId(LOOLWSD::GenSessionId()); try { @@ -261,7 +264,10 @@ void AdminRequestHandler::handleRequest(HTTPServerRequest& request, HTTPServerRe if (request.find("Upgrade") != request.end() && Poco::icompare(request["Upgrade"], "websocket") == 0) { - handleWSRequests(request, response, sessionId); + // First Upgrade. + WebSocketHandler ws(socketWeak, request); + +// handleWSRequests(request, response, sessionId); } } catch(const Poco::Net::NotAuthenticatedException& exc) @@ -270,17 +276,15 @@ void AdminRequestHandler::handleRequest(HTTPServerRequest& request, HTTPServerRe response.set("WWW-Authenticate", "Basic realm=\"online\""); response.setStatusAndReason(HTTPResponse::HTTP_UNAUTHORIZED); response.setContentLength(0); - response.send(); + socket->send(response); } catch (const std::exception& exc) { LOG_INF("Admin::handleRequest: Exception: " << exc.what()); response.setStatusAndReason(HTTPResponse::HTTP_BAD_REQUEST); response.setContentLength(0); - response.send(); + socket->send(response); } - - LOG_DBG("Thread finished."); } /// An admin command processor. diff --git a/wsd/Admin.hpp b/wsd/Admin.hpp index f1cbe86..82b5a5e 100644 --- a/wsd/Admin.hpp +++ b/wsd/Admin.hpp @@ -25,14 +25,16 @@ #include <LOOLWebSocket.hpp> class Admin; +class StreamSocket; /// Admin requests over HTTP(S) handler. -class AdminRequestHandler : public Poco::Net::HTTPRequestHandler +class AdminRequestHandler /// public Poco::Net::HTTPRequestHandler { public: AdminRequestHandler(Admin* adminManager); - void handleRequest(Poco::Net::HTTPServerRequest& request, Poco::Net::HTTPServerResponse& response) override; + static void handleInitialRequest(const std::weak_ptr<StreamSocket> &socket, + const Poco::Net::HTTPRequest& request); private: void handleWSRequests(Poco::Net::HTTPServerRequest& request, Poco::Net::HTTPServerResponse& response, int sessionId); diff --git a/wsd/LOOLWSD.cpp b/wsd/LOOLWSD.cpp index e141bb9..3aee14d 100644 --- a/wsd/LOOLWSD.cpp +++ b/wsd/LOOLWSD.cpp @@ -1807,7 +1807,9 @@ private: // Admin connections else if (reqPathSegs.size() >= 2 && reqPathSegs[0] == "lool" && reqPathSegs[1] == "adminws") { - handleAdminRequest(request); + LOG_ERR("Admin request: " << request.getURI()); + AdminRequestHandler::handleInitialRequest(_socket, request); + } // Client post and websocket connections else if ((request.getMethod() == HTTPRequest::HTTP_GET || @@ -1881,12 +1883,6 @@ private: socket->shutdown(); } - void handleAdminRequest(const Poco::Net::HTTPRequest& request) - { - LOG_ERR("Admin request: " << request.getURI()); - // requestHandler = Admin::createRequestHandler(); - } - void handleRootRequest(const Poco::Net::HTTPRequest& request) { LOG_DBG("HTTP request: " << request.getURI()); _______________________________________________ Libreoffice-commits mailing list [email protected] https://lists.freedesktop.org/mailman/listinfo/libreoffice-commits
