diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py index 473cff92e..1aebed2c3 100644 --- a/scrapy/core/scheduler.py +++ b/scrapy/core/scheduler.py @@ -152,7 +152,11 @@ class Scheduler(object): def _mq(self): """ Create a new priority queue instance, with in-memory storage """ - return create_instance(self.pqclass, None, self.crawler, self._newmq) + return create_instance(self.pqclass, + None, + self.crawler, + self.mqclass, + '') def _dq(self): """ Create a new priority queue instance, with disk storage """ diff --git a/scrapy/pqueues.py b/scrapy/pqueues.py index 454609470..9bebe3b7b 100644 --- a/scrapy/pqueues.py +++ b/scrapy/pqueues.py @@ -90,8 +90,73 @@ class _SlotPriorityQueues(object): return slot in self.pqueues -class ScrapyPriorityQueue(PriorityQueue): - pass +class ScrapyPriorityQueue(object): + + @classmethod + def from_crawler(cls, crawler, downstream_queue_cls, key, startprios=()): + return cls(crawler, downstream_queue_cls, key, startprios) + + def __init__(self, crawler, downstream_queue_cls, key, startprios=()): + self.crawler = crawler + self.downstream_queue_cls = downstream_queue_cls + self.key = key + self.queues = {} + self.curprio = None + self.read_prios(startprios) + + def read_prios(self, startprios): + if not startprios: + return + + if not isinstance(startprios, dict): + raise ValueError("ScrapyPriorityQueue accepts " + "``startprios`` as a dict; %r instance " + "is passed. Most likely, it means the state is" + "created by an incompatible priority queue. " + "Only a crawl started with the same priority " + "queue class can be resumed." % + slot_startprios.__class__) + + for priority, state in startprios.items(): + self.queues[priority] = self.qfactory(priority, state) + + self.curprio = min(startprios) + + def qfactory(self, key, startprios=()): + return self.downstream_queue_cls(self.crawler, + self.key + '/' + str(key), + startprios) + + def push(self, obj, priority=0): + if priority not in self.queues: + self.queues[priority] = self.qfactory(priority) + q = self.queues[priority] + q.push(obj) # this may fail (eg. serialization error) + if self.curprio is None or priority < self.curprio: + self.curprio = priority + + def pop(self): + if self.curprio is None: + return + q = self.queues[self.curprio] + m = q.pop() + if len(q) == 0: + del self.queues[self.curprio] + q.close() + prios = [p for p, q in self.queues.items() if len(q) > 0] + self.curprio = min(prios) if prios else None + return m + + def close(self): + active = {} + for p, q in self.queues.items(): + if len(q): + active.append(p) + q.close() + return active + + def __len__(self): + return sum(len(x) for x in self.queues.values()) if self.queues else 0 class DownloaderInterface(object):