From: Michael Tremer <michael.tremer@ipfire.org>
Date: Thu, 26 May 2022 14:51:40 +0000 (+0000)
Subject: hub: Add a new queue handler
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=b1a5b3840ecb28c326de4e185f7ec3dbd519887c;p=pbs.git

hub: Add a new queue handler

Builders can connect to this handler and will be sent build jobs
whenever some are available.

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

diff --git a/Makefile.am b/Makefile.am
index c492618a..068be00b 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -116,7 +116,8 @@ CLEANFILES += \
 
 hub_PYTHON = \
 	src/hub/__init__.py \
-	src/hub/handlers.py
+	src/hub/handlers.py \
+	src/hub/queue.py
 
 hubdir = $(buildservicedir)/hub
 
diff --git a/src/hub/__init__.py b/src/hub/__init__.py
index 53ff81e5..f8073ce9 100644
--- a/src/hub/__init__.py
+++ b/src/hub/__init__.py
@@ -1,10 +1,13 @@
 #!/usr/bin/python
 
+import functools
 import logging
+import tornado.ioloop
 import tornado.web
 
 from .. import Backend
 from . import handlers
+from . import queue
 
 class Application(tornado.web.Application):
 	def __init__(self, **settings):
@@ -40,6 +43,9 @@ class Application(tornado.web.Application):
 			# Packages
 			(r"/packages/(.*)", handlers.PackagesGetHandler),
 
+			# Queue
+			(r"/queue", queue.QueueHandler),
+
 			# Uploads
 			(r"/upload", handlers.UploadHandler),
 		], **settings)
@@ -47,4 +53,20 @@ class Application(tornado.web.Application):
 		# Launch backend
 		self.backend = Backend()
 
+		# Launch any background jobs
+		self._run_task(queue.dispatch_jobs, 5)
+
 		logging.info("Successfully initialied application")
+
+	def _run_task(self, callback, t):
+		"""
+			Runs the callback every t seconds in the background
+		"""
+		# Pass backend to the function
+		callback = functools.partial(callback, self.backend)
+
+		# Create a periodic callback object
+		task = tornado.ioloop.PeriodicCallback(callback, t * 1000)
+
+		# Start the task
+		task.start()
diff --git a/src/hub/queue.py b/src/hub/queue.py
new file mode 100644
index 00000000..b4b50d66
--- /dev/null
+++ b/src/hub/queue.py
@@ -0,0 +1,86 @@
+#!/usr/bin/python3
+###############################################################################
+#                                                                             #
+# Pakfire - The IPFire package management system                              #
+# Copyright (C) 2011 Pakfire development team                                 #
+#                                                                             #
+# This program is free software: you can redistribute it and/or modify        #
+# it under the terms of the GNU General Public License as published by        #
+# the Free Software Foundation, either version 3 of the License, or           #
+# (at your option) any later version.                                         #
+#                                                                             #
+# This program is distributed in the hope that it will be useful,             #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of              #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
+# GNU General Public License for more details.                                #
+#                                                                             #
+# You should have received a copy of the GNU General Public License           #
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
+#                                                                             #
+###############################################################################
+
+import logging
+import tornado.websocket
+
+from .handlers import BackendMixin, HTTPBasicAuthMixin
+
+log = logging.getLogger("pakfire.hub.queue")
+
+# A list of all builders that have a connection
+connections = []
+
+def dispatch_jobs(backend):
+	"""
+		Will be called regularly and will dispatch any pending jobs to any
+		available builders
+	"""
+	log.debug("Dispatching jobs...")
+
+	# Exit if there are no builders connected
+	if not connections:
+		log.debug("  No connections open")
+		return
+
+	# Process all connections
+	for connection in connections:
+		log.debug("  Processing connection for %s" % connection.builder)
+
+		with backend.db.transaction():
+			pass
+
+	# XXX TODO
+
+
+class QueueHandler(BackendMixin, HTTPBasicAuthMixin, tornado.websocket.WebSocketHandler):
+	"""
+		Builders connect to this handler which will add them to a list of connections.
+		
+		For all connections, we regularly check if we have any new build jobs, and if so,
+		we will send them the job.
+	"""
+	def get_current_user(self):
+		name, password = self.get_basic_auth_credentials()
+		if name is None:
+			return
+
+		return self.backend.builders.auth(name, password)
+
+	@property
+	def builder(self):
+		return self.current_user
+
+	@tornado.web.authenticated
+	def open(self):
+		log.debug("Connection opened by %s" % self.builder)
+
+		# Add this connection to the list
+		connections.append(self)
+
+	def on_close(self):
+		log.debug("Connection to %s closed" % self.builder)
+
+		# Remove the connection
+		try:
+			connections.remove(self)
+		except IndexError:
+			pass