Add a "run" consumer callback

Runs a configurable command when a message is received.

Fixes fedora-infra#155

Signed-off-by: Jeremy Cline <[email protected]>

docs/consuming.rst

@@ -197,6 +197,15 @@ called when a message arrives::
     api.consume(PrintMessage)
 
 
+Common Consumers
+================
+
+Several generally useful consumers are shipped with fedora-messaging.
+
+.. autofunction:: fedora_messaging.callbacks.printer
+.. autofunction:: fedora_messaging.callbacks.run
+
+
 .. _AMQP overview: https://www.rabbitmq.com/tutorials/amqp-concepts.html
 .. _RabbitMQ tutorials: https://www.rabbitmq.com/getstarted.html
 .. _pika: https://pika.readthedocs.io/

fedora_messaging/callbacks.py

@@ -17,8 +17,15 @@
 """This module contains a set of callbacks that are generally useful."""
 from __future__ import print_function
 
+import logging
+import subprocess
+
 import six
 
+from . import config, exceptions
+
+_log = logging.getLogger(__name__)
+
 
 def printer(message):
     """
@@ -30,3 +37,40 @@ def printer(message):
         message (fedora_messaging.api.Message): The message that was received.
     """
     print(six.text_type(message))
+
+
+def run(message):
+    """
+    Run a command in a subprocess.
+
+    **Configuration Options**
+
+    This consumer uses the following configuration keys in the :ref:`conf-consumer-config`
+    section of the configuration file:
+
+    * ``command`` (string or list of strings): The arguments to launch the process
+        as a string or list of strings.
+    * ``shell`` (boolean): Whether or not to execute the command in a shell. Refer to the
+        `security considerations
+        <https://docs.python.org/3/library/subprocess.html#security-considerations>`_
+        before setting this to true. Defaults to false.
+    * ``timeout`` (integer): The length of time, in seconds, to wait on the command
+        before halting the command and returning the message to the queue for later
+        processing. Defaults to no timeout.
+    """
+    try:
+        subprocess.check_call(
+            config.conf["consumer_config"]["command"],
+            shell=config.conf["consumer_config"].get("shell", False),
+            timeout=config.conf["consumer_config"].get("timeout"),
+        )
+    except subprocess.CalledProcessError:
+        raise exceptions.Drop()
+    except subprocess.TimeoutExpired as e:
+        _log.error(
+            "{} failed to return after {} seconds; returning {} to the queue",
+            e.cmd,
+            e.timeout,
+            str(message.id),
+        )
+        raise exceptions.Nack()