Cosmetic changes

celery/messaging.py

@@ -64,10 +64,9 @@ class TaskPublisher(Publisher):
         return task_id
 
 
-class TaskConsumerSet(ConsumerSet):
-    
-    def __init__(self, connection, queues=conf.AMQP_CONSUMER_QUEUES, consumers=[], **options):
-        super(TaskConsumerSet, self).__init__(connection, queues=queues, consumers=consumers, **options)
+def get_consumer_set(connection, queues=conf.AMQP_CONSUMER_QUEUES):
+    return ConsumerSet(connection, from_dict=queues, decoder=pickle.loads)
+
 
 class TaskConsumer(Consumer):
     """The AMQP Task Consumer class."""
@@ -95,4 +94,4 @@ class StatsConsumer(Consumer):
     no_ack=True
 
     def receive(self, message_data, message):
-        pass
+        pass

celery/worker.py

@@ -1,6 +1,6 @@
 """celery.worker"""
 from carrot.connection import DjangoAMQPConnection
-from celery.messaging import TaskConsumerSet
+from celery.messaging import get_consumer_set
 from celery.conf import DAEMON_CONCURRENCY, DAEMON_LOG_FILE
 from celery.conf import SEND_CELERY_TASK_ERROR_EMAILS
 from celery.log import setup_logger
@@ -326,7 +326,7 @@ class WorkController(object):
 
     .. attribute:: task_consumer
 
-        The :class:`celery.messaging.TaskConsumerSet` instance used.
+        The :class:`carrot.messaging.ConsumerSet` instance used.
 
     """
     loglevel = logging.ERROR
@@ -362,7 +362,7 @@ class WorkController(object):
         """
         self.close_connection()
         self.amqp_connection = DjangoAMQPConnection()
-        self.task_consumer = TaskConsumerSet(connection=self.amqp_connection)
+        self.task_consumer = get_consumer_set(connection=self.amqp_connection)
         self.task_consumer.register_callback(self._message_callback)
         return self.task_consumer
 

testproj/settings.py

@@ -27,8 +27,6 @@ CELERY_AMQP_EXCHANGE = "testcelery"
 CELERY_AMQP_ROUTING_KEY = "testcelery"
 CELERY_AMQP_CONSUMER_QUEUE = "testcelery"
 
-CELERY_TASK_META_USE_DB = True
-
 MANAGERS = ADMINS
 
 DATABASE_ENGINE = 'sqlite3'