jobs: refactor, use new set_message_unique_id

Ease the burden on subclasses of Job slightly, by providing a new
set_message_unique_id() method. It ensures that created Message
instances will use that id.

This removes the need to override the message() method, so it is no
longer in the 'abstract method' section.

Enable callers to Job's message() method to override the 'unique_id'.

2 files changed, 37 insertions, 26 deletions

diff --git a/src/buildstream/_scheduler/jobs/elementjob.py b/src/buildstream/_scheduler/jobs/elementjob.py
index fb5d38e11..d6aa81567 100644
--- a/src/buildstream/_scheduler/jobs/elementjob.py
+++ b/src/buildstream/_scheduler/jobs/elementjob.py
@@ -18,7 +18,7 @@
 #
 from ruamel import yaml
 
-from ..._message import Message, MessageType
+from ..._message import MessageType
 
 from .job import Job
 
@@ -72,7 +72,8 @@ class ElementJob(Job):
         self._action_cb = action_cb            # The action callable function
         self._complete_cb = complete_cb        # The complete callable function
 
-        # Set the task wide ID for logging purposes
+        # Set the ID for logging purposes
+        self.set_message_unique_id(element._unique_id)
         self.set_task_id(element._unique_id)
 
     @property
@@ -96,15 +97,6 @@ class ElementJob(Job):
     def parent_complete(self, status, result):
         self._complete_cb(self, self._element, status, self._result)
 
-    def message(self, message_type, message, **kwargs):
-        args = dict(kwargs)
-        args['scheduler'] = True
-        self._scheduler.context.message(
-            Message(self._element._unique_id,
-                    message_type,
-                    message,
-                    **args))
-
     def child_process_data(self):
         data = {}
 
diff --git a/src/buildstream/_scheduler/jobs/job.py b/src/buildstream/_scheduler/jobs/job.py
index 1c19d0221..203564af7 100644
--- a/src/buildstream/_scheduler/jobs/job.py
+++ b/src/buildstream/_scheduler/jobs/job.py
@@ -112,6 +112,7 @@ class Job():
         self._terminated = False               # Whether this job has been explicitly terminated
 
         self._logfile = logfile
+        self._message_unique_id = None
         self._task_id = None
 
     # spawn()
@@ -254,12 +255,25 @@ class Job():
             os.kill(self._process.pid, signal.SIGCONT)
             self._suspended = False
 
+    # set_message_unique_id()
+    #
+    # This is called by Job subclasses to set the plugin ID
+    # issuing the message (if an element is related to the Job).
+    #
+    # Args:
+    #     unique_id (int): The id to be supplied to the Message() constructor
+    #
+    def set_message_unique_id(self, unique_id):
+        self._message_unique_id = unique_id
+
     # set_task_id()
     #
     # This is called by Job subclasses to set a plugin ID
     # associated with the task at large (if any element is related
     # to the task).
     #
+    # This will only be used in the child process running the task.
+    #
     # The task ID helps keep messages in the frontend coherent
     # in the case that multiple plugins log in the context of
     # a single task (e.g. running integration commands should appear
@@ -272,6 +286,26 @@ class Job():
     def set_task_id(self, task_id):
         self._task_id = task_id
 
+    # message():
+    #
+    # Logs a message, this will be logged in the task's logfile and
+    # conditionally also be sent to the frontend.
+    #
+    # Args:
+    #    message_type (MessageType): The type of message to send
+    #    message (str): The message
+    #    kwargs: Remaining Message() constructor arguments, note that you can
+    #            override 'unique_id' this way.
+    #
+    def message(self, message_type, message, **kwargs):
+        kwargs['scheduler'] = True
+        unique_id = self._message_unique_id
+        if "unique_id" in kwargs:
+            unique_id = kwargs["unique_id"]
+            del kwargs["unique_id"]
+        self._scheduler.context.message(
+            Message(unique_id, message_type, message, **kwargs))
+
     # send_message()
     #
     # To be called from inside Job.child_process() implementations
@@ -329,21 +363,6 @@ class Job():
         raise ImplError("Job '{kind}' does not implement child_process()"
                         .format(kind=type(self).__name__))
 
-    # message():
-    #
-    # Logs a message, this will be logged in the task's logfile and
-    # conditionally also be sent to the frontend.
-    #
-    # Args:
-    #    message_type (MessageType): The type of message to send
-    #    message (str): The message
-    #    kwargs: Remaining Message() constructor arguments
-    #
-    def message(self, message_type, message, **kwargs):
-        args = dict(kwargs)
-        args['scheduler'] = True
-        self._scheduler.context.message(Message(None, message_type, message, **args))
-
     # child_process_data()
     #
     # Abstract method to retrieve additional data that should be