From 261549c8c253ef7b049d35803a9b71ef04fc31c5 Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Tue, 13 Dec 2016 12:00:48 -0500
Subject: [PATCH] Migrate scheduler to be run by fedmsg-hub

---
 Vagrantfile                                |   3 +-
 docker-compose.yml                         |   2 +-
 fedmsg.d/mbs-scheduler.py                  |   4 +
 module_build_service/builder.py            |   4 +-
 module_build_service/manage.py             |   8 +-
 module_build_service/messaging.py          |  45 +--
 module_build_service/scheduler/__init__.py |   3 +
 module_build_service/scheduler/consumer.py | 162 ++++++++++
 module_build_service/scheduler/main.py     | 357 ---------------------
 module_build_service/scheduler/producer.py | 160 +++++++++
 module_build_service/utils.py              |   8 +
 module_build_service_daemon.py             |  30 --
 setup.py                                   |   7 +-
 13 files changed, 352 insertions(+), 441 deletions(-)
 create mode 100644 fedmsg.d/mbs-scheduler.py
 create mode 100644 module_build_service/scheduler/consumer.py
 delete mode 100644 module_build_service/scheduler/main.py
 create mode 100644 module_build_service/scheduler/producer.py
 delete mode 100755 module_build_service_daemon.py

diff --git a/Vagrantfile b/Vagrantfile
index 21732364..769d5da5 100644
--- a/Vagrantfile
+++ b/Vagrantfile
@@ -6,6 +6,7 @@ $script = <<SCRIPT
     echo "export MODULE_BUILD_SERVICE_DEVELOPER_ENV=1" >> /etc/profile.d/module_build_service_developer_env.sh
     source /etc/profile.d/module_build_service_developer_env.sh
     dnf install -y \
+        fedmsg-hub \
         fedmsg-relay \
         fedpkg \
         gcc \
@@ -45,5 +46,5 @@ Vagrant.configure("2") do |config|
   config.vm.network "forwarded_port", guest: 13747, host: 13747
   config.vm.provision "shell", inline: $script
   config.vm.provision :shell, inline: "mbs-frontend &", run: "always"
-  config.vm.provision :shell, inline: "mbs-daemon &", run: "always"
+  config.vm.provision :shell, inline: "cd /tmp/module_build_service && fedmsg-hub &", run: "always"
 end
diff --git a/docker-compose.yml b/docker-compose.yml
index c90e4245..abd5f6c2 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -23,7 +23,7 @@ services:
     depends_on:
       - base
     build: .
-    command: mbs-daemon
+    command: fedmsg-hub
     links:
       - fedmsg-relay
     environment:
diff --git a/fedmsg.d/mbs-scheduler.py b/fedmsg.d/mbs-scheduler.py
new file mode 100644
index 00000000..5b6bd94f
--- /dev/null
+++ b/fedmsg.d/mbs-scheduler.py
@@ -0,0 +1,4 @@
+config = {
+    'mbsconsumer': True,
+    'mbsproducer': True,
+}
diff --git a/module_build_service/builder.py b/module_build_service/builder.py
index 16c46b9b..9c1941cd 100644
--- a/module_build_service/builder.py
+++ b/module_build_service/builder.py
@@ -53,8 +53,8 @@ from OpenSSL.SSL import SysCallError
 from module_build_service import conf, log, db
 from module_build_service.models import ModuleBuild
 import module_build_service.scm
-import module_build_service.scheduler.main
 import module_build_service.utils
+import module_build_service.scheduler
 
 logging.basicConfig(level=logging.DEBUG)
 
@@ -1165,7 +1165,7 @@ $repos
             msg_id='a faked internal message',
             repo_tag=self.tag_name + "-build",
         )
-        module_build_service.scheduler.main.outgoing_work_queue_put(msg)
+        module_build_service.scheduler.work_queue.put(msg)
 
     def _send_build_change(self, state, source, build_id):
         nvr = kobo.rpmlib.parse_nvr(source)
diff --git a/module_build_service/manage.py b/module_build_service/manage.py
index e3a6f178..4f6133d0 100644
--- a/module_build_service/manage.py
+++ b/module_build_service/manage.py
@@ -34,7 +34,6 @@ from module_build_service import models
 from module_build_service.pdc import (
     get_pdc_client_session, get_module, get_module_runtime_dependencies,
     get_module_tag, get_module_build_dependencies)
-import module_build_service.scheduler.main
 from module_build_service.utils import (
     submit_module_build,
     insert_fake_baseruntime,
@@ -146,9 +145,10 @@ def build_module_locally(url):
     username = getpass.getuser()
     submit_module_build(username, url, allow_local_url=True)
 
-    msgs = []
-    msgs.append(RidaModule("local module build", 2, 1))
-    module_build_service.scheduler.main.main(msgs, True)
+    # TODO: Ralph to the rescue
+    # msgs = []
+    # msgs.append(RidaModule("local module build", 2, 1))
+    # module_build_service.scheduler.main.main(msgs, True)
 
 
 @manager.command
diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index 33da8c57..431248b8 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -255,38 +255,11 @@ def publish(topic, msg, conf, service):
         raise KeyError("No messaging backend found for %r" % conf.messaging)
     return handler(topic, msg, conf, service)
 
-
-def listen(conf, **kwargs):
-    """
-    Yield messages from the messaging backend in conf.messaging.
-    :param conf: a Config object from the class in config.py
-    :param kwargs: any additional arguments to pass to the backend handler
-    :return: yields a message object (child class from BaseMessage)
-    """
-    try:
-        handler = _messaging_backends[conf.messaging]['listen']
-    except KeyError:
-        raise KeyError("No messaging backend found for %r" % conf.messaging)
-
-    for event in handler(conf, **kwargs):
-        yield event
-
-
 def _fedmsg_publish(topic, msg, conf, service):
     # fedmsg doesn't really need access to conf, however other backends do
     import fedmsg
     return fedmsg.publish(topic, msg=msg, modname=service)
 
-def _fedmsg_listen(conf, **kwargs):
-    """
-    Parses a fedmsg event and constructs it into the appropriate message object
-    """
-    import fedmsg
-    for name, endpoint, topic, msg in fedmsg.tail_messages(**kwargs):
-        msg_obj = BaseMessage.from_fedmsg(topic, msg)
-        if msg_obj:
-            yield msg_obj
-
 def _amq_get_messenger(conf):
     import proton
     for attr in ('amq_private_key_file', 'amq_trusted_cert_file', 'amq_cert_file'):
@@ -317,19 +290,6 @@ def _amq_get_messenger(conf):
         log.debug('proton.Messenger: Subscribing to address=%s' % url)
     return msngr
 
-def _amq_listen(conf, **kwargs):
-    import proton
-    msngr = _amq_get_messenger(conf)
-    msg = proton.Message()
-    while True:
-        msngr.recv()
-
-        while msngr.incoming:
-            msngr.get(msg)
-            msg_obj = BaseMessage.from_amq(msg.address, msg)
-            if msg_obj:
-                yield msg_obj
-
 def _amq_publish(topic, msg, conf, service):
     import proton
     msngr = _amq_get_messenger(conf)
@@ -356,6 +316,7 @@ def _in_memory_init(conf, **kwargs):
     _in_memory_msg_id = 0
     _in_memory_work_queue = queue.Queue()
 
+# TODO: Ralph to the rescue
 def _in_memory_publish(topic, msg, conf, service):
     """
     Puts the message to _in_memory_work_queue".
@@ -377,6 +338,7 @@ def _in_memory_publish(topic, msg, conf, service):
     # Put the message to queue.
     _in_memory_work_queue.put(wrapped_msg)
 
+# TODO: Ralph to the rescue
 def _in_memory_listen(conf, **kwargs):
     """
     Yields the message from the _in_memory_work_queue when ready.
@@ -394,16 +356,13 @@ _messaging_backends = {
     'fedmsg': {
         'init': _no_op,
         'publish': _fedmsg_publish,
-        'listen': _fedmsg_listen,
     },
     'amq': {
         'init': _no_op,
         'publish': _amq_publish,
-        'listen': _amq_listen,
     },
     'in_memory': {
         'init': _in_memory_init,
         'publish': _in_memory_publish,
-        'listen': _in_memory_listen,
     },
 }
diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index e69de29b..fcb40e2d 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -0,0 +1,3 @@
+import six.moves.queue as queue
+
+work_queue = queue.Queue()
diff --git a/module_build_service/scheduler/consumer.py b/module_build_service/scheduler/consumer.py
new file mode 100644
index 00000000..761c20d2
--- /dev/null
+++ b/module_build_service/scheduler/consumer.py
@@ -0,0 +1,162 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2016  Red Hat, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+
+""" The FedmsgConsumer class that acts as a consumer entry point for fedmsg-hub.
+This class reads and processes messages from the message bus it is configured
+to use.
+"""
+
+import koji
+import inspect
+import fedmsg.consumers
+
+from module_build_service.utils import module_build_state_from_msg
+import module_build_service.messaging
+import module_build_service.scheduler.handlers.repos
+import module_build_service.scheduler.handlers.components
+import module_build_service.scheduler.handlers.modules
+from module_build_service.scheduler import work_queue
+from module_build_service import models, log, conf
+
+
+class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
+    """ This is triggered by running fedmsg-hub. This class is responsible for
+    ingesting and processing messages from the message bus.
+    """
+    topic = '*'
+    config_key = 'mbsconsumer'
+
+    def __init__(self, hub, initial_msgs=[]):
+        super(MBSConsumer, self).__init__(hub)
+
+        for msg in initial_msgs:
+            work_queue.put(msg)
+
+        # These are our main lookup tables for figuring out what to run in
+        # response to what messaging events.
+        self.NO_OP = NO_OP = lambda config, session, msg: True
+        self.on_build_change = {
+            koji.BUILD_STATES["BUILDING"]: NO_OP,
+            koji.BUILD_STATES[
+                "COMPLETE"]: module_build_service.scheduler.handlers.components.complete,
+            koji.BUILD_STATES[
+                "FAILED"]: module_build_service.scheduler.handlers.components.failed,
+            koji.BUILD_STATES[
+                "CANCELED"]: module_build_service.scheduler.handlers.components.canceled,
+            koji.BUILD_STATES["DELETED"]: NO_OP,
+        }
+        self.on_module_change = {
+            models.BUILD_STATES["init"]: NO_OP,
+            models.BUILD_STATES[
+                "wait"]: module_build_service.scheduler.handlers.modules.wait,
+            models.BUILD_STATES["build"]: NO_OP,
+            models.BUILD_STATES[
+                "failed"]: module_build_service.scheduler.handlers.modules.failed,
+            models.BUILD_STATES[
+                "done"]: module_build_service.scheduler.handlers.modules.done,
+            # XXX: DIRECT TRANSITION TO READY
+            models.BUILD_STATES["ready"]: NO_OP,
+        }
+        # Only one kind of repo change event, though...
+        self.on_repo_change = module_build_service.scheduler.handlers.repos.done
+        self.sanity_check()
+
+    def consume(self, message):
+        # Add the message to the work queue
+        work_queue.put(self.get_abstracted_msg(message['body']))
+        # Process all the messages in the work queue
+        while not work_queue.empty():
+            msg = work_queue.get()
+            try:
+                with models.make_session(conf) as session:
+                    self.process_message(session, msg)
+            except Exception:
+                log.exception('Failed while handling {0!r}'.format(
+                    msg.msg_id))
+                log.info(msg)
+
+    def get_abstracted_msg(self, message):
+        # Convert the message to an abstracted message
+        if conf.messaging == 'fedmsg':
+            msg = module_build_service.messaging.BaseMessage.from_fedmsg(
+                message['topic'], message)
+        elif conf.messaging == 'amq':
+            msg = module_build_service.messaging.BaseMessage.from_amq(
+                message['topic'], message)
+        else:
+            raise ValueError('The messaging format "{0}" is not supported'
+                             .format(conf.messaging))
+        return msg
+
+    def sanity_check(self):
+        """ On startup, make sure our implementation is sane. """
+        # Ensure we have every state covered
+        for state in models.BUILD_STATES:
+            if models.BUILD_STATES[state] not in self.on_module_change:
+                raise KeyError("Module build states %r not handled." % state)
+        for state in koji.BUILD_STATES:
+            if koji.BUILD_STATES[state] not in self.on_build_change:
+                raise KeyError("Koji build states %r not handled." % state)
+
+        all_fns = (list(self.on_build_change.items()) +
+                   list(self.on_module_change.items()))
+        for key, callback in all_fns:
+            expected = ['config', 'session', 'msg']
+            argspec = inspect.getargspec(callback)[0]
+            if argspec != expected:
+                raise ValueError("Callback %r, state %r has argspec %r!=%r" % (
+                    callback, key, argspec, expected))
+
+    def process_message(self, session, msg):
+        log.debug('Received a message with an ID of "{0}" and of type "{1}"'
+                  .format(msg.msg_id, type(msg).__name__))
+
+        # Choose a handler for this message
+        if type(msg) == module_build_service.messaging.KojiBuildChange:
+            handler = self.on_build_change[msg.build_new_state]
+        elif type(msg) == module_build_service.messaging.KojiRepoChange:
+            handler = self.on_repo_change
+        elif type(msg) == module_build_service.messaging.RidaModule:
+            handler = self.on_module_change[module_build_state_from_msg(msg)]
+        else:
+            log.debug("Unhandled message...")
+            return
+
+        # Execute our chosen handler
+        idx = "%s: %s, %s" % (handler.__name__, type(msg).__name__,
+                              msg.msg_id)
+        if handler is self.NO_OP:
+            log.debug("Handler is NO_OP: %s" % idx)
+        else:
+            log.info("Calling %s" % idx)
+            further_work = handler(conf, session, msg) or []
+            log.info("Done with %s" % idx)
+
+            # Handlers can *optionally* return a list of fake messages that
+            # should be re-inserted back into the main work queue. We can use
+            # this (for instance) when we submit a new component build but (for
+            # some reason) it has already been built, then it can fake its own
+            # completion back to the scheduler so that work resumes as if it
+            # was submitted for real and koji announced its completion.
+            for event in further_work:
+                log.info("  Scheduling faked event %r" % event)
+                work_queue.put(event)
diff --git a/module_build_service/scheduler/main.py b/module_build_service/scheduler/main.py
deleted file mode 100644
index 8b0c6b5d..00000000
--- a/module_build_service/scheduler/main.py
+++ /dev/null
@@ -1,357 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright (c) 2016  Red Hat, Inc.
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-#
-# Written by Petr Šabata <contyk@redhat.com>
-#            Ralph Bean <rbean@redhat.com>
-
-"""The module build orchestrator for Modularity, the builder.
-
-This is the main component of the orchestrator and is responsible for
-proper scheduling component builds in the supported build systems.
-"""
-
-import inspect
-import operator
-import os
-import threading
-import time
-import six.moves.queue as queue
-
-import module_build_service.config
-import module_build_service.messaging
-import module_build_service.utils
-import module_build_service.scheduler.handlers.components
-import module_build_service.scheduler.handlers.modules
-import module_build_service.scheduler.handlers.repos
-
-import koji
-
-from module_build_service import conf, models, log
-
-from sqlalchemy.orm import lazyload
-
-
-class STOP_WORK(object):
-    """ A sentinel value, indicating that work should be stopped. """
-    pass
-
-
-def module_build_state_from_msg(msg):
-    state = int(msg.module_build_state)
-    # TODO better handling
-    assert state in models.BUILD_STATES.values(), (
-        'state=%s(%s) is not in %s'
-        % (state, type(state), list(models.BUILD_STATES.values())))
-    return state
-
-
-class MessageIngest(threading.Thread):
-    def __init__(self, outgoing_work_queue, stop_after_build, *args, **kwargs):
-        self.outgoing_work_queue = outgoing_work_queue
-        super(MessageIngest, self).__init__(*args, **kwargs)
-        self.stop_after_build = stop_after_build
-
-    def run(self):
-        for msg in module_build_service.messaging.listen(conf):
-            self.outgoing_work_queue.put(msg)
-
-            if type(msg) == module_build_service.messaging.RidaModule:
-                if (self.stop_after_build and module_build_state_from_msg(msg)
-                        in [models.BUILD_STATES["failed"], models.BUILD_STATES["ready"]]):
-                    break
-
-
-class MessageWorker(threading.Thread):
-
-    def __init__(self, incoming_work_queue, stop_after_build, *args, **kwargs):
-        self.incoming_work_queue = incoming_work_queue
-        self.stop_after_build = stop_after_build
-        super(MessageWorker, self).__init__(*args, **kwargs)
-
-        # These are our main lookup tables for figuring out what to run in response
-        # to what messaging events.
-        self.NO_OP = NO_OP = lambda config, session, msg: True
-        self.on_build_change = {
-            koji.BUILD_STATES["BUILDING"]: NO_OP,
-            koji.BUILD_STATES["COMPLETE"]: module_build_service.scheduler.handlers.components.complete,
-            koji.BUILD_STATES["FAILED"]: module_build_service.scheduler.handlers.components.failed,
-            koji.BUILD_STATES["CANCELED"]: module_build_service.scheduler.handlers.components.canceled,
-            koji.BUILD_STATES["DELETED"]: NO_OP,
-        }
-        self.on_module_change = {
-            models.BUILD_STATES["init"]: NO_OP,
-            models.BUILD_STATES["wait"]: module_build_service.scheduler.handlers.modules.wait,
-            models.BUILD_STATES["build"]: NO_OP,
-            models.BUILD_STATES["failed"]: module_build_service.scheduler.handlers.modules.failed,
-            models.BUILD_STATES["done"]: module_build_service.scheduler.handlers.modules.done,  # XXX: DIRECT TRANSITION TO READY
-            models.BUILD_STATES["ready"]: NO_OP,
-        }
-        # Only one kind of repo change event, though...
-        self.on_repo_change = module_build_service.scheduler.handlers.repos.done
-
-    def sanity_check(self):
-        """ On startup, make sure our implementation is sane. """
-        # Ensure we have every state covered
-        for state in models.BUILD_STATES:
-            if models.BUILD_STATES[state] not in self.on_module_change:
-                raise KeyError("Module build states %r not handled." % state)
-        for state in koji.BUILD_STATES:
-            if koji.BUILD_STATES[state] not in self.on_build_change:
-                raise KeyError("Koji build states %r not handled." % state)
-
-        all_fns = (list(self.on_build_change.items()) +
-                   list(self.on_module_change.items()))
-        for key, callback in all_fns:
-            expected = ['config', 'session', 'msg']
-            argspec = inspect.getargspec(callback)[0]
-            if argspec != expected:
-                raise ValueError("Callback %r, state %r has argspec %r!=%r" % (
-                    callback, key, argspec, expected))
-
-    def run(self):
-        self.sanity_check()
-
-        while True:
-            msg = self.incoming_work_queue.get()
-
-            if msg is STOP_WORK:
-                log.info("Worker thread received STOP_WORK, shutting down...")
-                break
-
-            try:
-                with models.make_session(conf) as session:
-                    self.process_message(session, msg)
-            except Exception:
-                log.exception("Failed while handling %r" % msg.msg_id)
-                log.info(msg)
-
-    def process_message(self, session, msg):
-        log.debug('Received a message with an ID of "{0}" and of type "{1}"'
-                  .format(msg.msg_id, type(msg).__name__))
-
-        # Choose a handler for this message
-        if type(msg) == module_build_service.messaging.KojiBuildChange:
-            handler = self.on_build_change[msg.build_new_state]
-        elif type(msg) == module_build_service.messaging.KojiRepoChange:
-            handler = self.on_repo_change
-        elif type(msg) == module_build_service.messaging.RidaModule:
-            handler = self.on_module_change[module_build_state_from_msg(msg)]
-            if (self.stop_after_build and module_build_state_from_msg(msg)
-                    in [models.BUILD_STATES["failed"], models.BUILD_STATES["ready"]]):
-                self.incoming_work_queue.put(STOP_WORK)
-        else:
-            log.debug("Unhandled message...")
-            return
-
-        # Execute our chosen handler
-        idx = "%s: %s, %s" % (handler.__name__, type(msg).__name__,
-                              msg.msg_id)
-        if handler is self.NO_OP:
-            log.debug("Handler is NO_OP: %s" % idx)
-        else:
-            log.info("Calling %s" % idx)
-            further_work = handler(conf, session, msg) or []
-            log.info("Done with %s" % idx)
-
-            # Handlers can *optionally* return a list of fake messages that
-            # should be re-inserted back into the main work queue. We can use
-            # this (for instance) when we submit a new component build but (for
-            # some reason) it has already been built, then it can fake its own
-            # completion back to the scheduler so that work resumes as if it
-            # was submitted for real and koji announced its completion.
-            for event in further_work:
-                log.info("  Scheduling faked event %r" % event)
-                self.incoming_work_queue.put(event)
-
-
-class Poller(threading.Thread):
-    def __init__(self, outgoing_work_queue, *args, **kwargs):
-        self.outgoing_work_queue = outgoing_work_queue
-        super(Poller, self).__init__(*args, **kwargs)
-        self.stop = False
-
-    def run(self):
-        while not self.stop:
-            with models.make_session(conf) as session:
-                self.log_summary(session)
-                # XXX: detect whether it's really stucked first
-                # self.process_waiting_module_builds(session)
-                self.process_open_component_builds(session)
-                self.fail_lost_builds(session)
-                self.process_paused_module_builds(conf, session)
-
-            log.info("Polling thread sleeping, %rs" % conf.polling_interval)
-            for i in range(0, conf.polling_interval):
-                time.sleep(1)
-                if self.stop:
-                    break
-
-    def fail_lost_builds(self, session):
-        # This function is supposed to be handling only
-        # the part which can't be updated trough messaging (srpm-build failures).
-        # Please keep it fit `n` slim. We do want rest to be processed elsewhere
-
-        # TODO re-use
-
-        if conf.system == "koji":
-            # we don't do this on behalf of users
-            koji_session = (
-                module_build_service.builder.KojiModuleBuilder.get_session(conf, None))
-            log.info("Querying tasks for statuses:")
-            res = models.ComponentBuild.query.filter_by(state=koji.BUILD_STATES['BUILDING']).options(lazyload('module_build')).all()
-
-            log.info("Checking status for %d tasks." % len(res))
-            for component_build in res:
-                log.debug(component_build.json())
-                if not component_build.task_id:  # Don't check tasks which has not been triggered yet
-                    continue
-
-                log.info("Checking status of task_id=%s" % component_build.task_id)
-                task_info = koji_session.getTaskInfo(component_build.task_id)
-
-                dead_states = (
-                    koji.TASK_STATES['CANCELED'],
-                    koji.TASK_STATES['FAILED'],
-                )
-                log.info("  task %r is in state %r" % (component_build.task_id, task_info['state']))
-                if task_info['state'] in dead_states:
-                    # Fake a fedmsg message on our internal queue
-                    msg = module_build_service.messaging.KojiBuildChange(
-                        msg_id='a faked internal message',
-                        build_id=component_build.task_id,
-                        task_id=component_build.task_id,
-                        build_name=component_build.package,
-                        build_new_state=koji.BUILD_STATES['FAILED'],
-                        build_release=None,
-                        build_version=None
-                    )
-                    self.outgoing_work_queue.put(msg)
-
-        elif conf.system == "copr":
-            # @TODO
-            pass
-
-        elif conf.system == "mock":
-            pass
-
-    def log_summary(self, session):
-        log.info("Current status:")
-        backlog = self.outgoing_work_queue.qsize()
-        log.info("  * internal queue backlog is %i." % backlog)
-        states = sorted(models.BUILD_STATES.items(), key=operator.itemgetter(1))
-        for name, code in states:
-            query = models.ModuleBuild.query.filter_by(state=code)
-            count = query.count()
-            if count:
-                log.info("  * %i module builds in the %s state." % (count, name))
-            if name == 'build':
-                for module_build in query.all():
-                    log.info("    * %r" % module_build)
-                    for i in range(module_build.batch):
-                        n = len([c for c in module_build.component_builds
-                                 if c.batch == i])
-                        log.info("      * %i components in batch %i" % (n, i))
-
-    def process_waiting_module_builds(self, session):
-        log.info("Looking for module builds stuck in the wait state.")
-        builds = models.ModuleBuild.by_state(session, "wait")
-        # TODO -- do throttling calculation here...
-        log.info(" %r module builds in the wait state..." % len(builds))
-        for build in builds:
-            # Fake a message to kickstart the build anew
-            msg = {
-                'topic': '.module.build.state.change',
-                'msg': build.json(),
-            }
-            module_build_service.scheduler.handlers.modules.wait(conf, session, msg)
-
-    def process_open_component_builds(self, session):
-        log.warning("process_open_component_builds is not yet implemented...")
-
-    def process_paused_module_builds(self, config, session):
-        if module_build_service.utils.at_concurrent_component_threshold(
-                config, session):
-            log.debug('Will not attempt to start paused module builds due to '
-                      'the concurrent build threshold being met')
-            return
-        # Check to see if module builds that are in build state but don't have
-        # any component builds being built can be worked on
-        for module_build in session.query(models.ModuleBuild).filter_by(
-                    state=models.BUILD_STATES['build']).all():
-            # If there are no components in the build state on the module build,
-            # then no possible event will start off new component builds
-            if not module_build.current_batch(koji.BUILD_STATES['BUILDING']):
-                further_work = module_build_service.utils.start_build_batch(
-                    config, module_build, session, config.system)
-                for event in further_work:
-                    log.info("  Scheduling faked event %r" % event)
-                    self.outgoing_work_queue.put(event)
-
-
-_work_queue = queue.Queue()
-
-
-def outgoing_work_queue_put(msg):
-    _work_queue.put(msg)
-
-
-def graceful_stop():
-    """
-    Here is the place to perform shutdown actions.
-
-    Do whatever is needed to do except for leaving the main thread, which
-    would result in losing control of POSIX signals handling.
-    """
-    log.warning("graceful_stop is not yet implemented, press Ctrl+C again...")
-    while True:
-        time.sleep(30)
-
-
-def main(initial_msgs=[], return_after_build=False):
-    log.info("Starting module_build_service_daemon.")
-
-    module_build_service.messaging.init(conf)
-
-    for msg in initial_msgs:
-        outgoing_work_queue_put(msg)
-
-    # This ingest thread puts work on the queue
-    messaging_thread = MessageIngest(_work_queue, return_after_build)
-    # This poller does other work, but also sometimes puts work in queue.
-    polling_thread = Poller(_work_queue)
-    # This worker takes work off the queue and handles it.
-    worker_thread = MessageWorker(_work_queue, return_after_build)
-
-    messaging_thread.start()
-    polling_thread.start()
-    worker_thread.start()
-
-    try:
-        while worker_thread.is_alive():
-            time.sleep(3)
-    except KeyboardInterrupt:
-        log.info("Stopping module_build_service_daemon. Press Ctrl+C again to force.")
-        try:
-            graceful_stop()
-        except KeyboardInterrupt:
-            os._exit(0)
-    finally:
-        polling_thread.stop = True
diff --git a/module_build_service/scheduler/producer.py b/module_build_service/scheduler/producer.py
new file mode 100644
index 00000000..980d2757
--- /dev/null
+++ b/module_build_service/scheduler/producer.py
@@ -0,0 +1,160 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2016  Red Hat, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+
+""" The PollingProducer class that acts as a producer entry point for
+fedmsg-hub. This class polls the database for tasks to do.
+"""
+
+import koji
+import operator
+from datetime import timedelta
+from sqlalchemy.orm import lazyload
+from moksha.hub.api.producer import PollingProducer
+
+from module_build_service.scheduler import work_queue
+import module_build_service.scheduler
+from module_build_service import conf, models, log
+
+
+class MBSProducer(PollingProducer):
+    frequency = timedelta(seconds=conf.polling_interval)
+
+    def poll(self):
+        with models.make_session(conf) as session:
+            self.log_summary(session)
+            # XXX: detect whether it's actually stuck first
+            # self.process_waiting_module_builds(session)
+            self.process_open_component_builds(session)
+            self.fail_lost_builds(session)
+            self.process_paused_module_builds(conf, session)
+
+        log.info('Poller will now sleep for "{}" seconds'
+                 .format(conf.polling_interval))
+
+    def fail_lost_builds(self, session):
+        # This function is supposed to be handling only the part which can't be
+        # updated through messaging (e.g. srpm-build failures). Please keep it
+        # fit `n` slim. We do want rest to be processed elsewhere
+        # TODO re-use
+
+        if conf.system == 'koji':
+            # We don't do this on behalf of users
+            koji_session = module_build_service.builder.KojiModuleBuilder\
+                .get_session(conf, None)
+            log.info('Querying tasks for statuses:')
+            res = models.ComponentBuild.query.filter_by(
+                state=koji.BUILD_STATES['BUILDING']).options(
+                    lazyload('module_build')).all()
+
+            log.info('Checking status for {0} tasks'.format(len(res)))
+            for component_build in res:
+                log.debug(component_build.json())
+                # Don't check tasks which haven't been triggered yet
+                if not component_build.task_id:
+                    continue
+
+                log.info('Checking status of task_id "{0}"'
+                         .format(component_build.task_id))
+                task_info = koji_session.getTaskInfo(component_build.task_id)
+
+                dead_states = (
+                    koji.TASK_STATES['CANCELED'],
+                    koji.TASK_STATES['FAILED'],
+                )
+
+                log.info('  task {0!r} is in state {0!r}'.format(
+                    component_build.task_id, task_info['state']))
+                if task_info['state'] in dead_states:
+                    # Fake a fedmsg message on our internal queue
+                    msg = module_build_service.messaging.KojiBuildChange(
+                        msg_id='a faked internal message',
+                        build_id=component_build.task_id,
+                        task_id=component_build.task_id,
+                        build_name=component_build.package,
+                        build_new_state=koji.BUILD_STATES['FAILED'],
+                        build_release=None,
+                        build_version=None
+                    )
+                    work_queue.put(msg)
+
+        elif conf.system == 'copr':
+            # @TODO
+            pass
+
+        elif conf.system == 'mock':
+            pass
+
+    def log_summary(self, session):
+        log.info('Current status:')
+        backlog = work_queue.qsize()
+        log.info('  * internal queue backlog is {0}'.format(backlog))
+        states = sorted(models.BUILD_STATES.items(), key=operator.itemgetter(1))
+        for name, code in states:
+            query = models.ModuleBuild.query.filter_by(state=code)
+            count = query.count()
+            if count:
+                log.info('  * {0} module builds in the {1} state'.format(
+                    count, name))
+            if name == 'build':
+                for module_build in query.all():
+                    log.info('    * {0!r}'.format(module_build))
+                    for i in range(module_build.batch):
+                        n = len([c for c in module_build.component_builds
+                                 if c.batch == i ])
+                        log.info('      * {0} components in batch {1}'
+                                 .format(n, i))
+
+    def process_waiting_module_builds(self, session):
+        log.info('Looking for module builds stuck in the wait state')
+        builds = models.ModuleBuild.by_state(session, 'wait')
+        log.info(' {0!r} module builds in the wait state...'
+                 .format(len(builds)))
+        for build in builds:
+            # Fake a message to kickstart the build anew
+            msg = {
+                'topic': '.module.build.state.change',
+                'msg': build.json(),
+            }
+            module_build_service.scheduler.handlers.modules.wait(
+                conf, session, msg)
+
+    def process_open_component_builds(self, session):
+        log.warning('process_open_component_builds is not yet implemented...')
+
+    def process_paused_module_builds(self, config, session):
+        if module_build_service.utils.at_concurrent_component_threshold(
+                config, session):
+            log.debug('Will not attempt to start paused module builds due to '
+                      'the concurrent build threshold being met')
+            return
+        # Check to see if module builds that are in build state but don't have
+        # any component builds being built can be worked on
+        for module_build in session.query(models.ModuleBuild).filter_by(
+                    state=models.BUILD_STATES['build']).all():
+            # If there are no components in the build state on the module build,
+            # then no possible event will start off new component builds
+            if not module_build.current_batch(koji.BUILD_STATES['BUILDING']):
+                further_work = module_build_service.utils.start_build_batch(
+                    config, module_build, session, config.system)
+                for event in further_work:
+                    log.info("  Scheduling faked event %r" % event)
+                    work_queue.put(event)
diff --git a/module_build_service/utils.py b/module_build_service/utils.py
index dd415944..df298518 100644
--- a/module_build_service/utils.py
+++ b/module_build_service/utils.py
@@ -548,3 +548,11 @@ def scm_url_schemes(terse=False):
         for scm_type, scm_schemes in scm_types.items():
             scheme_list.extend([scheme[:-3] for scheme in scm_schemes])
         return list(set(scheme_list))
+
+def module_build_state_from_msg(msg):
+    state = int(msg.module_build_state)
+    # TODO better handling
+    assert state in models.BUILD_STATES.values(), (
+        'state=%s(%s) is not in %s'
+        % (state, type(state), list(models.BUILD_STATES.values())))
+    return state
diff --git a/module_build_service_daemon.py b/module_build_service_daemon.py
deleted file mode 100755
index 11a3a5ae..00000000
--- a/module_build_service_daemon.py
+++ /dev/null
@@ -1,30 +0,0 @@
-#!/usr/bin/python3
-# -*- coding: utf-8 -*-
-# Copyright (c) 2016  Red Hat, Inc.
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-#
-# Written by Petr Šabata <contyk@redhat.com>
-#            Ralph Bean <rbean@redhat.com>
-"""The module build orchestrator for Modularity, the builder. """
-
-import module_build_service.scheduler.main
-
-if __name__ == '__main__':
-    module_build_service.scheduler.main.main()
diff --git a/setup.py b/setup.py
index 0494604e..c54cb360 100644
--- a/setup.py
+++ b/setup.py
@@ -24,11 +24,12 @@ setup(name='module-build-service',
       install_requires=requirements,
       tests_require=test_requirements,
       entry_points={
-          'console_scripts': ['mbs-daemon = module_build_service.scheduler.main:main',
-                              'mbs-upgradedb = module_build_service.manage:upgradedb',
+          'console_scripts': ['mbs-upgradedb = module_build_service.manage:upgradedb',
                               'mbs-gencert = module_build_service.manage:generatelocalhostcert',
                               'mbs-frontend = module_build_service.manage:runssl',
-                              'mbs-manager = module_build_service.manage:manager_wrapper']
+                              'mbs-manager = module_build_service.manage:manager_wrapper'],
+          'moksha.consumer': 'mbsconsumer = module_build_service.scheduler.consumer:MBSConsumer',
+          'moksha.producer': 'mbspoller = module_build_service.scheduler.producer:MBSProducer',
       },
       data_files=[('/etc/module-build-service/', ['conf/cacert.pem',
                                                   'conf/config.py',