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 c492618ab962e1b9d031b3a23184b94c80ba6701..068be00b78af8a951791565dea4aa84514d3b9ae 100644 (file)
--- 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 53ff81e5d25167d424c870f4d846d84a37ccfdbb..f8073ce90b11b351836b39dfe63a48b16e7fd1a9 100644 (file)
--- 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 (file)
index 0000000..b4b50d6
--- /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