Merge pull request #692 from Palakis/fix/obs-shutdown-crash

Events: Fix multiple shutdown crashes

src/WSEvents.cpp

@@ -249,6 +249,9 @@ void WSEvents::FrontendEventHandler(enum obs_frontend_event event, void* private
 void WSEvents::broadcastUpdate(const char* updateType,
 	obs_data_t* additionalFields = nullptr)
 {
+	if (!_srv->isListening()) {
+		return;
+	}
 	std::optional<uint64_t> streamTime;
 	if (obs_frontend_streaming_active()) {
 		streamTime = std::make_optional(getStreamingTime());

src/WSServer.cpp

@@ -60,6 +60,21 @@ WSServer::~WSServer()
 	stop();
 }
 
+void WSServer::serverRunner()
+{
+	blog(LOG_INFO, "IO thread started.");
+	try {
+		_server.run();
+	} catch (websocketpp::exception const & e) {
+		blog(LOG_ERROR, "websocketpp instance returned an error: %s", e.what());
+	} catch (const std::exception & e) {
+		blog(LOG_ERROR, "websocketpp instance returned an error: %s", e.what());
+	} catch (...) {
+		blog(LOG_ERROR, "websocketpp instance returned an error");
+	}
+	blog(LOG_INFO, "IO thread exited.");
+}
+
 void WSServer::start(quint16 port, bool lockToIPv4)
 {
 	if (_server.is_listening() && (port == _serverPort && _lockToIPv4 == lockToIPv4)) {
@@ -102,11 +117,7 @@ void WSServer::start(quint16 port, bool lockToIPv4)
 
 	_server.start_accept();
 
-	QtConcurrent::run([=]() {
+	_serverThread = std::thread(&WSServer::serverRunner, this);
-		blog(LOG_INFO, "io thread started");
-		_server.run();
-		blog(LOG_INFO, "io thread exited");
-	});
 
 	blog(LOG_INFO, "server started successfully on port %d", _serverPort);
 }
@@ -119,7 +130,18 @@ void WSServer::stop()
 
 	_server.stop_listening();
 	for (connection_hdl hdl : _connections) {
-		_server.close(hdl, websocketpp::close::status::going_away, "Server stopping");
+		websocketpp::lib::error_code errorCode;
+		_server.pause_reading(hdl, errorCode);
+		if (errorCode) {
+			blog(LOG_ERROR, "Error: %s", errorCode.message().c_str());
+			continue;
+		}
+
+		_server.close(hdl, websocketpp::close::status::going_away, "Server stopping", errorCode);
+		if (errorCode) {
+			blog(LOG_ERROR, "Error: %s", errorCode.message().c_str());
+			continue;
+		}
 	}
 
 	_threadPool.waitForDone();
@@ -128,6 +150,8 @@ void WSServer::stop()
 		std::this_thread::sleep_for(std::chrono::milliseconds(10));
 	}
 
+	_serverThread.join();
+
 	blog(LOG_INFO, "server stopped successfully");
 }
 
@@ -160,6 +184,11 @@ void WSServer::broadcast(const RpcEvent& event)
 	}
 }
 
+bool WSServer::isListening()
+{
+	return _server.is_listening();
+}
+
 void WSServer::onOpen(connection_hdl hdl)
 {
 	QMutexLocker locker(&_clMutex);
@@ -217,11 +246,12 @@ void WSServer::onClose(connection_hdl hdl)
 
 	auto conn = _server.get_con_from_hdl(hdl);
 	auto localCloseCode = conn->get_local_close_code();
+	auto localCloseReason = conn->get_local_close_reason();
+	QString clientIp = getRemoteEndpoint(hdl);
 
-	if (localCloseCode != websocketpp::close::status::going_away) {
+	blog(LOG_INFO, "Websocket connection with client '%s' closed (disconnected). Code is %d, reason is: '%s'", clientIp.toUtf8().constData(), localCloseCode, localCloseReason.c_str());
-		QString clientIp = getRemoteEndpoint(hdl);
+	if (localCloseCode != websocketpp::close::status::going_away && _server.is_listening()) {
 		notifyDisconnection(clientIp);
-		blog(LOG_INFO, "client %s disconnected", clientIp.toUtf8().constData());
 	}
 }
 

src/WSServer.h

@@ -47,11 +47,14 @@ public:
 	void start(quint16 port, bool lockToIPv4);
 	void stop();
 	void broadcast(const RpcEvent& event);
+	bool isListening();
 	QThreadPool* threadPool() {
 		return &_threadPool;
 	}
 
 private:
+	void serverRunner();
+
 	void onOpen(connection_hdl hdl);
 	void onMessage(connection_hdl hdl, server::message_ptr message);
 	void onClose(connection_hdl hdl);
@@ -60,6 +63,7 @@ private:
 	void notifyConnection(QString clientIp);
 	void notifyDisconnection(QString clientIp);
 
+	std::thread _serverThread;
 	server _server;
 	quint16 _serverPort;
 	bool _lockToIPv4;