Renamed amqp.get_consumer_set -> app.get_task_consumer

celery/app/amqp.py

@@ -206,14 +206,13 @@ class AMQP(object):
         # Make sure all queues are declared.
         global _queues_declared
         if not _queues_declared:
-            consumers = self.get_consumer_set(publisher.connection)
-            consumers.close()
+            self.get_task_consumer(publisher.connection).close()
             _queues_declared = True
         publisher.declare()
 
         return publisher
 
-    def get_consumer_set(self, connection, queues=None, **options):
+    def get_task_consumer(self, connection, queues=None, **options):
         queues = queues or self.queues
 
         cset = self.ConsumerSet(connection)

celery/messaging.py

@@ -35,4 +35,4 @@ def get_consumer_set(connection, queues=None, **options):
 
     """
     # FIXME: Deprecate!
-    return default_app.amqp.get_consumer_set(connection, queues, **options)
+    return default_app.amqp.get_task_consumer(connection, queues, **options)

celery/task/control.py

@@ -86,11 +86,11 @@ class Control(object):
         """
 
         def _do_discard(connection=None, connect_timeout=None):
-            consumers = self.app.amqp.get_consumer_set(connection=connection)
+            consumer = self.app.amqp.get_task_consumer(connection=connection)
             try:
-                return consumers.discard_all()
+                return consumer.discard_all()
             finally:
-                consumers.close()
+                consumer.close()
 
         return self.app.with_default_connection(_do_discard)(
                 connection=connection, connect_timeout=connect_timeout)

celery/worker/listener.py

@@ -386,7 +386,7 @@ class CarrotListener(object):
 
         self.connection = self._open_connection()
         self.logger.debug("CarrotListener: Connection Established.")
-        self.task_consumer = self.app.amqp.get_consumer_set(
+        self.task_consumer = self.app.amqp.get_task_consumer(
                                         connection=self.connection,
                                         queues=self.queues)
         # QoS: Reset prefetch window.

docs/internals/app-overview.rst

@@ -77,7 +77,7 @@ Other interesting attributes::
     >>> celery.amqp
     >>> celery.amqp.Router
     >>> celery.amqp.get_queues()
-    >>> celery.amqp.get_consumer_set()
+    >>> celery.amqp.get_task_consumer()
 
     # Loader
     >>> celery.loader
@@ -118,7 +118,7 @@ Aliases (Pending deprecation)
 * celery.messaging
     * .establish_connection -> {app.broker_connection}
     * .with_connection -> {app.with_connection}
-    * .get_consumer_set -> {app.amqp.get_consumer_set}
+    * .get_consumer_set -> {app.amqp.get_task_consumer}
     * .TaskPublisher -> {app.amqp.TaskPublisher}
     * .TaskConsumer -> {app.amqp.TaskConsumer}
     * .ConsumerSet -> {app.amqp.ConsumerSet}