From e0240edee6602abcd35cd0b1f8d01aea5b75ee05 Mon Sep 17 00:00:00 2001
From: Filip Valder <fvalder@redhat.com>
Date: Wed, 14 Dec 2016 14:00:26 +0100
Subject: [PATCH 01/29] fix #264: failed 'failed module' handling due to
 missing koji tag

---
 module_build_service/models.py                | 13 +++++-----
 .../scheduler/handlers/modules.py             | 25 ++++++++++++-------
 2 files changed, 23 insertions(+), 15 deletions(-)

diff --git a/module_build_service/models.py b/module_build_service/models.py
index 38996cc7..79126d9c 100644
--- a/module_build_service/models.py
+++ b/module_build_service/models.py
@@ -198,12 +198,13 @@ class ModuleBuild(RidaBase):
             self.state_reason = state_reason
 
         log.debug("%r, state %r->%r" % (self, old_state, self.state))
-        module_build_service.messaging.publish(
-            service='module_build_service',
-            topic='module.state.change',
-            msg=self.json(),  # Note the state is "init" here...
-            conf=conf,
-        )
+        if old_state != self.state:
+            module_build_service.messaging.publish(
+                service='module_build_service',
+                topic='module.state.change',
+                msg=self.json(),  # Note the state is "init" here...
+                conf=conf,
+            )
 
     @classmethod
     def by_state(cls, session, state):
diff --git a/module_build_service/scheduler/handlers/modules.py b/module_build_service/scheduler/handlers/modules.py
index 5fe58fc0..240d16c3 100644
--- a/module_build_service/scheduler/handlers/modules.py
+++ b/module_build_service/scheduler/handlers/modules.py
@@ -80,16 +80,23 @@ def failed(config, session, msg):
         session.commit()
         raise
 
-    builder = module_build_service.builder.GenericBuilder.create(
-        build.owner, build.name, config.system, config, tag_name=build.koji_tag)
-    builder.buildroot_connect(groups)
+    if build.koji_tag:
+        builder = module_build_service.builder.GenericBuilder.create(
+            build.owner, build.name, config.system, config, tag_name=build.koji_tag)
+        builder.buildroot_connect(groups)
 
-    for component in unbuilt_components:
-        if component.task_id:
-            builder.cancel_build(component.task_id)
-        component.state = koji.BUILD_STATES['FAILED']
-        component.state_reason = build.state_reason
-        session.add(component)
+        for component in unbuilt_components:
+            if component.task_id:
+                builder.cancel_build(component.task_id)
+            component.state = koji.BUILD_STATES['FAILED']
+            component.state_reason = build.state_reason
+            session.add(component)
+    else:
+        reason = "Missing koji tag. Assuming previously failed module lookup in PDC."
+        log.error(reason)
+        build.transition(config, state="failed", state_reason=reason)
+        session.commit()
+        return
 
     build.transition(config, state="failed")
     session.commit()

From 41187110b5e7bc040186f17c7cbb01f58d0e3d66 Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Mon, 12 Dec 2016 16:49:23 -0500
Subject: [PATCH 02/29] Add Vagrant support to submit_build.py

---
 Vagrantfile             | 1 +
 contrib/submit_build.py | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/Vagrantfile b/Vagrantfile
index 0895171d..21732364 100644
--- a/Vagrantfile
+++ b/Vagrantfile
@@ -42,6 +42,7 @@ Vagrant.configure("2") do |config|
   config.vm.synced_folder "./", "/tmp/module_build_service"
   config.vm.provision "file", source: "/var/tmp/krbcc", destination: "/var/tmp/krbcc", run: "always"
   config.vm.network "forwarded_port", guest: 5000, host: 5000
+  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"
diff --git a/contrib/submit_build.py b/contrib/submit_build.py
index 9ad3a413..de4a38f0 100644
--- a/contrib/submit_build.py
+++ b/contrib/submit_build.py
@@ -8,7 +8,7 @@ def listen_for_token():
     Listens on port 13747 on localhost for a redirect request by OIDC
     server, parses the response and returns the "access_token" value.
     """
-    TCP_IP = '127.0.0.1'
+    TCP_IP = '0.0.0.0'
     TCP_PORT = 13747
     BUFFER_SIZE = 1024
 

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 03/29] 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',

From c4091aff92f48dac2d6949ad5d0c0a500351ad71 Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Tue, 13 Dec 2016 14:39:20 -0500
Subject: [PATCH 04/29] Remove in-memory messaging

---
 conf/config.py                    |  1 -
 module_build_service/config.py    |  2 +-
 module_build_service/manage.py    |  3 --
 module_build_service/messaging.py | 70 +------------------------------
 4 files changed, 2 insertions(+), 74 deletions(-)

diff --git a/conf/config.py b/conf/config.py
index 30f4b161..5b3e39a0 100644
--- a/conf/config.py
+++ b/conf/config.py
@@ -134,7 +134,6 @@ class TestConfiguration(BaseConfiguration):
     SQLALCHEMY_DATABASE_URI = 'sqlite:///{0}'.format(
         path.join(dbdir, 'tests', 'test_module_build_service.db'))
     DEBUG = True
-    MESSAGING = 'in_memory'
 
     # Global network-related values, in seconds
     NET_TIMEOUT = 3
diff --git a/module_build_service/config.py b/module_build_service/config.py
index 35bbef6e..c265234b 100644
--- a/module_build_service/config.py
+++ b/module_build_service/config.py
@@ -344,7 +344,7 @@ class Config(object):
 
     def _setifok_messaging(self, s):
         s = str(s)
-        if s not in ("fedmsg", "amq", "in_memory"):
+        if s not in ("fedmsg", "amq"):
             raise ValueError("Unsupported messaging system.")
         self.messaging = s
 
diff --git a/module_build_service/manage.py b/module_build_service/manage.py
index 4f6133d0..8fd9fec1 100644
--- a/module_build_service/manage.py
+++ b/module_build_service/manage.py
@@ -118,7 +118,6 @@ def cleardb():
 @manager.command
 def build_module_locally(url):
     conf.set_item("system", "mock")
-    conf.set_item("messaging", "in_memory")
 
     # Use our own local SQLite3 database.
     confdir = os.path.abspath(os.path.dirname(__file__))
@@ -294,8 +293,6 @@ def runssl(host=conf.host, port=conf.port, debug=conf.debug):
     """
     logging.info('Starting Module Build Service frontend')
 
-    module_build_service.messaging.init(conf)
-
     ssl_ctx = _establish_ssl_context()
     app.run(
         host=host,
diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index 431248b8..98e25fc8 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -228,18 +228,6 @@ class RidaModule(BaseMessage):
         self.module_build_id = module_build_id
         self.module_build_state = module_build_state
 
-def init(conf, **kwargs):
-    """
-    Initialize the messaging backend.
-    :param conf: a Config object from the class in config.py
-    :param kwargs: any additional arguments to pass to the backend handler
-    """
-    try:
-        handler = _messaging_backends[conf.messaging]['init']
-    except KeyError:
-        raise KeyError("No messaging backend found for %r" % conf.messaging)
-    return handler(conf, **kwargs)
-
 def publish(topic, msg, conf, service):
     """
     Publish a single message to a given backend, and return
@@ -301,68 +289,12 @@ def _amq_publish(topic, msg, conf, service):
     msngr.put(message)
     msngr.send()
 
-# Queue for "in_memory" messaging.
-_in_memory_work_queue = queue.Queue()
-
-# Message id for "in_memory" messaging.
-_in_memory_msg_id = 0
-
-def _in_memory_init(conf, **kwargs):
-    """
-    Initializes the In Memory messaging backend.
-    """
-    global _in_memory_work_queue
-    global _in_memory_msg_id
-    _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".
-    """
-
-    # Increment the message ID.
-    global _in_memory_msg_id
-    _in_memory_msg_id += 1
-
-    # Create fake fedmsg from the message so we can reuse
-    # the BaseMessage.from_fedmsg code to get the particular BaseMessage
-    # class instance.
-    topic = service + "." + topic
-    wrapped_msg = {}
-    wrapped_msg["msg_id"] = str(_in_memory_msg_id)
-    wrapped_msg["msg"] = msg
-    wrapped_msg = BaseMessage.from_fedmsg(topic, wrapped_msg)
-
-    # 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.
-    """
-    while True:
-        yield _in_memory_work_queue.get(True)
-
-def _no_op(conf, **kwargs):
-    """
-    No operation.
-    """
-    pass
 
 _messaging_backends = {
     'fedmsg': {
-        'init': _no_op,
         'publish': _fedmsg_publish,
     },
     'amq': {
-        'init': _no_op,
         'publish': _amq_publish,
-    },
-    'in_memory': {
-        'init': _in_memory_init,
-        'publish': _in_memory_publish,
-    },
+    }
 }

From f11caf32e5da1ace28daecd0ff4d7f6202613348 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 14:41:08 -0500
Subject: [PATCH 05/29] Get the basic framework for local module builds working
 again with hub/consumer.

---
 module_build_service/builder.py            |  5 +-
 module_build_service/manage.py             | 58 +++++++++++++++--
 module_build_service/scheduler/consumer.py | 75 ++++++++++++++++------
 3 files changed, 114 insertions(+), 24 deletions(-)

diff --git a/module_build_service/builder.py b/module_build_service/builder.py
index 9c1941cd..10f6e770 100644
--- a/module_build_service/builder.py
+++ b/module_build_service/builder.py
@@ -55,6 +55,7 @@ from module_build_service.models import ModuleBuild
 import module_build_service.scm
 import module_build_service.utils
 import module_build_service.scheduler
+import module_build_service.scheduler.consumer
 
 logging.basicConfig(level=logging.DEBUG)
 
@@ -1165,7 +1166,7 @@ $repos
             msg_id='a faked internal message',
             repo_tag=self.tag_name + "-build",
         )
-        module_build_service.scheduler.work_queue.put(msg)
+        module_build_service.scheduler.consumer.work_queue_put(msg)
 
     def _send_build_change(self, state, source, build_id):
         nvr = kobo.rpmlib.parse_nvr(source)
@@ -1181,7 +1182,7 @@ $repos
             build_release=nvr["release"],
             build_version=nvr["version"]
         )
-        module_build_service.scheduler.main.outgoing_work_queue_put(msg)
+        module_build_service.scheduler.consumer.work_queue_put(msg)
 
     def _save_log(self, log_name, artifact_name):
         old_log = os.path.join(self.resultsdir, log_name)
diff --git a/module_build_service/manage.py b/module_build_service/manage.py
index 8fd9fec1..c93303d8 100644
--- a/module_build_service/manage.py
+++ b/module_build_service/manage.py
@@ -29,6 +29,11 @@ import ssl
 from shutil import rmtree
 import getpass
 
+import fedmsg.config
+import moksha.hub
+import moksha.hub.hub
+import moksha.hub.reactor
+
 from module_build_service import app, conf, db
 from module_build_service import models
 from module_build_service.pdc import (
@@ -40,6 +45,7 @@ from module_build_service.utils import (
 )
 from module_build_service.messaging import RidaModule
 import module_build_service.messaging
+import module_build_service.scheduler.consumer
 
 
 manager = Manager(app)
@@ -144,10 +150,54 @@ def build_module_locally(url):
     username = getpass.getuser()
     submit_module_build(username, url, allow_local_url=True)
 
-    # TODO: Ralph to the rescue
-    # msgs = []
-    # msgs.append(RidaModule("local module build", 2, 1))
-    # module_build_service.scheduler.main.main(msgs, True)
+    def stop_condition(message):
+        # XXX - We ignore the message here and instead just query the DB.
+
+        # Grab the latest module build.
+        module = db.session.query(models.ModuleBuild)\
+            .order_by(models.ModuleBuild.id.desc())\
+            .first()
+        done = (
+            module_build_service.models.BUILD_STATES["failed"],
+            module_build_service.models.BUILD_STATES["ready"],
+            # XXX should this one be removed?
+            module_build_service.models.BUILD_STATES["done"],
+        )
+        return module.state in done
+
+    config = fedmsg.config.load_config()
+    config['mbsconsumer'] = True
+    config['mbsconsumer.stopcondition'] = stop_condition
+    config['mbsconsumer.initial_messages'] = [
+        RidaModule("local module build", 2, 1)
+    ]
+
+    consumers = [module_build_service.scheduler.consumer.MBSConsumer]
+
+    # Rephrase the fedmsg-config.py config as moksha *.ini format for
+    # zeromq. If we're not using zeromq (say, we're using STOMP), then just
+    # assume that the moksha configuration is specified correctly already
+    # in /etc/fedmsg.d/
+    if config.get('zmq_enabled', True):
+        moksha_options = dict(
+            # XXX - replace this with a /dev/null endpoint.
+            zmq_subscribe_endpoints=','.join(
+                ','.join(bunch) for bunch in config['endpoints'].values()
+            ),
+        )
+        config.update(moksha_options)
+
+    # Note that the hub we kick off here cannot send any message.  You
+    # should use fedmsg.publish(...) still for that.
+    from moksha.hub import main
+    main(
+        # Pass in our config dict
+        options=config,
+        # Only run the specified consumers if any are so specified.
+        consumers=consumers,
+        # Tell moksha to quiet its logging.
+        framework=False,
+    )
 
 
 @manager.command
diff --git a/module_build_service/scheduler/consumer.py b/module_build_service/scheduler/consumer.py
index 761c20d2..d3485f6b 100644
--- a/module_build_service/scheduler/consumer.py
+++ b/module_build_service/scheduler/consumer.py
@@ -28,6 +28,7 @@ to use.
 import koji
 import inspect
 import fedmsg.consumers
+import moksha.hub
 
 from module_build_service.utils import module_build_state_from_msg
 import module_build_service.messaging
@@ -45,11 +46,15 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
     topic = '*'
     config_key = 'mbsconsumer'
 
-    def __init__(self, hub, initial_msgs=[]):
+    def __init__(self, hub):
         super(MBSConsumer, self).__init__(hub)
 
-        for msg in initial_msgs:
-            work_queue.put(msg)
+        # These two values are typically provided either by the unit tests or
+        # by the local build command.  They are empty in the production environ
+        self.stop_condition = hub.config.get('mbsconsumer.stop_condition')
+        initial_messages = hub.config.get('mbsconsumer.initial_messages', [])
+        for msg in initial_messages:
+            self.incoming.put(msg)
 
         # These are our main lookup tables for figuring out what to run in
         # response to what messaging events.
@@ -80,19 +85,35 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
         self.on_repo_change = module_build_service.scheduler.handlers.repos.done
         self.sanity_check()
 
+    def shutdown(self):
+        log.info("Shutting down..")
+        self.hub.stop()
+        from moksha.hub.reactor import reactor
+        reactor.callFromThread(reactor.stop)
+
     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)
+        log.info("Received %r" % message)
+
+        if self.stop_condition and self.stop_condition(message):
+            return self.shutdown()
+
+        # Sometimes, the messages put into our queue are artificially put there
+        # by other parts of our own codebase.  If they are already abstracted
+        # messages, then just use them as-is.  If they are not already
+        # instances of our message abstraction base class, then first transform
+        # them before proceeding.
+        if isinstance(message, module_build_service.messaging.BaseMessage):
+            msg = message
+        else:
+            msg = self.get_abstracted_msg(message['body'])
+
+        # Primary work is done here.
+        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
@@ -142,8 +163,7 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
             return
 
         # Execute our chosen handler
-        idx = "%s: %s, %s" % (handler.__name__, type(msg).__name__,
-                              msg.msg_id)
+        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:
@@ -159,4 +179,23 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
             # 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)
+                self.incoming.put(event)
+
+
+def get_global_consumer():
+    """ Return a handle to the active consumer object, if it exists. """
+    hub = moksha.hub._hub
+    if not hub:
+        raise ValueError("No global moksha-hub obj found.")
+
+    for consumer in hub.consumers:
+        if isinstance(consumer, MBSConsumer):
+            return consumer
+
+    raise ValueError("No MBSConsumer found.")
+
+
+def work_queue_put(msg):
+    """ Artificially put a message into the work queue of the consumer. """
+    consumer = get_global_consumer()
+    consumer.incoming.put(msg)

From 250b9510bc72a48d44c9da2f85a399096adfaf6d Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 14:44:07 -0500
Subject: [PATCH 06/29] Have the producer interact with the consumer queue
 appropriately.

---
 module_build_service/scheduler/producer.py | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/module_build_service/scheduler/producer.py b/module_build_service/scheduler/producer.py
index 980d2757..619f4b48 100644
--- a/module_build_service/scheduler/producer.py
+++ b/module_build_service/scheduler/producer.py
@@ -30,8 +30,8 @@ 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
+import module_build_service.scheduler.consumer
 from module_build_service import conf, models, log
 
 
@@ -94,7 +94,7 @@ class MBSProducer(PollingProducer):
                         build_release=None,
                         build_version=None
                     )
-                    work_queue.put(msg)
+                    module_build_service.scheduler.consumer.work_queue_put(msg)
 
         elif conf.system == 'copr':
             # @TODO
@@ -105,7 +105,8 @@ class MBSProducer(PollingProducer):
 
     def log_summary(self, session):
         log.info('Current status:')
-        backlog = work_queue.qsize()
+        consumer = module_build_service.scheduler.consumer.get_global_consumer()
+        backlog = consumer.incoming.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:
@@ -157,4 +158,4 @@ class MBSProducer(PollingProducer):
                     config, module_build, session, config.system)
                 for event in further_work:
                     log.info("  Scheduling faked event %r" % event)
-                    work_queue.put(event)
+                    module_build_service.scheduler.consumer.work_queue_put(event)

From 5fe359a42915c0080297ebc172b9be9acf83b432 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 14:45:34 -0500
Subject: [PATCH 07/29] Comments, updated to reflect new reality.

---
 module_build_service/builder.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/module_build_service/builder.py b/module_build_service/builder.py
index 10f6e770..992322a3 100644
--- a/module_build_service/builder.py
+++ b/module_build_service/builder.py
@@ -1206,7 +1206,7 @@ $repos
                                "--resultdir=%s" % self.resultsdir])
 
             # Emit messages simulating complete build. These messages
-            # are put in the scheduler.main._work_queue and are handled
+            # are put in the scheduler's work queue and are handled
             # by MBS after the build_srpm() method returns and scope gets
             # back to scheduler.main.main() method.
             self._send_repo_done()
@@ -1221,7 +1221,7 @@ $repos
                       str(e)))
 
             # Emit messages simulating complete build. These messages
-            # are put in the scheduler.main._work_queue and are handled
+            # are put in the scheduler's work queue and are handled
             # by MBS after the build_srpm() method returns and scope gets
             # back to scheduler.main.main() method.
             self._send_repo_done()

From feb1079fbae7f912fa778941af42d940379c035e Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 14:45:39 -0500
Subject: [PATCH 08/29] Remove unused work_queue.

---
 module_build_service/scheduler/__init__.py | 4 +---
 module_build_service/scheduler/consumer.py | 1 -
 2 files changed, 1 insertion(+), 4 deletions(-)

diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index fcb40e2d..d5b2f216 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -1,3 +1 @@
-import six.moves.queue as queue
-
-work_queue = queue.Queue()
+""" This is a sub-module for backend/scheduler functionality. """
diff --git a/module_build_service/scheduler/consumer.py b/module_build_service/scheduler/consumer.py
index d3485f6b..3c0bffb8 100644
--- a/module_build_service/scheduler/consumer.py
+++ b/module_build_service/scheduler/consumer.py
@@ -35,7 +35,6 @@ 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
 
 

From 4b39eda44ff3c0778f2ded01b6a55f779e560411 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 15:31:37 -0500
Subject: [PATCH 09/29] Move the moksha.hub.main code out into its own main
 function again.

---
 module_build_service/manage.py             | 51 ++--------------
 module_build_service/scheduler/__init__.py | 69 ++++++++++++++++++++++
 2 files changed, 73 insertions(+), 47 deletions(-)

diff --git a/module_build_service/manage.py b/module_build_service/manage.py
index c93303d8..21f2ef4b 100644
--- a/module_build_service/manage.py
+++ b/module_build_service/manage.py
@@ -150,54 +150,11 @@ def build_module_locally(url):
     username = getpass.getuser()
     submit_module_build(username, url, allow_local_url=True)
 
-    def stop_condition(message):
-        # XXX - We ignore the message here and instead just query the DB.
+    stop = module_build_service.scheduler.make_simple_stop_condition(db.session)
+    initial_messages = [RidaModule("local module build", 2, 1)]
 
-        # Grab the latest module build.
-        module = db.session.query(models.ModuleBuild)\
-            .order_by(models.ModuleBuild.id.desc())\
-            .first()
-        done = (
-            module_build_service.models.BUILD_STATES["failed"],
-            module_build_service.models.BUILD_STATES["ready"],
-            # XXX should this one be removed?
-            module_build_service.models.BUILD_STATES["done"],
-        )
-        return module.state in done
-
-    config = fedmsg.config.load_config()
-    config['mbsconsumer'] = True
-    config['mbsconsumer.stopcondition'] = stop_condition
-    config['mbsconsumer.initial_messages'] = [
-        RidaModule("local module build", 2, 1)
-    ]
-
-    consumers = [module_build_service.scheduler.consumer.MBSConsumer]
-
-    # Rephrase the fedmsg-config.py config as moksha *.ini format for
-    # zeromq. If we're not using zeromq (say, we're using STOMP), then just
-    # assume that the moksha configuration is specified correctly already
-    # in /etc/fedmsg.d/
-    if config.get('zmq_enabled', True):
-        moksha_options = dict(
-            # XXX - replace this with a /dev/null endpoint.
-            zmq_subscribe_endpoints=','.join(
-                ','.join(bunch) for bunch in config['endpoints'].values()
-            ),
-        )
-        config.update(moksha_options)
-
-    # Note that the hub we kick off here cannot send any message.  You
-    # should use fedmsg.publish(...) still for that.
-    from moksha.hub import main
-    main(
-        # Pass in our config dict
-        options=config,
-        # Only run the specified consumers if any are so specified.
-        consumers=consumers,
-        # Tell moksha to quiet its logging.
-        framework=False,
-    )
+    # Run the consumer until stop_condition returns True
+    module_build_service.scheduler.main(initial_messages, stop)
 
 
 @manager.command
diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index d5b2f216..3f925a1a 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -1 +1,70 @@
 """ This is a sub-module for backend/scheduler functionality. """
+
+import fedmsg
+import moksha.hub
+
+import module_build_service.scheduler.consumer
+
+
+def main(initial_messages, stop_condition):
+    """ Run the consumer until some condition is met.
+
+    Setting stop_condition to None will run the consumer forever.
+    """
+
+    config = fedmsg.config.load_config()
+    config['mbsconsumer'] = True
+    config['mbsconsumer.stopcondition'] = stop_condition
+    config['mbsconsumer.initial_messages'] = initial_messages
+
+    consumers = [module_build_service.scheduler.consumer.MBSConsumer]
+
+    # Rephrase the fedmsg-config.py config as moksha *.ini format for
+    # zeromq. If we're not using zeromq (say, we're using STOMP), then just
+    # assume that the moksha configuration is specified correctly already
+    # in /etc/fedmsg.d/
+    if config.get('zmq_enabled', True):
+        moksha_options = dict(
+            # XXX - replace this with a /dev/null endpoint.
+            zmq_subscribe_endpoints=','.join(
+                ','.join(bunch) for bunch in config['endpoints'].values()
+            ),
+        )
+        config.update(moksha_options)
+
+    # Note that the hub we kick off here cannot send any message.  You
+    # should use fedmsg.publish(...) still for that.
+    moksha.hub.main(
+        # Pass in our config dict
+        options=config,
+        # Only run the specified consumers if any are so specified.
+        consumers=consumers,
+        # Tell moksha to quiet its logging.
+        framework=False,
+    )
+
+def make_simple_stop_condition(session):
+    """ Return a simple stop_condition callable.
+
+    Intended to be used with the main() function here in manage.py and tests.
+
+    The stop_condition returns true when the latest module build enters the any
+    of the finished states.
+    """
+
+    def stop_condition(message):
+        # XXX - We ignore the message here and instead just query the DB.
+
+        # Grab the latest module build.
+        module = session.query(models.ModuleBuild)\
+            .order_by(models.ModuleBuild.id.desc())\
+            .first()
+        done = (
+            module_build_service.models.BUILD_STATES["failed"],
+            module_build_service.models.BUILD_STATES["ready"],
+            # XXX should this one be removed?
+            module_build_service.models.BUILD_STATES["done"],
+        )
+        return module.state in done
+
+    return stop_condition

From 7e3ee9b0b3b1373d084669d5d72003a9f16e9e9d Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 15:32:02 -0500
Subject: [PATCH 10/29] Begin porting the tests to use the new main function,
 but this doesn't work yet.

---
 tests/test_build/test_build.py | 39 ++++++++++++++--------------------
 1 file changed, 16 insertions(+), 23 deletions(-)

diff --git a/tests/test_build/test_build.py b/tests/test_build/test_build.py
index 9ff91d38..b23165c9 100644
--- a/tests/test_build/test_build.py
+++ b/tests/test_build/test_build.py
@@ -21,31 +21,24 @@
 # Written by Jan Kaluza <jkaluza@redhat.com>
 
 import unittest
-import munch
-import mock
 import koji
-import xmlrpclib
 from os import path, mkdir
 from shutil import copyfile
 
 from nose.tools import timed
 
-from module_build_service import db
-
 import module_build_service.messaging
 import module_build_service.scheduler.handlers.repos
-from module_build_service import models, conf
-from module_build_service.utils import submit_module_build
+from module_build_service import db, models, conf
 from module_build_service.messaging import RidaModule
 
 from mock import patch
 
 from tests import app, init_data
-from tests import conf as test_conf
 import json
 
 from module_build_service.builder import KojiModuleBuilder, GenericBuilder
-import module_build_service.scheduler.main
+import module_build_service.scheduler.consumer
 
 
 class MockedSCM(object):
@@ -128,7 +121,7 @@ class TestModuleBuilder(GenericBuilder):
             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.consumer.work_queue_put(msg)
 
     def _send_build_change(self, state, source, build_id):
         # build_id=1 and task_id=1 are OK here, because we are building just
@@ -142,7 +135,7 @@ class TestModuleBuilder(GenericBuilder):
             build_release="1",
             build_version="1"
         )
-        module_build_service.scheduler.main.outgoing_work_queue_put(msg)
+        module_build_service.scheduler.consumer.work_queue_put(msg)
 
     def build(self, artifact_name, source):
         print "Starting building artifact %s: %s" % (artifact_name, source)
@@ -202,7 +195,7 @@ class TestBuild(unittest.TestCase):
         Tests the build of testmodule.yaml using TestModuleBuilder which
         succeeds everytime.
         """
-        mocked_scm_obj = MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
+        MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
 
         rv = self.client.post('/module-build-service/1/module-builds/', data=json.dumps(
             {'scmurl': 'git://pkgs.stg.fedoraproject.org/modules/'
@@ -211,9 +204,9 @@ class TestBuild(unittest.TestCase):
         data = json.loads(rv.data)
         module_build_id = data['id']
 
-        msgs = []
-        msgs.append(RidaModule("fake msg", 1, 1))
-        module_build_service.scheduler.main.main(msgs, True)
+        msgs = [RidaModule("fake msg", 1, 1)]
+        stop = module_build_service.scheduler.make_simple_stop_condition(db.session)
+        module_build_service.scheduler.main(msgs, stop)
 
         # All components should be built and module itself should be in "done"
         # or "ready" state.
@@ -230,7 +223,7 @@ class TestBuild(unittest.TestCase):
         """
         Submit all builds for a module and cancel the module build later.
         """
-        mocked_scm_obj = MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
+        MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
 
         rv = self.client.post('/module-build-service/1/module-builds/', data=json.dumps(
             {'scmurl': 'git://pkgs.stg.fedoraproject.org/modules/'
@@ -257,9 +250,9 @@ class TestBuild(unittest.TestCase):
         TestModuleBuilder.on_build_cb = on_build_cb
         TestModuleBuilder.on_cancel_cb = on_cancel_cb
 
-        msgs = []
-        msgs.append(RidaModule("fake msg", 1, 1))
-        module_build_service.scheduler.main.main(msgs, True)
+        msgs = [RidaModule("fake msg", 1, 1)]
+        stop = module_build_service.scheduler.make_simple_stop_condition(db.session)
+        module_build_service.scheduler.main(msgs, stop)
 
         # Because we did not finished single component build and canceled the
         # module build, all components and even the module itself should be in
@@ -284,7 +277,7 @@ class TestBuild(unittest.TestCase):
         Tests the build of testmodule.yaml using TestModuleBuilder which
         succeeds everytime.
         """
-        mocked_scm_obj = MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
+        MockedSCM(mocked_scm, "testmodule", "testmodule.yaml")
 
         rv = self.client.post('/module-build-service/1/module-builds/', data=json.dumps(
             {'scmurl': 'git://pkgs.stg.fedoraproject.org/modules/'
@@ -296,9 +289,9 @@ class TestBuild(unittest.TestCase):
         TestModuleBuilder.BUILD_STATE = "BUILDING"
         TestModuleBuilder.INSTANT_COMPLETE = True
 
-        msgs = []
-        msgs.append(RidaModule("fake msg", 1, 1))
-        module_build_service.scheduler.main.main(msgs, True)
+        msgs = [RidaModule("fake msg", 1, 1)]
+        stop = module_build_service.scheduler.make_simple_stop_condition(db.session)
+        module_build_service.scheduler.main(msgs, stop)
 
         # All components should be built and module itself should be in "done"
         # or "ready" state.

From c3f73e76debafa8de5f8fe41ee29eff4703c38d7 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 15:35:44 -0500
Subject: [PATCH 11/29] Disable these testmodule integration tests for now.

---
 tests/test_build/test_build.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/tests/test_build/test_build.py b/tests/test_build/test_build.py
index b23165c9..f109b33a 100644
--- a/tests/test_build/test_build.py
+++ b/tests/test_build/test_build.py
@@ -170,6 +170,7 @@ class TestModuleBuilder(GenericBuilder):
             TestModuleBuilder.on_cancel_cb(self, task_id)
 
 
+@unittest.skip("These are disabled until ralph can figure out how to run the reactor in the tests.")
 class TestBuild(unittest.TestCase):
 
     def setUp(self):

From 63beecbb0eafe3d7210e726b359990866e7120a8 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Wed, 14 Dec 2016 15:38:20 -0500
Subject: [PATCH 12/29] Remove tests of deleted code.

---
 .../test_messaging_functions.py               | 82 -------------------
 1 file changed, 82 deletions(-)

diff --git a/tests/test_messaging/test_messaging_functions.py b/tests/test_messaging/test_messaging_functions.py
index df4e650e..42100954 100644
--- a/tests/test_messaging/test_messaging_functions.py
+++ b/tests/test_messaging/test_messaging_functions.py
@@ -21,9 +21,7 @@
 # Written by Matt Prahl <mprahl@redhat.com>
 from __future__ import unicode_literals
 import unittest
-import mock
 from datetime import datetime
-import module_build_service.messaging
 
 
 class TestUtilFunctions(unittest.TestCase):
@@ -42,83 +40,3 @@ class TestUtilFunctions(unittest.TestCase):
         'time_submitted': datetime(2016, 9, 1, 2, 28),
         'version': '4.3.43'
     }
-
-    @mock.patch('fedmsg.tail_messages')
-    def test_fedmsg_listen_build_msg(self, mock_tail_messages):
-        endpoint = 'tcp://hub.fedoraproject.org:9940'
-        topic = 'org.fedoraproject.prod.buildsys.build.state.change'
-        msg = {
-            'source_name': 'datanommer',
-            'i': 2,
-            'timestamp': 1473252386.0,
-            'msg_id': '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c',
-            'topic': 'org.fedoraproject.prod.buildsys.build.state.change',
-            'source_version': '1.2.3',
-            'msg': {
-                'build_id': 2345678,
-                'old': None,
-                'name': 'some-package',
-                'task_id': 1234567,
-                'attribute': 'state',
-                'instance': 'arm',
-                'version': '2.1.0',
-                'owner': 'some_owner',
-                'new': 0,
-                'release': '1.fc26'
-            }
-        }
-        mock_tail_messages.side_effect = \
-            lambda: [('fedora-infrastructure', endpoint, topic, msg)]
-        msg_obj = next(module_build_service.messaging._fedmsg_listen(None))
-        self.assertEquals(type(msg_obj), module_build_service.messaging.KojiBuildChange)
-        self.assertEquals(msg_obj.build_id, 2345678)
-        self.assertEquals(msg_obj.task_id, 1234567)
-        self.assertEquals(msg_obj.build_new_state, 0)
-        self.assertEquals(msg_obj.build_name, 'some-package')
-        self.assertEquals(msg_obj.build_version, '2.1.0')
-        self.assertEquals(msg_obj.build_release, '1.fc26')
-        self.assertEquals(msg_obj.msg_id,
-                          '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c')
-
-    @mock.patch('fedmsg.tail_messages')
-    def test_fedmsg_listen_repo_msg(self, mock_tail_messages):
-        endpoint = 'tcp://hub.fedoraproject.org:9940'
-        topic = 'org.fedoraproject.prod.buildsys.repo.done'
-        msg = {
-            'source_name': 'datanommer',
-            'i': 1,
-            'timestamp': 1473252506.0,
-            'msg_id': '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c',
-            'topic': 'org.fedoraproject.prod.buildsys.repo.done',
-            'source_version': '1.2.0',
-            'msg': {
-                'instance': 'arm',
-                'repo_id': 402102,
-                'tag': 'f23-build',
-                'tag_id': 155
-            }
-        }
-        mock_tail_messages.side_effect = \
-            lambda: [('fedora-infrastructure', endpoint, topic, msg)]
-        msg_obj = next(module_build_service.messaging._fedmsg_listen(None))
-        self.assertEquals(type(msg_obj), module_build_service.messaging.KojiRepoChange)
-        self.assertEquals(msg_obj.repo_tag, 'f23-build')
-        self.assertEquals(msg_obj.msg_id,
-                          '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c')
-
-    @mock.patch('fedmsg.tail_messages')
-    def test_fedmsg_listen_module_build_service_msg(self, mock_tail_messages):
-        endpoint = 'tcp://hub.fedoraproject.org:9940'
-        topic = 'org.fedoraproject.prod.module_build_service.module.state.change'
-        msg = {
-            'msg_id': '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c',
-            'topic': 'org.fedoraproject.prod.module_build_service.module.state.change',
-            'msg': self.module_build_service_msg
-        }
-        mock_tail_messages.side_effect = \
-            lambda: [('fedora-infrastructure', endpoint, topic, msg)]
-        msg_obj = next(module_build_service.messaging._fedmsg_listen(None))
-        self.assertEquals(msg_obj.module_build_id, msg['msg']['id'])
-        self.assertEquals(msg_obj.module_build_state, msg['msg']['state'])
-        self.assertEquals(msg_obj.msg_id,
-                          '2016-e05415d9-9b35-4f13-8b25-0daddeabfb8c')

From 2e9e31eb5656fa15082a69ffe99382b42e30578a Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Wed, 14 Dec 2016 15:40:40 -0500
Subject: [PATCH 13/29] Fix the 'process_waiting_module_builds' function in the
 scheduler

---
 module_build_service/scheduler/producer.py | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/module_build_service/scheduler/producer.py b/module_build_service/scheduler/producer.py
index 619f4b48..837780f5 100644
--- a/module_build_service/scheduler/producer.py
+++ b/module_build_service/scheduler/producer.py
@@ -30,6 +30,7 @@ from datetime import timedelta
 from sqlalchemy.orm import lazyload
 from moksha.hub.api.producer import PollingProducer
 
+import module_build_service.messaging
 import module_build_service.scheduler
 import module_build_service.scheduler.consumer
 from module_build_service import conf, models, log
@@ -131,12 +132,16 @@ class MBSProducer(PollingProducer):
                  .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)
+            msg = module_build_service.messaging.RidaModule(
+                'fake message',
+                build.id,
+                module_build_service.models.BUILD_STATES['wait']
+            )
+            further_work = module_build_service.scheduler.handlers.modules.wait(
+                conf, session, msg) or []
+            for event in further_work:
+                log.info("  Scheduling faked event %r" % event)
+                module_build_service.scheduler.consumer.work_queue_put(event)
 
     def process_open_component_builds(self, session):
         log.warning('process_open_component_builds is not yet implemented...')

From 40afb55ccf56ef2472c0acbb5b1f6a217dcca63d Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 11:55:03 -0500
Subject: [PATCH 14/29] Revert "Disable these testmodule integration tests for
 now."

This reverts commit c2853026cc0e89e1cb34ed884af4e64a2da39052.
---
 tests/test_build/test_build.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/test_build/test_build.py b/tests/test_build/test_build.py
index f109b33a..b23165c9 100644
--- a/tests/test_build/test_build.py
+++ b/tests/test_build/test_build.py
@@ -170,7 +170,6 @@ class TestModuleBuilder(GenericBuilder):
             TestModuleBuilder.on_cancel_cb(self, task_id)
 
 
-@unittest.skip("These are disabled until ralph can figure out how to run the reactor in the tests.")
 class TestBuild(unittest.TestCase):
 
     def setUp(self):

From b2cf02080dc85517abbd2f5df72c539cc6f101e5 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 12:18:40 -0500
Subject: [PATCH 15/29] A few bugfixes.

---
 module_build_service/scheduler/__init__.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index 3f925a1a..232ef74d 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -3,6 +3,7 @@
 import fedmsg
 import moksha.hub
 
+import module_build_service.models
 import module_build_service.scheduler.consumer
 
 
@@ -14,7 +15,7 @@ def main(initial_messages, stop_condition):
 
     config = fedmsg.config.load_config()
     config['mbsconsumer'] = True
-    config['mbsconsumer.stopcondition'] = stop_condition
+    config['mbsconsumer.stop_condition'] = stop_condition
     config['mbsconsumer.initial_messages'] = initial_messages
 
     consumers = [module_build_service.scheduler.consumer.MBSConsumer]
@@ -56,8 +57,8 @@ def make_simple_stop_condition(session):
         # XXX - We ignore the message here and instead just query the DB.
 
         # Grab the latest module build.
-        module = session.query(models.ModuleBuild)\
-            .order_by(models.ModuleBuild.id.desc())\
+        module = session.query(module_build_service.models.ModuleBuild)\
+            .order_by(module_build_service.models.ModuleBuild.id.desc())\
             .first()
         done = (
             module_build_service.models.BUILD_STATES["failed"],

From 29747a7ea7956d9b778cdde40a9252439fd6d7c0 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 12:18:51 -0500
Subject: [PATCH 16/29] Allow restarting the reactor for multiple tests.

---
 tests/test_build/test_build.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/tests/test_build/test_build.py b/tests/test_build/test_build.py
index b23165c9..66f36ed8 100644
--- a/tests/test_build/test_build.py
+++ b/tests/test_build/test_build.py
@@ -185,6 +185,12 @@ class TestBuild(unittest.TestCase):
         conf.set_item("system", "koji")
         TestModuleBuilder.reset()
 
+        # Necessary to restart the twisted reactor for the next test.
+        import sys
+        del sys.modules['twisted.internet.reactor']
+        del sys.modules['moksha.hub.reactor']
+        import moksha.hub.reactor
+
     @timed(30)
     @patch('module_build_service.auth.get_username', return_value='Homer J. Simpson')
     @patch('module_build_service.auth.assert_is_packager')

From 21a902ab5e52637dd58004522cce854b7f125d65 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 12:19:30 -0500
Subject: [PATCH 17/29] Some debugging on the stop_condition.

---
 module_build_service/scheduler/__init__.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index 232ef74d..fd1903c7 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -6,6 +6,9 @@ import moksha.hub
 import module_build_service.models
 import module_build_service.scheduler.consumer
 
+import logging
+log = logging.getLogger(__name__)
+
 
 def main(initial_messages, stop_condition):
     """ Run the consumer until some condition is met.
@@ -66,6 +69,8 @@ def make_simple_stop_condition(session):
             # XXX should this one be removed?
             module_build_service.models.BUILD_STATES["done"],
         )
-        return module.state in done
+        result = module.state in done
+        log.debug("stop_condition checking %r, got %r" % (module, result))
+        return result
 
     return stop_condition

From a94cd53afd25fa034c47a3f99d30ee2fe5579867 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 12:19:46 -0500
Subject: [PATCH 18/29] Make this more easily discoverable by nosetests.

---
 tests/test_build/__init__.py | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 create mode 100644 tests/test_build/__init__.py

diff --git a/tests/test_build/__init__.py b/tests/test_build/__init__.py
new file mode 100644
index 00000000..e69de29b

From 9d3b85426d5e1e3193488851fced80f3019a1f6e Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:13:05 -0500
Subject: [PATCH 19/29] Bring back the in-memory messaging backend.

---
 conf/config.py                             |  1 +
 module_build_service/config.py             |  2 +-
 module_build_service/messaging.py          | 39 +++++++++++++++++++++-
 module_build_service/scheduler/consumer.py |  8 ++++-
 4 files changed, 47 insertions(+), 3 deletions(-)

diff --git a/conf/config.py b/conf/config.py
index 5b3e39a0..825dce34 100644
--- a/conf/config.py
+++ b/conf/config.py
@@ -134,6 +134,7 @@ class TestConfiguration(BaseConfiguration):
     SQLALCHEMY_DATABASE_URI = 'sqlite:///{0}'.format(
         path.join(dbdir, 'tests', 'test_module_build_service.db'))
     DEBUG = True
+    MESSAGING = 'memory'
 
     # Global network-related values, in seconds
     NET_TIMEOUT = 3
diff --git a/module_build_service/config.py b/module_build_service/config.py
index c265234b..aca60a8d 100644
--- a/module_build_service/config.py
+++ b/module_build_service/config.py
@@ -344,7 +344,7 @@ class Config(object):
 
     def _setifok_messaging(self, s):
         s = str(s)
-        if s not in ("fedmsg", "amq"):
+        if s not in ("fedmsg", "amq", "memory"):
             raise ValueError("Unsupported messaging system.")
         self.messaging = s
 
diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index 98e25fc8..9f991ec1 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -33,7 +33,7 @@ except ImportError:
     from funcsigs import signature
 
 from module_build_service import log
-import six.moves.queue as queue
+
 
 class BaseMessage(object):
     def __init__(self, msg_id):
@@ -248,6 +248,40 @@ def _fedmsg_publish(topic, msg, conf, service):
     import fedmsg
     return fedmsg.publish(topic, msg=msg, modname=service)
 
+
+# A counter used for in-memory messages.
+_in_memory_msg_id = 0
+_initial_messages = []
+
+
+def _in_memory_publish(topic, msg, conf, service):
+    """ Puts the message into the in memory work queue. """
+    # Increment the message ID.
+    global _in_memory_msg_id
+    _in_memory_msg_id += 1
+
+    # Create fake fedmsg from the message so we can reuse
+    # the BaseMessage.from_fedmsg code to get the particular BaseMessage
+    # class instance.
+    wrapped_msg = BaseMessage.from_fedmsg(
+        service + "." + topic,
+        {"msg_id": str(_in_memory_msg_id), "msg": msg},
+    )
+
+    # Put the message to queue.
+    from module_build_service.scheduler.consumer import work_queue_put
+    try:
+        work_queue_put(wrapped_msg)
+    except ValueError as e:
+        log.warn("No MBSConsumer found.  Shutting down?  %r" % e)
+    except AttributeError as e:
+        # In the event that `moksha.hub._hub` hasn't yet been initialized, we
+        # need to store messages on the side until it becomes available.
+        # As a last-ditch effort, try to hang initial messages in the config.
+        log.warn("Hub not initialized.  Queueing on the side.")
+        _initial_messages.append(wrapped_msg)
+
+
 def _amq_get_messenger(conf):
     import proton
     for attr in ('amq_private_key_file', 'amq_trusted_cert_file', 'amq_cert_file'):
@@ -296,5 +330,8 @@ _messaging_backends = {
     },
     'amq': {
         'publish': _amq_publish,
+    },
+    'memory': {
+        'publish': _in_memory_publish,
     }
 }
diff --git a/module_build_service/scheduler/consumer.py b/module_build_service/scheduler/consumer.py
index 3c0bffb8..27fd0f29 100644
--- a/module_build_service/scheduler/consumer.py
+++ b/module_build_service/scheduler/consumer.py
@@ -55,6 +55,12 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
         for msg in initial_messages:
             self.incoming.put(msg)
 
+        # Furthermore, extend our initial messages with any that were queued up
+        # in the test environment before our hub was initialized.
+        while module_build_service.messaging._initial_messages:
+            msg = module_build_service.messaging._initial_messages.pop()
+            self.incoming.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
@@ -191,7 +197,7 @@ def get_global_consumer():
         if isinstance(consumer, MBSConsumer):
             return consumer
 
-    raise ValueError("No MBSConsumer found.")
+    raise ValueError("No MBSConsumer found among %r." % len(hub.consumers))
 
 
 def work_queue_put(msg):

From 775819e87c7fdab3be95b6e31ee974fa61329850 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:13:22 -0500
Subject: [PATCH 20/29] Bugfix to shutdown and restart for the test suite.

---
 module_build_service/scheduler/consumer.py | 10 +++++-----
 tests/test_build/test_build.py             |  1 +
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/module_build_service/scheduler/consumer.py b/module_build_service/scheduler/consumer.py
index 27fd0f29..9896b7d6 100644
--- a/module_build_service/scheduler/consumer.py
+++ b/module_build_service/scheduler/consumer.py
@@ -91,17 +91,14 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
         self.sanity_check()
 
     def shutdown(self):
-        log.info("Shutting down..")
-        self.hub.stop()
+        log.info("Scheduling shutdown.")
         from moksha.hub.reactor import reactor
+        reactor.callFromThread(self.hub.stop)
         reactor.callFromThread(reactor.stop)
 
     def consume(self, message):
         log.info("Received %r" % message)
 
-        if self.stop_condition and self.stop_condition(message):
-            return self.shutdown()
-
         # Sometimes, the messages put into our queue are artificially put there
         # by other parts of our own codebase.  If they are already abstracted
         # messages, then just use them as-is.  If they are not already
@@ -120,6 +117,9 @@ class MBSConsumer(fedmsg.consumers.FedmsgConsumer):
             log.exception('Failed while handling {0!r}'.format(msg.msg_id))
             log.info(msg)
 
+        if self.stop_condition and self.stop_condition(message):
+            self.shutdown()
+
     def get_abstracted_msg(self, message):
         # Convert the message to an abstracted message
         if conf.messaging == 'fedmsg':
diff --git a/tests/test_build/test_build.py b/tests/test_build/test_build.py
index 66f36ed8..f6aff7b5 100644
--- a/tests/test_build/test_build.py
+++ b/tests/test_build/test_build.py
@@ -189,6 +189,7 @@ class TestBuild(unittest.TestCase):
         import sys
         del sys.modules['twisted.internet.reactor']
         del sys.modules['moksha.hub.reactor']
+        del sys.modules['moksha.hub']
         import moksha.hub.reactor
 
     @timed(30)

From 3bf952cda67b1ff00b95bf2bd7acb12d7db02c5c Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:24:29 -0500
Subject: [PATCH 21/29] Tell moksha to just listen to /dev/null for local
 builds.

---
 module_build_service/scheduler/__init__.py | 20 +++++++-------------
 1 file changed, 7 insertions(+), 13 deletions(-)

diff --git a/module_build_service/scheduler/__init__.py b/module_build_service/scheduler/__init__.py
index fd1903c7..ccde5b70 100644
--- a/module_build_service/scheduler/__init__.py
+++ b/module_build_service/scheduler/__init__.py
@@ -21,20 +21,13 @@ def main(initial_messages, stop_condition):
     config['mbsconsumer.stop_condition'] = stop_condition
     config['mbsconsumer.initial_messages'] = initial_messages
 
-    consumers = [module_build_service.scheduler.consumer.MBSConsumer]
+    # Moksha requires that we subscribe to *something*, so tell it /dev/null
+    # since we'll just be doing in-memory queue-based messaging for this single
+    # build.
+    config['zmq_enabled'] = True
+    config['zmq_subscribe_endpoints'] = 'ipc:///dev/null'
 
-    # Rephrase the fedmsg-config.py config as moksha *.ini format for
-    # zeromq. If we're not using zeromq (say, we're using STOMP), then just
-    # assume that the moksha configuration is specified correctly already
-    # in /etc/fedmsg.d/
-    if config.get('zmq_enabled', True):
-        moksha_options = dict(
-            # XXX - replace this with a /dev/null endpoint.
-            zmq_subscribe_endpoints=','.join(
-                ','.join(bunch) for bunch in config['endpoints'].values()
-            ),
-        )
-        config.update(moksha_options)
+    consumers = [module_build_service.scheduler.consumer.MBSConsumer]
 
     # Note that the hub we kick off here cannot send any message.  You
     # should use fedmsg.publish(...) still for that.
@@ -47,6 +40,7 @@ def main(initial_messages, stop_condition):
         framework=False,
     )
 
+
 def make_simple_stop_condition(session):
     """ Return a simple stop_condition callable.
 

From a45232b73450cbec6cef18aa2c06494af56c42bd Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:26:19 -0500
Subject: [PATCH 22/29] We require moksha.hub now.

---
 requirements.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/requirements.txt b/requirements.txt
index 5ad0a69c..e682b631 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,6 +3,7 @@ Flask-Migrate
 Flask-SQLAlchemy
 Flask-Script
 fedmsg
+moksha.hub
 funcsigs # Python2 only
 httplib2
 kobo

From c9d1a000be5e8d59c11a1c115cac334bb9019023 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:30:27 -0500
Subject: [PATCH 23/29] Also psutil.

---
 requirements.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/requirements.txt b/requirements.txt
index e682b631..0c5c09d1 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,6 +4,7 @@ Flask-SQLAlchemy
 Flask-Script
 fedmsg
 moksha.hub
+psutil
 funcsigs # Python2 only
 httplib2
 kobo

From 908f4403570c66aaabed3c4a26743ecd5adf1095 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:50:43 -0500
Subject: [PATCH 24/29] Make BaseMessage instances appear a bit more like JSON
 so they can make it through moksha validation.

---
 module_build_service/messaging.py | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index 9f991ec1..307bd307 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -43,6 +43,10 @@ class BaseMessage(object):
         """
         self.msg_id = msg_id
 
+        # Unused; just placeholder attributes to appear more like JSON.
+        self.body = {}
+        self.topic = None
+
     def __repr__(self):
         init_sig = signature(self.__init__)
 
@@ -54,6 +58,15 @@ class BaseMessage(object):
 
         return "{}({})".format(type(self).__name__, ', '.join(args_strs))
 
+    def __getitem__(self, key):
+        return getattr(self, key)
+
+    def __setitem__(self, key, value):
+        return setattr(self, key, value)
+
+    def get(self, key, value=None):
+        return getattr(self, key, value)
+
     @staticmethod
     def from_amq(topic, msg):
         msg_obj = None

From 5c3d46595a8d165c2c821fbf2d142a7eb9f68020 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 13:54:34 -0500
Subject: [PATCH 25/29] Use the same key for this that we used to use.

---
 conf/config.py                    | 2 +-
 module_build_service/config.py    | 2 +-
 module_build_service/messaging.py | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/conf/config.py b/conf/config.py
index 825dce34..30f4b161 100644
--- a/conf/config.py
+++ b/conf/config.py
@@ -134,7 +134,7 @@ class TestConfiguration(BaseConfiguration):
     SQLALCHEMY_DATABASE_URI = 'sqlite:///{0}'.format(
         path.join(dbdir, 'tests', 'test_module_build_service.db'))
     DEBUG = True
-    MESSAGING = 'memory'
+    MESSAGING = 'in_memory'
 
     # Global network-related values, in seconds
     NET_TIMEOUT = 3
diff --git a/module_build_service/config.py b/module_build_service/config.py
index aca60a8d..35bbef6e 100644
--- a/module_build_service/config.py
+++ b/module_build_service/config.py
@@ -344,7 +344,7 @@ class Config(object):
 
     def _setifok_messaging(self, s):
         s = str(s)
-        if s not in ("fedmsg", "amq", "memory"):
+        if s not in ("fedmsg", "amq", "in_memory"):
             raise ValueError("Unsupported messaging system.")
         self.messaging = s
 
diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index 307bd307..e450dbb2 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -344,7 +344,7 @@ _messaging_backends = {
     'amq': {
         'publish': _amq_publish,
     },
-    'memory': {
+    'in_memory': {
         'publish': _in_memory_publish,
     }
 }

From f2a02027a9af7d0556969e5b53036ce278ff46f3 Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Thu, 15 Dec 2016 15:11:23 -0500
Subject: [PATCH 26/29] Fix Vagrant

---
 Vagrantfile                                     |  2 ++
 client_secrets.json => conf/client_secrets.json |  0
 conf/config.py                                  |  2 +-
 conf/mbs-scheduler.service                      | 15 ---------------
 docs/CONTRIBUTING.rst                           |  4 ++--
 5 files changed, 5 insertions(+), 18 deletions(-)
 rename client_secrets.json => conf/client_secrets.json (100%)
 delete mode 100644 conf/mbs-scheduler.service

diff --git a/Vagrantfile b/Vagrantfile
index 769d5da5..1c461a41 100644
--- a/Vagrantfile
+++ b/Vagrantfile
@@ -28,6 +28,8 @@ $script = <<SCRIPT
         rpm-build \
         swig \
         systemd-devel
+    mkdir /usr/share/fedmsg
+    chown fedmsg:fedmsg /usr/share/fedmsg
     systemctl enable fedmsg-relay
     systemctl start fedmsg-relay
     mkdir /etc/module-build-service/
diff --git a/client_secrets.json b/conf/client_secrets.json
similarity index 100%
rename from client_secrets.json
rename to conf/client_secrets.json
diff --git a/conf/config.py b/conf/config.py
index 30f4b161..f77076c5 100644
--- a/conf/config.py
+++ b/conf/config.py
@@ -119,7 +119,7 @@ class DevConfiguration(BaseConfiguration):
     KOJI_ARCHES = ['x86_64']
     KOJI_REPOSITORY_URL = 'http://kojipkgs.stg.fedoraproject.org/repos'
 
-    OIDC_CLIENT_SECRETS = "client_secrets.json"
+    OIDC_CLIENT_SECRETS = path.join(confdir, 'client_secrets.json')
 
     SSL_CERTIFICATE_FILE = path.join(confdir, 'server.crt')
     SSL_CERTIFICATE_KEY_FILE = path.join(confdir, 'server.key')
diff --git a/conf/mbs-scheduler.service b/conf/mbs-scheduler.service
deleted file mode 100644
index 1ba57ab6..00000000
--- a/conf/mbs-scheduler.service
+++ /dev/null
@@ -1,15 +0,0 @@
-[Unit]
-Description=Scheduler for the Module Build Service
-After=network.target
-Documentation=https://pagure.io/fm-orchestrator
-
-[Service]
-ExecStart=/usr/bin/mbs-daemon
-Type=simple
-User=fedmsg
-Group=fedmsg
-Restart=on-failure
-
-[Install]
-WantedBy=multi-user.target
-
diff --git a/docs/CONTRIBUTING.rst b/docs/CONTRIBUTING.rst
index 4b10542c..f999bf1c 100644
--- a/docs/CONTRIBUTING.rst
+++ b/docs/CONTRIBUTING.rst
@@ -73,9 +73,9 @@ To start the frontend manually, run the following inside the guest::
 
     $ mbs-frontend
 
-To start the scheduler manually, run the following inside the guest::
+To start the scheduler manually, run the following at `/tmp/module_build_service` inside the guest::
 
-    $ mbs-daemon
+    $ fedmsg-hub
 
 Alternatively, you can restart the Vagrant guest, which inherently starts/restarts the frontend and the scheduler with::
 

From 66ed35d867c4beb30f6d561bfd1b58652e7df946 Mon Sep 17 00:00:00 2001
From: Matt Prahl <mprahl@redhat.com>
Date: Thu, 15 Dec 2016 16:15:04 -0500
Subject: [PATCH 27/29] Modified the spec file for fedmsg-hub

---
 module-build-service.spec | 18 ++++--------------
 1 file changed, 4 insertions(+), 14 deletions(-)

diff --git a/module-build-service.spec b/module-build-service.spec
index 12b6ef57..9608eeb0 100644
--- a/module-build-service.spec
+++ b/module-build-service.spec
@@ -47,6 +47,7 @@ BuildRequires:	systemd
 
 Requires:	systemd
 Requires:	fedmsg
+Requires:	fedmsg-hub
 Requires:	git
 Requires:	kobo
 Requires:	kobo-rpmlib
@@ -98,28 +99,14 @@ tasks:
 %install
 %py2_install
 
-mkdir -p %{buildroot}%{_unitdir}/
-%{__install} -pm644 conf/mbs-scheduler.service \
-    %{buildroot}%{_unitdir}/mbs-scheduler.service
-
 %check
 nosetests-2
 
-%post
-%systemd_post mbs-scheduler.service
-
-%preun
-%systemd_preun mbs-scheduler.service
-
-%postun
-%systemd_postun_with_restart mbs-scheduler.service
-
 %files
 %doc README.rst
 %license LICENSE
 %{python2_sitelib}/module_build_service*
 %{_bindir}/mbs-*
-%{_unitdir}/mbs-scheduler.service
 %dir %{_sysconfdir}/module-build-service
 %config(noreplace) %{_sysconfdir}/module-build-service/config.py
 %config(noreplace) %{_sysconfdir}/module-build-service/koji.conf
@@ -134,6 +121,9 @@ nosetests-2
 
 
 %changelog
+* Thu Dec 15 2016 Matt Prahl <mprahl@redhat.com> - 1.0.3-1
+- Replace systemd unit with fedmsg-hub
+
 * Wed Dec 14 2016 Ralph Bean <rbean@redhat.com> - 1.0.2-1
 - Enable test suite in the check section.
 - Add systemd scriptlets, per review feedback.

From 546e2f5529c3df42d6d9135600ee7d6661a251f2 Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 16:37:48 -0500
Subject: [PATCH 28/29] Remove test_messaging_functions as per review feedback.

---
 .../test_messaging_functions.py               | 42 -------------------
 1 file changed, 42 deletions(-)
 delete mode 100644 tests/test_messaging/test_messaging_functions.py

diff --git a/tests/test_messaging/test_messaging_functions.py b/tests/test_messaging/test_messaging_functions.py
deleted file mode 100644
index 42100954..00000000
--- a/tests/test_messaging/test_messaging_functions.py
+++ /dev/null
@@ -1,42 +0,0 @@
-# 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 Matt Prahl <mprahl@redhat.com>
-from __future__ import unicode_literals
-import unittest
-from datetime import datetime
-
-
-class TestUtilFunctions(unittest.TestCase):
-
-    module_build_service_msg = msg = {
-        'component_builds': [1, 2],
-        'id': 1,
-        'name': 'testmodule',
-        'owner': 'some_user',
-        'release': '16',
-        'scmurl': 'git://domain.local/modules/testmodule.git?#c23acdc',
-        'state': 3,
-        'state_name': 'done',
-        'time_completed': datetime(2016, 9, 1, 2, 30),
-        'time_modified': datetime(2016, 9, 1, 2, 30),
-        'time_submitted': datetime(2016, 9, 1, 2, 28),
-        'version': '4.3.43'
-    }

From e5270d09329a55b65a3459749c090108344935ab Mon Sep 17 00:00:00 2001
From: Ralph Bean <rbean@redhat.com>
Date: Thu, 15 Dec 2016 16:45:32 -0500
Subject: [PATCH 29/29] Some clarifying comments.

---
 module_build_service/messaging.py | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/module_build_service/messaging.py b/module_build_service/messaging.py
index e450dbb2..1b747f93 100644
--- a/module_build_service/messaging.py
+++ b/module_build_service/messaging.py
@@ -43,7 +43,15 @@ class BaseMessage(object):
         """
         self.msg_id = msg_id
 
-        # Unused; just placeholder attributes to appear more like JSON.
+        # Moksha calls `consumer.validate` on messages that it receives, and
+        # even though we have validation turned off in the config there's still
+        # a step that tries to access `msg['body']`, `msg['topic']` and
+        # `msg.get('topic')`.
+        # These are here just so that the `validate` method won't raise an
+        # exception when we push our fake messages through.
+        # Note that, our fake message pushing has worked for a while... but the
+        # *latest* version of fedmsg has some code that exercises the bug.  I
+        # didn't hit this until I went to test in jenkins.
         self.body = {}
         self.topic = None
 
@@ -59,12 +67,15 @@ class BaseMessage(object):
         return "{}({})".format(type(self).__name__, ', '.join(args_strs))
 
     def __getitem__(self, key):
+        """ Used to trick moksha into thinking we are a dict. """
         return getattr(self, key)
 
     def __setitem__(self, key, value):
+        """ Used to trick moksha into thinking we are a dict. """
         return setattr(self, key, value)
 
     def get(self, key, value=None):
+        """ Used to trick moksha into thinking we are a dict. """
         return getattr(self, key, value)
 
     @staticmethod