From: Michael Tremer <michael.tremer@ipfire.org>
Date: Wed, 5 Feb 2025 09:24:57 +0000 (+0000)
Subject: daemon: Handle messages sent to a specific job
X-Git-Tag: 0.9.30~135
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ff6a78bc981e4d10eccd859d128e8217ea0bbe18;p=pakfire.git

daemon: Handle messages sent to a specific job

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>
---

diff --git a/src/pakfire/daemon.c b/src/pakfire/daemon.c
index 25c4dbe7..057c1858 100644
--- a/src/pakfire/daemon.c
+++ b/src/pakfire/daemon.c
@@ -112,6 +112,21 @@ struct pakfire_daemon {
 	unsigned int running_jobs;
 };
 
+static struct pakfire_job* pakfire_daemon_find_job(
+		struct pakfire_daemon* self, const char* job_id) {
+	// Walk through all jobs
+	for (unsigned int i = 0; i < MAX_JOBS; i++) {
+		if (!self->jobs[i])
+			continue;
+
+		// Return the matching job
+		if (pakfire_job_has_id(self->jobs[i], job_id))
+			return pakfire_job_ref(self->jobs[i]);
+	}
+
+	return NULL;
+}
+
 /*
 	Terminates all running jobs
 */
@@ -438,6 +453,33 @@ static int pakfire_daemon_release_inhibit_shutdown(struct pakfire_daemon* daemon
 	return 0;
 }
 
+/*
+	Called when we have received a message for a specific job
+*/
+static int pakfire_daemon_handle_job_message(
+		struct pakfire_daemon* self, const char* job_id, struct json_object* message) {
+	struct pakfire_job* job = NULL;
+	int r;
+
+	// Find the job
+	job = pakfire_daemon_find_job(self, job_id);
+	if (!job) {
+		WARN(self->ctx, "Received message for job %s which does not exist. Ignoring.\n", job_id);
+		return 0;
+	}
+
+	// Dispatch the message to the job
+	r = pakfire_job_handle_message(job, message);
+	if (r < 0)
+		goto ERROR;
+
+ERROR:
+	if (job)
+		pakfire_job_unref(job);
+
+	return r;
+}
+
 /*
 	Called when a new job has been received
 */
@@ -500,6 +542,8 @@ static int pakfire_daemon_recv(struct pakfire_xfer* xfer,
 	struct json_object* m = NULL;
 	int r;
 
+	const char* job_id = NULL;
+
 	// Parse the JSON message
 	m = pakfire_json_parse(daemon->ctx, message, size);
 	if (!m) {
@@ -507,6 +551,29 @@ static int pakfire_daemon_recv(struct pakfire_xfer* xfer,
 		goto ERROR;
 	}
 
+	// Does this message have a build job ID?
+	r = pakfire_json_get_string(m, "job_id", &job_id);
+	if (r < 0) {
+		switch (-r) {
+			// Fall through if the key did not exist
+			case ENOMSG:
+				break;
+
+			// Otherwise raise the error
+			default:
+				goto ERROR;
+		}
+	}
+
+	// If we have a job ID, this message is for a specific job
+	if (job_id) {
+		r = pakfire_daemon_handle_job_message(daemon, job_id, m);
+		if (r < 0)
+			goto ERROR;
+
+		goto ERROR;
+	}
+
 	// Fetch the message type
 	if (!json_object_object_get_ex(m, "type", &type)) {
 		ERROR(daemon->ctx, "Invalid message with missing type\n");
diff --git a/src/pakfire/job.c b/src/pakfire/job.c
index ea7d9225..2505c1a6 100644
--- a/src/pakfire/job.c
+++ b/src/pakfire/job.c
@@ -952,6 +952,10 @@ struct pakfire_job* pakfire_job_unref(struct pakfire_job* job) {
 	return NULL;
 }
 
+int pakfire_job_has_id(struct pakfire_job* job, const char* id) {
+	return pakfire_string_equals(job->id, id);
+}
+
 /*
 	Terminates the job (if it is still running)
 */
@@ -972,4 +976,10 @@ int pakfire_job_terminate(struct pakfire_job* job, int signal) {
 	return 0;
 }
 
+int pakfire_job_handle_message(struct pakfire_job* self, struct json_object* message) {
+	DEBUG(self->ctx, "Message received for job %s\n", self->id);
+
+	return 0; // XXX TODO
+}
+
 #endif /* CURL_HAS_WEBSOCKETS */
diff --git a/src/pakfire/job.h b/src/pakfire/job.h
index 0e290476..5f216f6a 100644
--- a/src/pakfire/job.h
+++ b/src/pakfire/job.h
@@ -34,6 +34,9 @@ int pakfire_job_create(struct pakfire_job** worker,
 struct pakfire_job* pakfire_job_ref(struct pakfire_job* worker);
 struct pakfire_job* pakfire_job_unref(struct pakfire_job* worker);
 
+// ID
+int pakfire_job_has_id(struct pakfire_job* job, const char* id);
+
 // Launch
 int pakfire_job_launch(struct pakfire_job* job);
 
@@ -44,5 +47,8 @@ int pakfire_job_terminate(struct pakfire_job* worker, int signal);
 int pakfire_job_launch_log_stream(struct pakfire_job* job);
 int pakfire_job_terminate_log_stream(struct pakfire_job* job);
 
+// Message Received
+int pakfire_job_handle_message(struct pakfire_job* self, struct json_object* message);
+
 #endif /* CURL_HAS_WEBSOCKETS */
 #endif /* PAKFIRE_JOB_H */