Merge pull request #312 from Palakis/feature/async-handling

Call the request handler asynchronously

CI/install-build-obs-macos.sh

@@ -37,6 +37,7 @@ cmake .. \
 	-DBUILD_CAPTIONS=true \
 	-DCMAKE_OSX_DEPLOYMENT_TARGET=10.11 \
 	-DDISABLE_PLUGINS=true \
+    -DENABLE_SCRIPTING=0 \
 	-DDepsPath=/tmp/obsdeps \
 	-DCMAKE_PREFIX_PATH=/usr/local/opt/qt/lib/cmake \
 && make -j4

CMakeLists.txt

@@ -30,6 +30,7 @@ find_package(Boost REQUIRED)
 set(obs-websocket_SOURCES
 	src/obs-websocket.cpp
 	src/WSServer.cpp
+	src/ConnectionProperties.cpp
 	src/WSRequestHandler.cpp
 	src/WSRequestHandler_General.cpp
 	src/WSRequestHandler_Profiles.cpp
@@ -50,6 +51,7 @@ set(obs-websocket_SOURCES
 set(obs-websocket_HEADERS
 	src/obs-websocket.h
 	src/WSServer.h
+	src/ConnectionProperties.h
 	src/WSRequestHandler.h
 	src/WSEvents.h
 	src/Config.h

src/ConnectionProperties.cpp

@@ -0,0 +1,34 @@
+/*
+obs-websocket
+Copyright (C) 2016-2019	Stéphane Lepin <stephane.lepin@gmail.com>
+
+This program is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along
+with this program. If not, see <https://www.gnu.org/licenses/>
+*/
+
+#include "ConnectionProperties.h"
+
+ConnectionProperties::ConnectionProperties()
+    : _authenticated(false)
+{
+}
+
+bool ConnectionProperties::isAuthenticated()
+{
+    return _authenticated.load();
+}
+
+void ConnectionProperties::setAuthenticated(bool authenticated)
+{
+    _authenticated.store(authenticated);
+}

src/ConnectionProperties.h

@@ -0,0 +1,31 @@
+/*
+obs-websocket
+Copyright (C) 2016-2019	Stéphane Lepin <stephane.lepin@gmail.com>
+
+This program is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation; either version 2 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License along
+with this program. If not, see <https://www.gnu.org/licenses/>
+*/
+
+#pragma once
+
+#include <atomic>
+
+class ConnectionProperties
+{
+public:
+    explicit ConnectionProperties();
+    bool isAuthenticated();
+    void setAuthenticated(bool authenticated);
+private:
+    std::atomic<bool> _authenticated;
+};

src/WSRequestHandler.cpp

@@ -132,7 +132,7 @@ QSet<QString> WSRequestHandler::authNotRequired {
 	"Authenticate"
 };
 
-WSRequestHandler::WSRequestHandler(QVariantHash& connProperties) :
+WSRequestHandler::WSRequestHandler(ConnectionProperties& connProperties) :
 	_messageId(0),
 	_requestType(""),
 	data(nullptr),
@@ -174,7 +174,7 @@ HandlerResponse WSRequestHandler::processRequest(std::string& textMessage){
 
 	if (Config::Current()->AuthRequired
 		&& (!authNotRequired.contains(_requestType))
-		&& (_connProperties.value(PROP_AUTHENTICATED).toBool() == false))
+		&& (!_connProperties.isAuthenticated()))
 	{
 		return SendErrorResponse("Not Authenticated");
 	}

src/WSRequestHandler.h

@@ -23,10 +23,13 @@ with this program. If not, see <https://www.gnu.org/licenses/>
 #include <QtCore/QSet>
 #include <QtCore/QVariantHash>
 #include <QtCore/QString>
+#include <QtCore/QSharedPointer>
 
 #include <obs.hpp>
 #include <obs-frontend-api.h>
 
+#include "ConnectionProperties.h"
+
 #include "obs-websocket.h"
 
 typedef obs_data_t* HandlerResponse;
@@ -35,7 +38,7 @@ class WSRequestHandler : public QObject {
 	Q_OBJECT
 
 	public:
-		explicit WSRequestHandler(QVariantHash& connProperties);
+		explicit WSRequestHandler(ConnectionProperties& connProperties);
 		~WSRequestHandler();
 		std::string processIncomingMessage(std::string& textMessage);
 		bool hasField(QString name);
@@ -43,7 +46,7 @@ class WSRequestHandler : public QObject {
 	private:
 		const char* _messageId;
 		const char* _requestType;
-		QVariantHash& _connProperties;
+		ConnectionProperties& _connProperties;
 		OBSDataAutoRelease data;
 
 		HandlerResponse processRequest(std::string& textMessage);

src/WSRequestHandler_General.cpp

@@ -82,7 +82,7 @@ HandlerResponse WSRequestHandler::HandleAuthenticate(WSRequestHandler* req) {
 		return req->SendErrorResponse("missing request parameters");
 	}
 
-	if (req->_connProperties.value(PROP_AUTHENTICATED).toBool() == true) {
+	if (req->_connProperties.isAuthenticated()) {
 		return req->SendErrorResponse("already authenticated");
 	}
 
@@ -95,7 +95,7 @@ HandlerResponse WSRequestHandler::HandleAuthenticate(WSRequestHandler* req) {
 		return req->SendErrorResponse("Authentication Failed.");
 	}
 
-	req->_connProperties.insert(PROP_AUTHENTICATED, true);
+	req->_connProperties.setAuthenticated(true);
 	return req->SendOKResponse();
 }
 

src/WSServer.cpp

@@ -25,6 +25,7 @@ with this program. If not, see <https://www.gnu.org/licenses/>
 #include <QtWidgets/QMessageBox>
 #include <QtConcurrent/QtConcurrent>
 #include <obs-frontend-api.h>
+#include <util/platform.h>
 
 #include "WSServer.h"
 #include "obs-websocket.h"
@@ -127,6 +128,8 @@ void WSServer::stop()
 	_connections.clear();
 	_connectionProperties.clear();
 	
+	_threadPool.waitForDone();
+
 	while (!_server.stopped()) {
 		std::this_thread::sleep_for(std::chrono::milliseconds(1));
 	}
@@ -139,7 +142,7 @@ void WSServer::broadcast(std::string message)
 	QMutexLocker locker(&_clMutex);
 	for (connection_hdl hdl : _connections) {
 		if (Config::Current()->AuthRequired) {
-			bool authenticated = _connectionProperties[hdl].value(PROP_AUTHENTICATED).toBool();
+			bool authenticated = _connectionProperties[hdl].isAuthenticated();
 			if (!authenticated) {
 				continue;
 			}
@@ -166,24 +169,18 @@ void WSServer::onMessage(connection_hdl hdl, server::message_ptr message)
 		return;
 	}
 
-	std::string payload = message->get_payload();
+	QtConcurrent::run(&_threadPool, [=]() {
+		std::string payload = message->get_payload();
 
-	QMutexLocker locker(&_clMutex);
-	QVariantHash connProperties = _connectionProperties[hdl];
-	locker.unlock();
+		QMutexLocker locker(&_clMutex);
+		ConnectionProperties& connProperties = _connectionProperties[hdl];
+		locker.unlock();
 
-	WSRequestHandler handler(connProperties);
-	std::string response = handler.processIncomingMessage(payload);
+		WSRequestHandler handler(connProperties);
+		std::string response = handler.processIncomingMessage(payload);
 
-	_server.send(hdl, response, websocketpp::frame::opcode::text);
-
-	locker.relock();
-	// In multithreaded processing this would be problematic to put back
-	// a copy of the connection properties, because there might conflicts
-	// between several simultaneous handlers.
-	// In our case, it's fine because all messages are processed in one thread.
-	_connectionProperties[hdl] = connProperties;
-	locker.unlock();
+		_server.send(hdl, response, websocketpp::frame::opcode::text);
+	});
 }
 
 void WSServer::onClose(connection_hdl hdl)

src/WSServer.h

@@ -18,17 +18,19 @@ with this program. If not, see <https://www.gnu.org/licenses/>
 
 #pragma once
 
+#include <map>
+#include <set>
 #include <QtCore/QObject>
 #include <QtCore/QMutex>
 #include <QtCore/QSharedPointer>
 #include <QtCore/QVariantHash>
-
-#include <map>
-#include <set>
+#include <QtCore/QThreadPool>
 
 #include <websocketpp/config/asio_no_tls.hpp>
 #include <websocketpp/server.hpp>
 
+#include "ConnectionProperties.h"
+
 #include "WSRequestHandler.h"
 
 QT_FORWARD_DECLARE_CLASS(QWebSocketServer)
@@ -54,6 +56,9 @@ public:
 	void start(quint16 port);
 	void stop();
 	void broadcast(std::string message);
+	QThreadPool* threadPool() {
+		return &_threadPool;
+	}
 
 private:
 	static WSServerPtr _instance;
@@ -69,6 +74,7 @@ private:
 	server _server;
 	quint16 _serverPort;
 	std::set<connection_hdl, std::owner_less<connection_hdl>> _connections;
-	std::map<connection_hdl, QVariantHash, std::owner_less<connection_hdl>> _connectionProperties;
+	std::map<connection_hdl, ConnectionProperties, std::owner_less<connection_hdl>> _connectionProperties;
 	QMutex _clMutex;
+	QThreadPool _threadPool;
 };

src/obs-websocket.h

@@ -37,7 +37,6 @@ using OBSDataArrayAutoRelease =
 using OBSOutputAutoRelease =
 	OBSRef<obs_output_t*, ___output_dummy_addref, obs_output_release>;
 
-#define PROP_AUTHENTICATED "wsclient_authenticated"
 #define OBS_WEBSOCKET_VERSION "4.6.0"
 
 #define blog(level, msg, ...) blog(level, "[obs-websocket] " msg, ##__VA_ARGS__)