base: Move some direct crosstalk to callback system

src/eventhandler/EventHandler.cpp

@@ -73,19 +73,10 @@ EventHandler::~EventHandler()
 	blog_debug("[EventHandler::~EventHandler] Finished.");
 }
 
-void EventHandler::SetBroadcastCallback(EventHandler::BroadcastCallback cb)
+// Function to increment or decrement refcounts for high volume event subscriptions
-{
+void EventHandler::ProcessSubscriptionChange(bool type, uint64_t eventSubscriptions)
-	_broadcastCallback = cb;
-}
-
-void EventHandler::SetObsReadyCallback(EventHandler::ObsReadyCallback cb)
-{
-	_obsReadyCallback = cb;
-}
-
-// Function to increment refcounts for high volume event subscriptions
-void EventHandler::ProcessSubscription(uint64_t eventSubscriptions)
 {
+	if (type) {
 		if ((eventSubscriptions & EventSubscription::InputVolumeMeters) != 0) {
 			if (_inputVolumeMetersRef.fetch_add(1) == 0) {
 				if (_inputVolumeMetersHandler)
@@ -101,11 +92,7 @@ void EventHandler::ProcessSubscription(uint64_t eventSubscriptions)
 			_inputShowStateChangedRef++;
 		if ((eventSubscriptions & EventSubscription::SceneItemTransformChanged) != 0)
 			_sceneItemTransformChangedRef++;
-}
+	} else {
-
-// Function to decrement refcounts for high volume event subscriptions
-void EventHandler::ProcessUnsubscription(uint64_t eventSubscriptions)
-{
 		if ((eventSubscriptions & EventSubscription::InputVolumeMeters) != 0) {
 			if (_inputVolumeMetersRef.fetch_sub(1) == 1)
 				_inputVolumeMetersHandler.reset();
@@ -116,6 +103,7 @@ void EventHandler::ProcessUnsubscription(uint64_t eventSubscriptions)
 			_inputShowStateChangedRef--;
 		if ((eventSubscriptions & EventSubscription::SceneItemTransformChanged) != 0)
 			_sceneItemTransformChangedRef--;
+	}
 }
 
 // Function required in order to use default arguments

src/eventhandler/EventHandler.h

@@ -36,12 +36,18 @@ public:
 
 	typedef std::function<void(uint64_t, std::string, json, uint8_t)>
 		BroadcastCallback; // uint64_t requiredIntent, std::string eventType, json eventData, uint8_t rpcVersion
-	void SetBroadcastCallback(BroadcastCallback cb);
+	inline void SetBroadcastCallback(BroadcastCallback cb)
-	typedef std::function<void(bool)> ObsReadyCallback; // bool ready
+	{
-	void SetObsReadyCallback(ObsReadyCallback cb);
+		_broadcastCallback = cb;
+	}
 
-	void ProcessSubscription(uint64_t eventSubscriptions);
+	typedef std::function<void(bool)> ObsReadyCallback; // bool ready
-	void ProcessUnsubscription(uint64_t eventSubscriptions);
+	inline void SetObsReadyCallback(ObsReadyCallback cb)
+	{
+		_obsReadyCallback = cb;
+	}
+
+	void ProcessSubscriptionChange(bool type, uint64_t eventSubscriptions);
 
 private:
 	BroadcastCallback _broadcastCallback;

src/obs-websocket.cpp

@@ -81,6 +81,11 @@ bool obs_module_load(void)
 	// Initialize the WebSocket server
 	_webSocketServer = std::make_shared<WebSocketServer>();
 
+	// Attach event handlers between WebSocket server and event handler
+	_eventHandler->SetBroadcastCallback(std::bind(&WebSocketServer::BroadcastEvent, _webSocketServer.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
+	_eventHandler->SetObsReadyCallback(std::bind(&WebSocketServer::SetObsReady, _webSocketServer.get(), std::placeholders::_1));
+	_webSocketServer->SetClientSubscriptionCallback(std::bind(&EventHandler::ProcessSubscriptionChange, _eventHandler.get(), std::placeholders::_1, std::placeholders::_2));
+
 	// Initialize the settings dialog
 	obs_frontend_push_ui_translation(obs_module_get_string);
 	QMainWindow *mainWindow = static_cast<QMainWindow *>(obs_frontend_get_main_window());
@@ -123,6 +128,11 @@ void obs_module_unload(void)
 		_webSocketServer->Stop();
 	}
 
+	// Disconnect event handler from WebSocket server
+	_eventHandler->SetObsReadyCallback(nullptr);
+	_eventHandler->SetBroadcastCallback(nullptr);
+	_webSocketServer->SetClientSubscriptionCallback(nullptr);
+
 	// Release the WebSocket server
 	_webSocketServer = nullptr;
 

src/websocketserver/WebSocketServer.cpp

@@ -24,7 +24,6 @@ with this program. If not, see <https://www.gnu.org/licenses/>
 #include <obs-frontend-api.h>
 
 #include "WebSocketServer.h"
-#include "../eventhandler/EventHandler.h"
 #include "../obs-websocket.h"
 #include "../Config.h"
 #include "../utils/Crypto.h"
@@ -47,23 +46,10 @@ WebSocketServer::WebSocketServer() : QObject(nullptr)
 	_server.set_close_handler(websocketpp::lib::bind(&WebSocketServer::onClose, this, websocketpp::lib::placeholders::_1));
 	_server.set_message_handler(websocketpp::lib::bind(&WebSocketServer::onMessage, this, websocketpp::lib::placeholders::_1,
 							   websocketpp::lib::placeholders::_2));
-
-	auto eventHandler = GetEventHandler();
-	if (eventHandler) {
-		eventHandler->SetBroadcastCallback(std::bind(&WebSocketServer::BroadcastEvent, this, std::placeholders::_1,
-							     std::placeholders::_2, std::placeholders::_3, std::placeholders::_4));
-		eventHandler->SetObsReadyCallback(std::bind(&WebSocketServer::onObsReady, this, std::placeholders::_1));
-	}
 }
 
 WebSocketServer::~WebSocketServer()
 {
-	auto eventHandler = GetEventHandler();
-	if (eventHandler) {
-		eventHandler->SetObsReadyCallback(nullptr);
-		eventHandler->SetBroadcastCallback(nullptr);
-	}
-
 	if (_server.is_listening())
 		Stop();
 }
@@ -215,7 +201,7 @@ std::vector<WebSocketServer::WebSocketSessionState> WebSocketServer::GetWebSocke
 	return webSocketSessions;
 }
 
-void WebSocketServer::onObsReady(bool ready)
+void WebSocketServer::SetObsReady(bool ready)
 {
 	_obsReady = ready;
 }
@@ -327,11 +313,9 @@ void WebSocketServer::onClose(websocketpp::connection_hdl hdl)
 	_sessions.erase(hdl);
 	lock.unlock();
 
-	// If client was identified, decrement appropriate refs in eventhandler.
+	// If client was identified, announce unsubscription
-	if (isIdentified) {
+	if (isIdentified && _clientSubscriptionCallback)
-		auto eventHandler = GetEventHandler();
+		_clientSubscriptionCallback(false, eventSubscriptions);
-		eventHandler->ProcessUnsubscription(eventSubscriptions);
-	}
 
 	// Build SessionState object for signal
 	WebSocketSessionState state;

src/websocketserver/WebSocketServer.h

@@ -57,12 +57,20 @@ public:
 	void InvalidateSession(websocketpp::connection_hdl hdl);
 	void BroadcastEvent(uint64_t requiredIntent, const std::string &eventType, const json &eventData = nullptr,
 			    uint8_t rpcVersion = 0);
+	void SetObsReady(bool ready);
 
-	bool IsListening() { return _server.is_listening(); }
+	// Callback for when a client subscribes or unsubscribes. `true` for sub, `false` for unsub
+	typedef std::function<void(bool, uint64_t)> ClientSubscriptionCallback; // bool type, uint64_t eventSubscriptions
+	inline void SetClientSubscriptionCallback(ClientSubscriptionCallback cb)
+	{
+		_clientSubscriptionCallback = cb;
+	}
+
+	inline bool IsListening() { return _server.is_listening(); }
 
 	std::vector<WebSocketSessionState> GetWebSocketSessions();
 
-	QThreadPool *GetThreadPool() { return &_threadPool; }
+	inline QThreadPool *GetThreadPool() { return &_threadPool; }
 
 signals:
 	void ClientConnected(WebSocketSessionState state);
@@ -77,7 +85,6 @@ private:
 
 	void ServerRunner();
 
-	void onObsReady(bool loaded);
 	bool onValidate(websocketpp::connection_hdl hdl);
 	void onOpen(websocketpp::connection_hdl hdl);
 	void onClose(websocketpp::connection_hdl hdl);
@@ -98,4 +105,6 @@ private:
 	std::map<websocketpp::connection_hdl, SessionPtr, std::owner_less<websocketpp::connection_hdl>> _sessions;
 
 	std::atomic<bool> _obsReady = false;
+
+	ClientSubscriptionCallback _clientSubscriptionCallback;
 };

src/websocketserver/WebSocketServer_Protocol.cpp

@@ -23,7 +23,6 @@ with this program. If not, see <https://www.gnu.org/licenses/>
 #include "WebSocketServer.h"
 #include "../requesthandler/RequestHandler.h"
 #include "../requesthandler/RequestBatchHandler.h"
-#include "../eventhandler/EventHandler.h"
 #include "../obs-websocket.h"
 #include "../Config.h"
 #include "../utils/Crypto.h"
@@ -149,9 +148,9 @@ void WebSocketServer::ProcessMessage(SessionPtr session, WebSocketServer::Proces
 		if (ret.closeCode != WebSocketCloseCode::DontClose)
 			return;
 
-		// Increment refs for event subscriptions
+		// Announce subscribe
-		auto eventHandler = GetEventHandler();
+		if (_clientSubscriptionCallback)
-		eventHandler->ProcessSubscription(session->EventSubscriptions());
+			_clientSubscriptionCallback(true, session->EventSubscriptions());
 
 		// Mark session as identified
 		session->SetIsIdentified(true);
@@ -172,16 +171,17 @@ void WebSocketServer::ProcessMessage(SessionPtr session, WebSocketServer::Proces
 	case WebSocketOpCode::Reidentify: { // Reidentify
 		std::unique_lock<std::mutex> sessionLock(session->OperationMutex);
 
-		// Decrement refs for current subscriptions
+		// Announce unsubscribe
-		auto eventHandler = GetEventHandler();
+		if (_clientSubscriptionCallback)
-		eventHandler->ProcessUnsubscription(session->EventSubscriptions());
+			_clientSubscriptionCallback(false, session->EventSubscriptions());
 
 		SetSessionParameters(session, ret, payloadData);
 		if (ret.closeCode != WebSocketCloseCode::DontClose)
 			return;
 
-		// Increment refs for new subscriptions
+		// Announce subscribe
-		eventHandler->ProcessSubscription(session->EventSubscriptions());
+		if (_clientSubscriptionCallback)
+			_clientSubscriptionCallback(true, session->EventSubscriptions());
 
 		ret.result["op"] = WebSocketOpCode::Identified;
 		ret.result["d"]["negotiatedRpcVersion"] = session->RpcVersion();