Skip to main content

Rejected is a Python RabbitMQ Consumer Framework and Controller Daemon

Project description

Rejected is a AMQP consumer daemon and message processing framework. It allows for rapid development of message processing consumers by handling all of the core functionality of communicating with RabbitMQ and management of consumer processes.

Rejected runs as a master process with multiple consumer configurations that are each run it an isolated process. It has the ability to collect statistical data from the consumer processes and report on it.

Version Downloads Status License

Features

  • Dynamic QoS

  • Automatic exception handling including connection management and consumer restarting

  • Smart consumer classes that can automatically decode and deserialize message bodies based upon message headers

  • Metrics logging and submission to statsd

  • Built-in profiling of consumer code

  • Ability to write asynchronous code in consumers allowing for parallel communication with external resources

Documentation

https://rejected.readthedocs.org

Example Consumers

from rejected import consumer
import logging

LOGGER = logging.getLogger(__name__)


class Test(consumer.Consumer):
    def process(self, message):
        LOGGER.debug('In Test.process: %s' % message.body)

Async Consumer

To make a consumer async, you can decorate the Consumer.prepare and Consumer.process methods using Tornado’s @gen.coroutine. Asynchronous consumers to handle multiple messages in the same process, but rather allow you to use asynchronous clients like Tornado’s AsyncHTTPClient and the Queries PostgreSQL library to perform parallel tasks using coroutines.

import logging

from rejected import consumer

from tornado import gen
from tornado import httpclient


class AsyncExampleConsumer(consumer.Consumer):

    @gen.coroutine
    def process(self):
        LOGGER.debug('Message: %r', self.body)
        http_client = httpclient.AsyncHTTPClient()
        results = yield [http_client.fetch('http://www.github.com'),
                         http_client.fetch('http://www.reddit.com')]
        LOGGER.info('Length: %r', [len(r.body) for r in results])

Example Configuration

%YAML 1.2
---
Application:
  poll_interval: 10.0
  log_stats: True
  statsd:
    enabled: True
    host: localhost
    port: 8125
    prefix: applications.rejected
  Connections:
    rabbitmq:
      host: localhost
      port: 5672
      user: guest
      pass: guest
      ssl: False
      vhost: /
      heartbeat_interval: 300
  Consumers:
    example:
      consumer: rejected.example.Consumer
      connections: [rabbitmq]
      qty: 2
      queue: generated_messages
      dynamic_qos: True
      ack: True
      max_errors: 100
      config:
        foo: True
        bar: baz

 Daemon:
   user: rejected
   group: daemon
   pidfile: /var/run/rejected/example.%(pid)s.pid

 Logging:
   version: 1
   formatters:
     verbose:
       format: "%(levelname) -10s %(asctime)s %(process)-6d %(processName) -15s %(name) -25s %(funcName) -20s: %(message)s"
       datefmt: "%Y-%m-%d %H:%M:%S"
     syslog:
       format: "%(levelname)s <PID %(process)d:%(processName)s> %(name)s.%(funcName)s(): %(message)s"
   filters: []
   handlers:
     console:
       class: logging.StreamHandler
       formatter: verbose
       debug_only: true
     syslog:
       class: logging.handlers.SysLogHandler
       facility: local6
       address: /var/run/syslog
       #address: /dev/log
       formatter: syslog
   loggers:
     my_consumer:
       level: INFO
       propagate: true
       handlers: [console, syslog]
     rejected:
       level: INFO
       propagate: true
       handlers: [console, syslog]
     urllib3:
       level: ERROR
       propagate: true
   disable_existing_loggers: false
   incremental: false

Version History

Available at https://rejected.readthedocs.org/en/latest/history.html

Project details


Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

rejected-3.6.1.tar.gz (27.2 kB view hashes)

Uploaded Source

Built Distribution

rejected-3.6.1-py2-none-any.whl (34.8 kB view hashes)

Uploaded Python 2

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page