openclaw/deskflow
1
0
Fork 0
Code Issues Pull Requests Actions 17 Packages Projects Releases Wiki Activity

feat: Restart process from daemon through new Qt IPC impl

Browse Source
This commit is contained in:
Nick Bolton
2025-02-06 13:31:18 +00:00
parent 0d2d9f385d
commit c1d7474700
10 changed files with 212 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
cspell.json
View File

@ -59,6 +59,7 @@
"Priddy",
"psutil",
"pyproject",
"qobject",
"qputenv",
"readf",
"Regen",

2
src/apps/deskflow-gui/MainWindow.cpp
View File

@ -169,8 +169,6 @@ MainWindow::MainWindow(ConfigScopes &configScopes, AppConfig &appConfig)
regenerateLocalFingerprints();
}
}
m_daemonIpcClient->connect();
}
MainWindow::~MainWindow()

30
src/lib/deskflow/DaemonApp.cpp
View File

@ -101,6 +101,10 @@ DaemonApp::DaemonApp(IEventQueue *events, int argc, char **argv)
m_events(events),
m_ipcServer{new ipc::DaemonIpcServer(this)}
{
connect(m_ipcServer, &ipc::DaemonIpcServer::elevateModeChanged, this, &DaemonApp::handleElevateModeChanged);
connect(m_ipcServer, &ipc::DaemonIpcServer::commandChanged, this, &DaemonApp::handleCommandChanged);
connect(m_ipcServer, &ipc::DaemonIpcServer::restartRequested, this, &DaemonApp::handleRestartRequested);
s_instance = this;
}
@ -109,6 +113,28 @@ DaemonApp::~DaemonApp()
s_instance = nullptr;
}
void DaemonApp::handleElevateModeChanged(int mode)
{
LOG_DEBUG("elevate mode changed: %d", mode);
m_elevateMode = mode;
}
void DaemonApp::handleCommandChanged(const QString &command)
{
LOG_DEBUG("service command updated");
m_command = command.toStdString();
}
void DaemonApp::handleRestartRequested()
{
LOG_DEBUG("service restart requested");
#if SYSAPI_WIN32
m_watchdog->setCommand(m_command, m_elevateMode);
#else
LOG_ERR("restart not implemented on this platform");
#endif
}
void DaemonApp::startAsync()
{
#if SYSAPI_WIN32
@ -205,7 +231,7 @@ void DaemonApp::mainLoop(bool logToFile, bool foreground)
std::string command = ARCH->setting("Command");
bool elevate = ARCH->setting("Elevate") == "1";
if (command != "") {
LOG((CLOG_INFO "using last known command: %s", command.c_str()));
LOG_DEBUG("using last known command: %s", command.c_str());
m_watchdog->setCommand(command, elevate);
}
@ -217,8 +243,6 @@ void DaemonApp::mainLoop(bool logToFile, bool foreground)
m_watchdog->stop();
#endif
m_events->removeHandler(m_events->forIpcServer().messageReceived(), m_ipcServer.get());
CLOG->remove(m_ipcLogOutputter.get());
DAEMON_RUNNING(false);

11
src/lib/deskflow/DaemonApp.h
View File

@ -13,6 +13,8 @@
#include <QCoreApplication>
#include "common/common.h"
class Event;
class IEventQueue;
class IpcLogOutputter;
@ -44,6 +46,11 @@ private:
std::string logFilename();
void handleIpcMessage(const Event &, void *);
private slots:
void handleElevateModeChanged(int mode);
void handleCommandChanged(const QString &command);
void handleRestartRequested();
public:
static DaemonApp *s_instance;
@ -55,5 +62,7 @@ private:
std::unique_ptr<IpcLogOutputter> m_ipcLogOutputter;
IEventQueue *m_events = nullptr;
FileLogOutputter *m_fileLogOutputter = nullptr;
std::unique_ptr<deskflow::core::ipc::DaemonIpcServer> m_ipcServer;
deskflow::core::ipc::DaemonIpcServer *m_ipcServer = nullptr;
std::string m_command = "";
int m_elevateMode = 0;
};

99
src/lib/deskflow/ipc/DaemonIpcServer.cpp
View File

@ -42,28 +42,93 @@ void DaemonIpcServer::handleNewConnection()
}
LOG_DEBUG("ipc server got new connection");
m_clients.insert(clientSocket);
connect(clientSocket, &QLocalSocket::readyRead, this, [clientSocket]() {
LOG_DEBUG("ipc server ready to read data");
connect(clientSocket, &QLocalSocket::readyRead, this, &DaemonIpcServer::handleReadyRead);
connect(clientSocket, &QLocalSocket::disconnected, this, &DaemonIpcServer::handleDisconnected);
connect(clientSocket, &QLocalSocket::errorOccurred, this, &DaemonIpcServer::handleErrorOccurred);
}
QByteArray data = clientSocket->readAll();
if (data.isEmpty()) {
LOG_WARN("ipc server got empty message");
void DaemonIpcServer::handleReadyRead()
{
const auto clientSocket = qobject_cast<QLocalSocket *>(sender());
LOG_DEBUG("ipc server ready to read data");
QByteArray data = clientSocket->readAll();
if (data.isEmpty()) {
LOG_WARN("ipc server got empty message");
return;
}
// we don't handle incomplete messages yet; each socket read must have delimiters.
if (!data.contains('\n')) {
LOG_WARN("ipc server got incomplete message: %s", data.constData());
return;
}
// each message is delimited by a newline to keep the protocol super simple.
while (data.contains('\n')) {
int index = data.indexOf('\n');
QByteArray messageData = data.left(index);
data.remove(0, index + 1);
QString message = QString::fromUtf8(messageData);
processMessage(clientSocket, message);
}
}
void DaemonIpcServer::handleDisconnected()
{
const auto clientSocket = qobject_cast<QLocalSocket *>(sender());
LOG_DEBUG("ipc server client disconnected");
m_clients.remove(clientSocket);
clientSocket->deleteLater();
}
void DaemonIpcServer::handleErrorOccurred()
{
const auto clientSocket = qobject_cast<QLocalSocket *>(sender());
LOG_ERR("ipc server client error: %s", clientSocket->errorString().toUtf8().constData());
m_clients.remove(clientSocket);
clientSocket->deleteLater();
}
void DaemonIpcServer::processMessage(QLocalSocket *clientSocket, const QString &message)
{
const auto kAckMessage = "ok\n";
const auto kErrorMessage = "error\n";
LOG_DEBUG("ipc server got message: %s", message.toUtf8().constData());
if (message == "hello") {
clientSocket->write("hello\n");
} else if (message.startsWith("elevate=")) {
const auto elevateMode = message.split('=')[1].toInt();
if (elevateMode < 0 || elevateMode > 2) {
LOG_ERR("ipc server got invalid elevate mode: %d", elevateMode);
clientSocket->write(kErrorMessage);
return;
} else {
LOG_DEBUG("ipc server got new elevate mode: %d", elevateMode);
Q_SIGNAL elevateModeChanged(elevateMode);
clientSocket->write(kAckMessage);
}
QString dataStr = QString::fromUtf8(data);
if (dataStr.isEmpty()) {
LOG_ERR("ipc server failed to convert message to string");
return;
} else if (message.startsWith("command=")) {
const auto command = message.split('=')[1];
if (command.isEmpty()) {
LOG_ERR("ipc server got empty command");
clientSocket->write(kErrorMessage);
} else {
LOG_DEBUG("ipc server got new command: %s", command.toUtf8().constData());
Q_SIGNAL commandChanged(command);
clientSocket->write(kAckMessage);
}
if (dataStr == "hello") {
LOG_DEBUG("ipc server got message: %s", data.constData());
clientSocket->write("hello");
clientSocket->flush();
}
});
} else if (message == "restart") {
LOG_DEBUG("ipc server got restart message");
Q_SIGNAL restartRequested();
clientSocket->write(kAckMessage);
} else {
LOG_WARN("ipc server got unknown message: %s", message.toUtf8().constData());
}
clientSocket->flush();
}
} // namespace deskflow::core::ipc

14
src/lib/deskflow/ipc/DaemonIpcServer.h
View File

@ -7,8 +7,10 @@
#pragma once
#include <QObject>
#include <QSet>
class QLocalServer;
class QLocalSocket;
namespace deskflow::core::ipc {
@ -20,11 +22,23 @@ public:
DaemonIpcServer(QObject *parent);
~DaemonIpcServer();
signals:
void elevateModeChanged(int mode);
void commandChanged(const QString &command);
void restartRequested();
private:
void processMessage(QLocalSocket *clientSocket, const QString &message);
private slots:
void handleNewConnection();
void handleReadyRead();
void handleDisconnected();
void handleErrorOccurred();
private:
QLocalServer *m_server;
QSet<QLocalSocket *> m_clients;
};
} // namespace deskflow::core::ipc

33
src/lib/gui/core/CoreProcess.cpp
View File

@ -9,9 +9,9 @@
#include "common/constants.h"
#include "gui/config/IAppConfig.h"
#include "gui/core/CoreTool.h"
#include "gui/ipc/DaemonIpcClient.h"
#include "gui/paths.h"
#include "tls/TlsUtility.h"
#include <qlogging.h>
#if defined(Q_OS_MAC)
#include "OSXHelpers.h"
@ -25,7 +25,6 @@
#include <QRegularExpression>
#include <QStandardPaths>
#include <QTimer>
#include <QtGlobal>
namespace deskflow::gui {
@ -156,19 +155,9 @@ QString CoreProcess::Deps::getProfileRoot() const
CoreProcess::CoreProcess(const IAppConfig &appConfig, const IServerConfig &serverConfig, std::shared_ptr<Deps> deps)
: m_appConfig(appConfig),
m_serverConfig(serverConfig),
m_pDeps(deps)
m_pDeps(deps),
m_daemonIpcClient{new ipc::DaemonIpcClient(this)}
{
connect(
&m_pDeps->ipcClient(), &QIpcClient::read, this, //
&CoreProcess::onIpcClientRead
);
connect(
&m_pDeps->ipcClient(), &QIpcClient::serviceReady, this, //
&CoreProcess::onIpcClientServiceReady
);
connect(&m_pDeps->process(), &QProcessProxy::finished, this, &CoreProcess::onProcessFinished);
connect(
@ -186,6 +175,8 @@ CoreProcess::CoreProcess(const IAppConfig &appConfig, const IServerConfig &serve
qDebug("retry cancelled, process state is not retry pending");
}
});
m_daemonIpcClient->connectToServer();
}
void CoreProcess::onIpcClientServiceReady()
@ -205,12 +196,6 @@ void CoreProcess::onIpcClientServiceReady()
void CoreProcess::onIpcClientError(const QString &text) const
{
qCritical().noquote() << text;
if (m_appConfig.processMode() != ProcessMode::kService) {
// if not meant to be in service mode and there is an ipc connection error,
// then abandon the ipc client connection.
m_pDeps->ipcClient().disconnectFromHost();
}
}
void CoreProcess::onIpcClientRead(const QString &text)
@ -290,17 +275,15 @@ void CoreProcess::startProcessFromDaemon(const QString &app, const QStringList &
qFatal("core process must be in starting state");
}
if (!m_pDeps->ipcClient().isConnected()) {
// when service state changes, start will be called again.
qDebug("cannot start process, ipc not connected, connecting instead");
m_pDeps->ipcClient().connectToHost();
if (!m_daemonIpcClient->isConnected()) {
qFatal("cannot start process, daemon ipc not connected");
return;
}
QString commandQuoted = makeQuotedArgs(app, args);
qInfo("running command: %s", qPrintable(commandQuoted));
m_pDeps->ipcClient().sendCommand(commandQuoted, m_appConfig.elevateMode());
m_daemonIpcClient->sendCommand(commandQuoted, m_appConfig.elevateMode());
setProcessState(Started);
}

5
src/lib/gui/core/CoreProcess.h
View File

@ -20,6 +20,10 @@
namespace deskflow::gui {
namespace ipc {
class DaemonIpcClient;
}
class CoreProcess : public QObject
{
using IServerConfig = deskflow::gui::IServerConfig;
@ -170,6 +174,7 @@ private:
std::optional<ProcessMode> m_lastProcessMode = std::nullopt;
QTimer m_retryTimer;
int m_connections = 0;
deskflow::gui::ipc::DaemonIpcClient *m_daemonIpcClient = nullptr;
};
} // namespace deskflow::gui

58
src/lib/gui/ipc/DaemonIpcClient.cpp
View File

@ -16,7 +16,7 @@ namespace deskflow::gui::ipc {
const auto kTimeout = 1000;
DaemonIpcClient::DaemonIpcClient(QObject *parent) : QObject(parent), socket{new QLocalSocket(this)}
DaemonIpcClient::DaemonIpcClient(QObject *parent) : QObject(parent), m_socket{new QLocalSocket(this)}
{
}
@ -24,26 +24,62 @@ DaemonIpcClient::~DaemonIpcClient()
{
}
void DaemonIpcClient::connect()
void DaemonIpcClient::connectToServer()
{
socket->connectToServer(kDaemonIpcName);
if (!socket->waitForConnected(kTimeout)) {
m_socket->connectToServer(kDaemonIpcName);
if (!m_socket->waitForConnected(kTimeout)) {
qCritical() << "ipc client failed to connect to server:" << kDaemonIpcName;
return;
}
socket->write("hello");
if (!socket->waitForBytesWritten(kTimeout)) {
qCritical() << "ipc client failed to write message";
sendMessage("hello", "hello", false);
connect(m_socket, &QLocalSocket::disconnected, this, &DaemonIpcClient::handleDisconnected);
connect(m_socket, &QLocalSocket::errorOccurred, this, &DaemonIpcClient::handleErrorOccurred);
m_connected = true;
qInfo() << "ipc client connected to server:" << kDaemonIpcName;
}
void DaemonIpcClient::handleDisconnected()
{
qInfo() << "ipc client disconnected from server";
m_connected = false;
}
void DaemonIpcClient::handleErrorOccurred()
{
qCritical() << "ipc client error:" << m_socket->errorString();
m_connected = false;
}
void DaemonIpcClient::sendCommand(const QString &command, ElevateMode elevateMode)
{
sendMessage("elevate=" + QString::number(static_cast<int>(elevateMode)));
sendMessage("command=" + command);
sendMessage("restart");
}
void DaemonIpcClient::sendMessage(const QString &message, const QString &expectAck, const bool expectConnected)
{
if (expectConnected && !m_connected) {
qCritical() << "cannot send command, ipc not connected";
return;
}
if (!socket->waitForReadyRead(kTimeout)) {
QByteArray messageData = message.toUtf8() + "\n";
m_socket->write(messageData);
if (!m_socket->waitForBytesWritten(kTimeout)) {
qCritical() << "ipc client failed to write command";
return;
}
if (!m_socket->waitForReadyRead(kTimeout)) {
qCritical() << "ipc client failed to read response";
return;
}
QByteArray response = socket->readAll();
QByteArray response = m_socket->readAll();
if (response.isEmpty()) {
qCritical() << "ipc client got empty response";
return;
@ -55,12 +91,12 @@ void DaemonIpcClient::connect()
return;
}
if (responseData != "hello") {
if (responseData != expectAck + "\n") {
qCritical() << "ipc client got unexpected response: " << responseData;
return;
}
qInfo() << "ipc client connected to server:" << kDaemonIpcName;
qInfo() << "ipc client sent message: " << messageData;
}
} // namespace deskflow::gui::ipc

20
src/lib/gui/ipc/DaemonIpcClient.h
View File

@ -8,6 +8,8 @@
#include <QObject>
#include "gui/config/ElevateMode.h"
class QLocalSocket;
namespace deskflow::gui::ipc {
@ -19,10 +21,24 @@ class DaemonIpcClient : public QObject
public:
DaemonIpcClient(QObject *parent = nullptr);
~DaemonIpcClient();
void connect();
void connectToServer();
void sendCommand(const QString &command, ElevateMode elevateMode);
bool isConnected() const
{
return m_connected;
}
private slots:
void handleDisconnected();
void handleErrorOccurred();
private:
QLocalSocket *socket;
void sendMessage(const QString &message, const QString &expectAck = "ok", const bool expectConnected = true);
private:
QLocalSocket *m_socket;
bool m_connected{false};
};
} // namespace deskflow::gui::ipc