Renamed app.amqp.ConsumerSet -> Consumer

Changelog

@@ -95,7 +95,7 @@ Important Notes
     broadcast exchange.
 
     This opens up a lot of possibilities, for example the workers could listen
-    for worker events, to know what workers are in the neighborhood, and even
+    for worker events to know what workers are in the neighborhood, and even
     restart workers when they go down (or use this information to optimize
     tasks/autoscaling).
 
@@ -147,8 +147,8 @@ News
     multiple results at once, unlike `join()` which fetches the results
     one by one.
 
-    So far only supported by the AMQP result backend. Support for memcached
-    and Redis will be added later.
+    So far only supported by the AMQP result backend.  Support for memcached
+    and Redis may be added later.
 
 * `celerybeat`: Now has built-in daemonization support using the `--detach``
    option.

celery/app/amqp.py

@@ -29,7 +29,7 @@ BROKER_FORMAT = """\
 %(transport)s://%(userid)s@%(hostname)s%(port)s%(virtual_host)s\
 """
 
-#: Set to :cosnt:`True` when the configured queues has been declared.
+#: Set to :const:`True` when the configured queues has been declared.
 _queues_declared = False
 
 #: Set of exchange names that has already been declared.
@@ -37,6 +37,8 @@ _exchanges_declared = set()
 
 
 def extract_msg_options(options, keep=MSG_OPTIONS):
+    """Extracts known options to `basic_publish` from a dict,
+    as a new dict."""
     return dict((name, options.get(name)) for name in keep)
 
 
@@ -170,7 +172,7 @@ class AMQP(object):
     def __init__(self, app):
         self.app = app
 
-    def ConsumerSet(self, *args, **kwargs):
+    def Consumer(self, *args, **kwargs):
         return messaging.ConsumerSet(*args, **kwargs)
 
     def Queues(self, queues):
@@ -211,8 +213,8 @@ class AMQP(object):
         return publisher
 
     def get_task_consumer(self, connection, queues=None, **kwargs):
-        return self.ConsumerSet(connection, from_dict=queues or self.queues,
-                                **kwargs)
+        return self.Consumer(connection, from_dict=queues or self.queues,
+                             **kwargs)
 
     def get_default_queue(self):
         q = self.app.conf.CELERY_DEFAULT_QUEUE

celery/messaging.py

@@ -8,7 +8,7 @@ from celery.app import app_or_default
 
 default_app = app_or_default()
 TaskPublisher = default_app.amqp.TaskPublisher
-ConsumerSet = default_app.amqp.ConsumerSet
+ConsumerSet = default_app.amqp.Consumer
 TaskConsumer = default_app.amqp.TaskConsumer