osc-origin: update: provide --listen option.

Listens for source changes (both request creation and package updates) to
issue updates based on origin configuration.

osc-origin.py

@@ -35,6 +35,8 @@ OSRT_ORIGIN_LOOKUP_TTL = 60 * 60 * 24 * 7
 @cmdln.option('--dry', action='store_true', help='perform a dry-run where applicable')
 @cmdln.option('--force-refresh', action='store_true', help='force refresh of data')
 @cmdln.option('--format', default='plain', help='output format')
+@cmdln.option('--listen', action='store_true', help='listen to events')
+@cmdln.option('--listen-seconds', help='number of seconds to listen to events')
 @cmdln.option('--mail', action='store_true', help='mail report to <confg:mail-release-list>')
 @cmdln.option('--origins-only', action='store_true', help='list origins instead of expanded config')
 @cmdln.option('-p', '--project', help='project on which to operate')
@@ -62,7 +64,7 @@ def do_origin(self, subcmd, opts, *args):
         osc origin potentials [--format json|yaml] PACKAGE
         osc origin projects [--format json|yaml]
         osc origin report [--diff] [--force-refresh] [--mail]
-        osc origin update [PACKAGE...]
+        osc origin update [--listen] [--listen-seconds] [PACKAGE...]
     """
 
     if len(args) == 0:
@@ -344,6 +346,21 @@ def osrt_origin_report(apiurl, opts, *args):
                   body, None, dry=opts.dry)
 
 def osrt_origin_update(apiurl, opts, *packages):
+    if opts.listen:
+        import logging
+        from osclib.origin_listener import OriginSourceChangeListener
+
+        logger = logging.getLogger()
+        logger.setLevel(logging.INFO)
+        listener = OriginSourceChangeListener(apiurl, logger, opts.project, opts.dry)
+        try:
+            runtime = int(opts.listen_seconds) if opts.listen_seconds else None
+            listener.run(runtime=runtime)
+        except KeyboardInterrupt:
+            listener.stop()
+
+        return
+
     if not opts.project:
         for project in origin_updatable(apiurl):
             opts.project = project

osclib/origin_listener.py

@@ -0,0 +1,88 @@
+import json
+from osclib.core import package_kind
+from osclib.origin import origin_updatable_map
+from osclib.origin import origin_update
+from osclib.PubSubConsumer import PubSubConsumer
+
+
+class OriginSourceChangeListener(PubSubConsumer):
+    def __init__(self, apiurl, logger, project=None, dry=False):
+        self.apiurl = apiurl
+        self.project = project
+        self.dry = dry
+
+        amqp_prefix = 'suse' if self.apiurl.endswith('suse.de') else 'opensuse'
+        super().__init__(amqp_prefix, logger)
+
+    def routing_keys(self):
+        return [self._prefix + k for k in [
+            '.obs.package.update',
+            '.obs.request.create',
+        ]]
+
+    def on_message(self, unused_channel, method, properties, body):
+        super().on_message(unused_channel, method, properties, body)
+
+        payload = json.loads(body)
+        if method.routing_key == '{}.obs.package.update'.format(self._prefix):
+            self.on_message_package_update(payload)
+        elif method.routing_key == '{}.obs.request.create'.format(self._prefix):
+            self.on_message_request_create(payload)
+        else:
+            raise Exception('Unrequested message: {}'.format(method.routing_key))
+
+    def on_message_package_update(self, payload):
+        origins = origin_updatable_map(self.apiurl)
+        self.update_consider(origins, payload['project'], payload['package'])
+
+    def on_message_request_create(self, payload):
+        origins = origin_updatable_map(self.apiurl, pending=True)
+        for action in payload['actions']:
+            # The following code demonstrates the quality of the data structure.
+            # The base structure is inconsistent enough and yet the event data
+            # structure manages to be different from XML structure (for no
+            # reason) and even more inconsistent at that.
+            if action['type'] == 'delete':
+                if not action.get('targetpackage'):
+                    continue
+
+                project = action['targetproject']
+                package = action['targetpackage']
+            elif action['type'] == 'maintenance_incident':
+                project = action['target_releaseproject']
+                if not action.get('targetpackage'):
+                    package = action['sourcepackage']
+                else:
+                    repository_suffix_length = len(project) + 1 # package.project
+                    package = action['targetpackage'][:-repository_suffix_length]
+            elif action['type'] == 'maintenance_release':
+                project = action['targetproject']
+                repository_suffix_length = len(project) + 1 # package.project
+                package = action['sourcepackage'][:-repository_suffix_length]
+            elif action['type'] == 'submit':
+                project = action['targetproject']
+                package = action['targetpackage']
+            else:
+                # Unsupported action type.
+                continue
+
+            self.update_consider(origins, project, package)
+
+    def update_consider(self, origins, origin_project, package):
+        if origin_project not in origins:
+            self.logger.info('skipped irrelevant project: {}'.format(origin_project))
+            return
+
+        for project in origins[origin_project]:
+            if self.project and project != self.project:
+                self.logger.info('skipping filtered target project: {}'.format(project))
+                continue
+
+            kind = package_kind(self.apiurl, project, package)
+            if kind == 'source':
+                request_future = origin_update(self.apiurl, project, package)
+                if request_future:
+                    request_future.print_and_create(self.dry)
+            else:
+                # This eliminates the possibility for deletes by listener.
+                self.logger.info('skipped updating non-existant package {}/{}'.format(project, package))