Renamed celery.scheduler -> celery.worker.scheduler

celery/tests/test_worker.py

@@ -9,7 +9,7 @@ from carrot.backends.base import BaseMessage
 from celery import registry
 from celery.serialization import pickle
 from celery.utils import gen_unique_id
-from celery.scheduler import Scheduler
+from celery.worker.scheduler import Scheduler
 from datetime import datetime, timedelta
 
 

celery/worker/__init__.py

@@ -7,6 +7,7 @@ from carrot.connection import DjangoBrokerConnection, AMQPConnectionException
 from celery.worker.controllers import Mediator, ScheduleController
 from celery.beat import ClockServiceThread
 from celery.worker.job import TaskWrapper
+from celery.worker.scheduler import Scheduler
 from celery.exceptions import NotRegistered
 from celery.messaging import get_consumer_set
 from celery.conf import DAEMON_CONCURRENCY, DAEMON_LOG_FILE
@@ -15,7 +16,6 @@ from celery.log import setup_logger
 from celery.pool import TaskPool
 from celery.utils import retry_over_time
 from celery.datastructures import SharedCounter
-from celery.scheduler import Scheduler
 from Queue import Queue
 import traceback
 import logging

celery/scheduler.py → celery/worker/scheduler.py

@@ -48,30 +48,3 @@ class Scheduler(object):
     def queue(self):
         events = list(self._queue)
         return map(heapq.heappop, [events]*len(events))
-
-
-if __name__ == "__main__":
-    from Queue import Queue, Empty
-    from datetime import datetime, timedelta
-
-    bucket = Queue()
-    schedule = Scheduler(bucket)
-
-    # Enter some eta tasks in the scheduler
-    for i in reversed(range(10)):
-        task = "Task %d" % i
-        schedule.enter(task, eta=datetime.now() + timedelta(seconds=i + 5))
-
-    # Run the scheduler
-    for delay in schedule:
-        print("delay->%d" % delay)
-        if schedule.empty():
-            break
-        time.sleep(delay)
-
-    # Dump out the contents of the bucket queue.
-    while True:
-        try:
-            print(bucket.get_nowait())
-        except Empty:
-            break