scrapy源码1：engine的源码分析

旧城等待， 2023-02-21 11:49 11阅读 0赞

##### scrapy中engine.py的源码分析如下： #####

""" engine.py提供了2个类：Slot和ExecutionEngine Slot: 提供了几个方法添加请求，删除请求，关闭自己，触发关闭方法 它使用Twisted的主循环reactor来不断的调度执行Engine的"_next_request"方法，这个方法也是核心循环方法。 ExecutionEngine: 引擎的执行任务 爬虫引擎是控制调度器，下载器和爬虫的。 This is the Scrapy engine which controls the Scheduler, Downloader and Spiders. For more information see docs/topics/architecture.rst """
    import logging
    from time import time
    
    # Twisted是用Python实现的基于事件驱动的网络引擎框架，这里引用了它，可能用于网络方面
    from twisted.internet import defer, task
    from twisted.python.failure import Failure
    
    # 接下来导入了一些自己的包
    from scrapy import signals  # 这些信号记录在 docs/topics/signals.rst. 请不要在没有记录的情况下在此处添加新信号。
    '''signals源码如下： """ Scrapy signals These signals are documented in docs/topics/signals.rst. Please don't add new signals here without documenting them there. """ engine_started = object() engine_stopped = object() spider_opened = object() spider_idle = object() # 爬虫空闲 spider_closed = object() spider_error = object() request_scheduled = object() request_dropped = object() request_reached_downloader = object() response_received = object() response_downloaded = object() item_scraped = object() item_dropped = object() item_error = object() # for backwards compatibility stats_spider_opened = spider_opened stats_spider_closing = spider_closed stats_spider_closed = spider_closed item_passed = item_scraped request_received = request_scheduled '''
    from scrapy.core.scraper import Scraper  # scraper:数据抓取器，主要用于从网页中抓取数据的处理。也就是ItemPipeLine的处理。
    from scrapy.exceptions import DontCloseSpider  # 从自定义的类里面导入了一个异常类
    from scrapy.http import Response, Request  # 从http里面导入了response,request
    from scrapy.utils.misc import load_object  # 加载给定对象绝对路径的对象，然后返回它;这个方法是传递一个str，返回一个类，方法，变量或者实例
    '''load_object源码如下： def load_object(path): """Load an object given its absolute object path, and return it. object can be a class, function, variable or an instance. path ie: 'scrapy.downloadermiddlewares.redirect.RedirectMiddleware' """ try: dot = path.rindex('.') except ValueError: raise ValueError("Error loading object '%s': not a full path" % path) module, name = path[:dot], path[dot+1:] mod = import_module(module) try: obj = getattr(mod, name) except AttributeError: raise NameError("Module '%s' doesn't define any object named '%s'" % (module, name)) return obj '''
    from scrapy.utils.reactor import CallLaterOnce  # 调度要在下一个reactor循环中调用的函数，但前提是该函数自上次运行以来尚未被调度
    '''CallLaterOnce源码如下： class CallLaterOnce(object): """Schedule a function to be called in the next reactor loop, but only if it hasn't been already scheduled since the last time it ran. """ def __init__(self, func, *a, **kw): self._func = func self._a = a self._kw = kw self._call = None def schedule(self, delay=0): if self._call is None: self._call = reactor.callLater(delay, self) def cancel(self): if self._call: self._call.cancel() def __call__(self): self._call = None return self._func(*self._a, **self._kw) '''
    from scrapy.utils.log import logformatter_adapter, failure_to_exc_info  # 日志相关
    
    logger = logging.getLogger(__name__)  # 全局的日志对象
    
    # slot代表一次nextcall的执行，实际上就是执行一次engine的_next_request。
    # slot创建了一个hearbeat，即为一个心跳。通过twisted的task.LoopingCall实现。
    # 每隔5s执行一次，尝试处理一个新的request，这属于被动执行。后面还会有主动执行的代码。
    # slot可以理解为一个request的生命周期。
    class Slot(object):
    
        def __init__(self, start_requests, close_if_idle, nextcall, scheduler):
            self.closing = False
            self.inprogress = set()  # requests in progress
            self.start_requests = iter(start_requests)
            self.close_if_idle = close_if_idle
            self.nextcall = nextcall
            self.scheduler = scheduler
            self.heartbeat = task.LoopingCall(nextcall.schedule)  # heartbeat属性在init的时候初始化了，在close的时候调用stop。
    
        def add_request(self, request):
            self.inprogress.add(request)
    
        def remove_request(self, request):
            self.inprogress.remove(request)
            self._maybe_fire_closing()
    
        def close(self):
            self.closing = defer.Deferred()
            self._maybe_fire_closing()
            return self.closing
    
        def _maybe_fire_closing(self):
            if self.closing and not self.inprogress:
                if self.nextcall:
                    self.nextcall.cancel()
                    if self.heartbeat.running:
                        self.heartbeat.stop()
                self.closing.callback(None)
    
    
    class ExecutionEngine(object):
        # 接受crawler爬虫，spider_close_callback 完成初始化工作
        # 接受初始化的几个参数，设置、信号、日志格式、从crawler那里获取到，从设置中加载日志调度类，从设置加载下载类
        # 其中的设置scheduler_cls, downloader_cls, 默认值可以从default_settings.py获取
        # SCHEDULER = 'scrapy.core.scheduler.Scheduler'
        # DOWNLOADER = 'scrapy.core.downloader.Downloader'
        def __init__(self, crawler, spider_closed_callback):
            self.crawler = crawler
            self.settings = crawler.settings
            self.signals = crawler.signals
            self.logformatter = crawler.logformatter
            self.slot = None
            self.spider = None
            self.running = False
            self.paused = False
            self.scheduler_cls = load_object(self.settings['SCHEDULER'])
            downloader_cls = load_object(self.settings['DOWNLOADER'])
            self.downloader = downloader_cls(crawler)
            self.scraper = Scraper(crawler)
            self._spider_closed_callback = spider_closed_callback
    
        # start: 启动爬虫引擎，方法上面带了个装饰器 @defer.inlineCallbacks
        # 先去断言引擎的运行状态，记录下开始时间，发送一个引擎启动的信号，设置运行状态为运行，设置_closewait为延迟对象，返回_closewait
        # 或者说：记录启动时间；发送一个"engine_started"消息；设置running标志；创建一个_closewait的Deferred对象并返回。
        @defer.inlineCallbacks
        def start(self):
            """Start the execution engine"""
            assert not self.running, "Engine already running"
            self.start_time = time()
            yield self.signals.send_catch_log_deferred(signal=signals.engine_started)
            self.running = True
            # 这个_closewait会返回给CrawlerProcess，
            # 这个Deferred在引擎结束时才会调用，因此用它来向CrawlerProcess通知一个Crawler已经爬取完毕。
            self._closewait = defer.Deferred()
            yield self._closewait
    
        # stop : 优雅地停止执行引擎
        # 标记状态running为false, 关闭所有的爬虫, 调用_finish_stopping_engine
        def stop(self):
            """Stop the execution engine gracefully"""
            assert self.running, "Engine not running"
            self.running = False
            dfd = self._close_all_spiders()
            return dfd.addBoth(lambda _: self._finish_stopping_engine())
    
        # close: 优雅的关闭执行引擎
        # 调用stop方法，完成引擎的关闭工作，其他情况下，关闭爬虫和下载器
        def close(self):
            """Close the execution engine gracefully. If it has already been started, stop it. In all cases, close all spiders and the downloader. """
            if self.running:
                # Will also close spiders and downloader
                return self.stop()
            elif self.open_spiders:
                # Will also close downloader
                return self._close_all_spiders()
            else:
                return defer.succeed(self.downloader.close())
    
        # pause:暂停执行引擎
        def pause(self):
            """Pause the execution engine"""
            self.paused = True
    
        # unpause:解除引擎的暂停
        def unpause(self):
            """Resume the execution engine"""
            self.paused = False
    
        # _next_request:下次请求
        def _next_request(self, spider):
            slot = self.slot
            if not slot:  # 状态判断
                return
    
            if self.paused:  # 状态判断
                return
    
            while not self._needs_backout(spider):
                if not self._next_request_from_scheduler(spider):
                    break
    
            # 下面if语句，请求不为空，并且爬虫没有处理完毕
            if slot.start_requests and not self._needs_backout(spider):
                try:
                    request = next(slot.start_requests)  # 调用next方法
                except StopIteration:
                    slot.start_requests = None
                except Exception:
                    slot.start_requests = None
                    logger.error('Error while obtaining start requests',
                                 exc_info=True, extra={ 'spider': spider})
                else:
                    self.crawl(request, spider)  # 否则去调用crawl方法。
    
            # 如果爬虫是空闲的，并且爬虫空闲则关闭 是true的话，调用_spider_idle方法。
            if self.spider_is_idle(spider) and slot.close_if_idle:
                self._spider_idle(spider)
    
        # _needs_backout: 返回一个布尔值
        # 如果引擎关闭则返回true, 或者slot关闭，或者下载器那里返回了true, 或者爬虫那里返回true,
        # 后面的那2个needs_backout需要具体到downloader, scrper类里面去看。
        # 我们可以对这个方法的理解为没有接下来的工作了就返回true
        def _needs_backout(self, spider):
            slot = self.slot
            return not self.running \
                or slot.closing \
                or self.downloader.needs_backout() \
                or self.scraper.slot.needs_backout()
    
        # _next_request_from_scheduler: 从调度器获取下一个请求， 判断request,下载请求
        def _next_request_from_scheduler(self, spider):
            slot = self.slot
            request = slot.scheduler.next_request()
            if not request:
                return
            d = self._download(request, spider)
            d.addBoth(self._handle_downloader_output, request, spider)
            d.addErrback(lambda f: logger.info('Error while handling downloader output',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            d.addBoth(lambda _: slot.remove_request(request))
            d.addErrback(lambda f: logger.info('Error while removing request from slot',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            d.addBoth(lambda _: slot.nextcall.schedule())
            d.addErrback(lambda f: logger.info('Error while scheduling new request',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            return d
    
        # _handle_downloader_output : 处理下载的输出
        # 断言response为request,response,failure ,如果是request则调用crawl方法，如果是响应enqueue_scrape处理。
        def _handle_downloader_output(self, response, request, spider):
            assert isinstance(response, (Request, Response, Failure)), response
            # downloader middleware can return requests (for example, redirects)
            if isinstance(response, Request):
                self.crawl(response, spider)
                return
            # response is a Response or Failure
            d = self.scraper.enqueue_scrape(response, request, spider)
            d.addErrback(lambda f: logger.error('Error while enqueuing downloader output',
                                                exc_info=failure_to_exc_info(f),
                                                extra={ 'spider': spider}))
            return d
    
        # spider_is_idle: 判定爬虫是否是空闲的
        # 判定slot空闲，判定下载空闲，判定请求为空，判定调度器没有要处理的请求
        def spider_is_idle(self, spider):
            if not self.scraper.slot.is_idle():
                # scraper is not idle
                return False
    
            if self.downloader.active:
                # downloader has pending requests
                return False
    
            if self.slot.start_requests is not None:
                # not all start requests are handled
                return False
    
            if self.slot.scheduler.has_pending_requests():
                # scheduler has pending requests
                return False
    
            return True
    
        # open_spiders: 打开爬虫
        @property
        def open_spiders(self):
            return [self.spider] if self.spider else []
    
        # has_capacity: 判断是否有能力处理更多的爬虫引擎
        def has_capacity(self):
            """Does the engine have capacity to handle more spiders"""
            return not bool(self.slot)  # 初始化时规定了self.slot = None
    
        # crawl : 爬取，先断言爬虫是打开的，执行调度，执行回调的调度
        def crawl(self, request, spider):
            assert spider in self.open_spiders, \
                "Spider %r not opened when crawling: %s" % (spider.name, request)
            self.schedule(request, spider)
            self.slot.nextcall.schedule()
    
        # schedule : 调度，发出请求调度事件，如果enqueue_request , 则触发请求丢弃事件。
        def schedule(self, request, spider):
            self.signals.send_catch_log(signal=signals.request_scheduled,
                    request=request, spider=spider)
            if not self.slot.scheduler.enqueue_request(request):
                self.signals.send_catch_log(signal=signals.request_dropped,
                                            request=request, spider=spider)
    
        # download : 下载, 调用内部方法_download
        def download(self, request, spider):
            d = self._download(request, spider)
            d.addBoth(self._downloaded, self.slot, request, spider)
            return d
    
        # _download : 内部方法
        def _downloaded(self, response, slot, request, spider):
            slot.remove_request(request)
            return self.download(response, spider) \
                    if isinstance(response, Request) else response
    
        # 添加请求， 定义一个成功的方法，一个完成的方法，从下载器里面提取对象, getaway添加成功回调，添加完成。
        # addCallbacks,addBoth 这2个方法用的挺多的
        # addcallbacks 接受一个成功的回调方法， 一个失败的回调方法，
        # addBoth函数向callback与errback链中添加了相同的回调函数
        def _download(self, request, spider):
            slot = self.slot
            slot.add_request(request)
            def _on_success(response):
                assert isinstance(response, (Response, Request))
                if isinstance(response, Response):
                    response.request = request  # tie request to response received
                    logkws = self.logformatter.crawled(request, response, spider)
                    logger.log(*logformatter_adapter(logkws), extra={ 'spider': spider})
                    self.signals.send_catch_log(signal=signals.response_received, \
                        response=response, request=request, spider=spider)
                return response
    
            def _on_complete(_):
                slot.nextcall.schedule()
                return _
    
            dwld = self.downloader.fetch(request, spider)
            dwld.addCallbacks(_on_success)
            dwld.addBoth(_on_complete)
            return dwld
    
        # open_spider：打开爬虫
        # 先断言容量，记录info日志，获取nextcall;
        # 通过crawler构造scheduler调度器，构造slot对象，调度器打开爬虫，爬虫打开，出发爬虫打开事件，启动心跳信息。
        @defer.inlineCallbacks
        def open_spider(self, spider, start_requests=(), close_if_idle=True):
            assert self.has_capacity(), "No free spider slot when opening %r" % \
                spider.name
            logger.info("Spider opened", extra={ 'spider': spider})
            nextcall = CallLaterOnce(self._next_request, spider)
            scheduler = self.scheduler_cls.from_crawler(self.crawler)
            start_requests = yield self.scraper.spidermw.process_start_requests(start_requests, spider)
            slot = Slot(start_requests, close_if_idle, nextcall, scheduler)
            self.slot = slot
            self.spider = spider
            yield scheduler.open(spider)
            yield self.scraper.open_spider(spider)
            self.crawler.stats.open_spider(spider)
            yield self.signals.send_catch_log_deferred(signals.spider_opened, spider=spider)
            slot.nextcall.schedule()
            slot.heartbeat.start(5)
    
        # _spider_idle: 爬虫空闲， 判定空闲， 如果空闲的话，关闭爬虫
        def _spider_idle(self, spider):
            """Called when a spider gets idle. This function is called when there are no remaining pages to download or schedule. It can be called multiple times. If some extension raises a DontCloseSpider exception (in the spider_idle signal handler) the spider is not closed until the next loop and this function is guaranteed to be called (at least) once again for this spider. 翻译：当爬虫空闲时调用。在没有剩余的页面可供下载或调度时，调用此函数。可以称之为 多次。如果某个扩展引发DontCloseSpider异常（在spider_idle信号处理器中）直到 下一个循环这个爬虫才关闭，这个函数保证爬虫被调用（至少）一次。 """
            res = self.signals.send_catch_log(signal=signals.spider_idle, \
                spider=spider, dont_log=DontCloseSpider)
            if any(isinstance(x, Failure) and isinstance(x.value, DontCloseSpider) \
                    for _, x in res):
                return
    
            if self.spider_is_idle(spider):
                self.close_spider(spider, reason='finished')
    
        # close_spider : 关闭爬虫，绑定各种错误回调。
        # 关闭（取消）spider并清除所有未完成的请求
        def close_spider(self, spider, reason='cancelled'):
            """Close (cancel) spider and clear all its outstanding requests"""
    
            slot = self.slot
            if slot.closing:
                return slot.closing
            logger.info("Closing spider (%(reason)s)",
                        { 'reason': reason},
                        extra={ 'spider': spider})
    
            dfd = slot.close()
    
            def log_failure(msg):
                def errback(failure):
                    logger.error(
                        msg,
                        exc_info=failure_to_exc_info(failure),
                        extra={ 'spider': spider}
                    )
                return errback
    
            dfd.addBoth(lambda _: self.downloader.close())
            dfd.addErrback(log_failure('Downloader close failure'))
    
            dfd.addBoth(lambda _: self.scraper.close_spider(spider))
            dfd.addErrback(log_failure('Scraper close failure'))
    
            dfd.addBoth(lambda _: slot.scheduler.close(reason))
            dfd.addErrback(log_failure('Scheduler close failure'))
    
            dfd.addBoth(lambda _: self.signals.send_catch_log_deferred(
                signal=signals.spider_closed, spider=spider, reason=reason))
            dfd.addErrback(log_failure('Error while sending spider_close signal'))
    
            dfd.addBoth(lambda _: self.crawler.stats.close_spider(spider, reason=reason))
            dfd.addErrback(log_failure('Stats close failure'))
    
            dfd.addBoth(lambda _: logger.info("Spider closed (%(reason)s)",
                                              { 'reason': reason},
                                              extra={ 'spider': spider}))
    
            dfd.addBoth(lambda _: setattr(self, 'slot', None))
            dfd.addErrback(log_failure('Error while unassigning slot'))
    
            dfd.addBoth(lambda _: setattr(self, 'spider', None))
            dfd.addErrback(log_failure('Error while unassigning spider'))
    
            dfd.addBoth(lambda _: self._spider_closed_callback(spider))
    
            return dfd
    
        # _close_all_spiders :关闭所有的爬虫，遍历爬虫，执行关闭操作
        def _close_all_spiders(self):
            dfds = [self.close_spider(s, reason='shutdown') for s in self.open_spiders]
            dlist = defer.DeferredList(dfds)
            return dlist
    
        # _finish_stopping_engine : 结束引擎，触发引擎关闭操作。
        @defer.inlineCallbacks
        def _finish_stopping_engine(self):
            yield self.signals.send_catch_log_deferred(signal=signals.engine_stopped)
            self._closewait.callback(None)

##### 实际源码中engine.py #####

""" This is the Scrapy engine which controls the Scheduler, Downloader and Spiders. For more information see docs/topics/architecture.rst """
    import logging
    from time import time
    
    from twisted.internet import defer, task
    from twisted.python.failure import Failure
    
    from scrapy import signals
    from scrapy.core.scraper import Scraper
    from scrapy.exceptions import DontCloseSpider
    from scrapy.http import Response, Request
    from scrapy.utils.misc import load_object
    from scrapy.utils.reactor import CallLaterOnce
    from scrapy.utils.log import logformatter_adapter, failure_to_exc_info
    
    logger = logging.getLogger(__name__)
    
    
    class Slot(object):
    
        def __init__(self, start_requests, close_if_idle, nextcall, scheduler):
            self.closing = False
            self.inprogress = set() # requests in progress
            self.start_requests = iter(start_requests)
            self.close_if_idle = close_if_idle
            self.nextcall = nextcall
            self.scheduler = scheduler
            self.heartbeat = task.LoopingCall(nextcall.schedule)
    
        def add_request(self, request):
            self.inprogress.add(request)
    
        def remove_request(self, request):
            self.inprogress.remove(request)
            self._maybe_fire_closing()
    
        def close(self):
            self.closing = defer.Deferred()
            self._maybe_fire_closing()
            return self.closing
    
        def _maybe_fire_closing(self):
            if self.closing and not self.inprogress:
                if self.nextcall:
                    self.nextcall.cancel()
                    if self.heartbeat.running:
                        self.heartbeat.stop()
                self.closing.callback(None)
    
    
    class ExecutionEngine(object):
    
        def __init__(self, crawler, spider_closed_callback):
            self.crawler = crawler
            self.settings = crawler.settings
            self.signals = crawler.signals
            self.logformatter = crawler.logformatter
            self.slot = None
            self.spider = None
            self.running = False
            self.paused = False
            self.scheduler_cls = load_object(self.settings['SCHEDULER'])
            downloader_cls = load_object(self.settings['DOWNLOADER'])
            self.downloader = downloader_cls(crawler)
            self.scraper = Scraper(crawler)
            self._spider_closed_callback = spider_closed_callback
    
        @defer.inlineCallbacks
        def start(self):
            """Start the execution engine"""
            assert not self.running, "Engine already running"
            self.start_time = time()
            yield self.signals.send_catch_log_deferred(signal=signals.engine_started)
            self.running = True
            self._closewait = defer.Deferred()
            yield self._closewait
    
        def stop(self):
            """Stop the execution engine gracefully"""
            assert self.running, "Engine not running"
            self.running = False
            dfd = self._close_all_spiders()
            return dfd.addBoth(lambda _: self._finish_stopping_engine())
    
        def close(self):
            """Close the execution engine gracefully. If it has already been started, stop it. In all cases, close all spiders and the downloader. """
            if self.running:
                # Will also close spiders and downloader
                return self.stop()
            elif self.open_spiders:
                # Will also close downloader
                return self._close_all_spiders()
            else:
                return defer.succeed(self.downloader.close())
    
        def pause(self):
            """Pause the execution engine"""
            self.paused = True
    
        def unpause(self):
            """Resume the execution engine"""
            self.paused = False
    
        def _next_request(self, spider):
            slot = self.slot
            if not slot:
                return
    
            if self.paused:
                return
    
            while not self._needs_backout(spider):
                if not self._next_request_from_scheduler(spider):
                    break
    
            if slot.start_requests and not self._needs_backout(spider):
                try:
                    request = next(slot.start_requests)
                except StopIteration:
                    slot.start_requests = None
                except Exception:
                    slot.start_requests = None
                    logger.error('Error while obtaining start requests',
                                 exc_info=True, extra={ 'spider': spider})
                else:
                    self.crawl(request, spider)
    
            if self.spider_is_idle(spider) and slot.close_if_idle:
                self._spider_idle(spider)
    
        def _needs_backout(self, spider):
            slot = self.slot
            return not self.running \
                or slot.closing \
                or self.downloader.needs_backout() \
                or self.scraper.slot.needs_backout()
    
        def _next_request_from_scheduler(self, spider):
            slot = self.slot
            request = slot.scheduler.next_request()
            if not request:
                return
            d = self._download(request, spider)
            d.addBoth(self._handle_downloader_output, request, spider)
            d.addErrback(lambda f: logger.info('Error while handling downloader output',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            d.addBoth(lambda _: slot.remove_request(request))
            d.addErrback(lambda f: logger.info('Error while removing request from slot',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            d.addBoth(lambda _: slot.nextcall.schedule())
            d.addErrback(lambda f: logger.info('Error while scheduling new request',
                                               exc_info=failure_to_exc_info(f),
                                               extra={ 'spider': spider}))
            return d
    
        def _handle_downloader_output(self, response, request, spider):
            assert isinstance(response, (Request, Response, Failure)), response
            # downloader middleware can return requests (for example, redirects)
            if isinstance(response, Request):
                self.crawl(response, spider)
                return
            # response is a Response or Failure
            d = self.scraper.enqueue_scrape(response, request, spider)
            d.addErrback(lambda f: logger.error('Error while enqueuing downloader output',
                                                exc_info=failure_to_exc_info(f),
                                                extra={ 'spider': spider}))
            return d
    
        def spider_is_idle(self, spider):
            if not self.scraper.slot.is_idle():
                # scraper is not idle
                return False
    
            if self.downloader.active:
                # downloader has pending requests
                return False
    
            if self.slot.start_requests is not None:
                # not all start requests are handled
                return False
    
            if self.slot.scheduler.has_pending_requests():
                # scheduler has pending requests
                return False
    
            return True
    
        @property
        def open_spiders(self):
            return [self.spider] if self.spider else []
    
        def has_capacity(self):
            """Does the engine have capacity to handle more spiders"""
            return not bool(self.slot)
    
        def crawl(self, request, spider):
            assert spider in self.open_spiders, \
                "Spider %r not opened when crawling: %s" % (spider.name, request)
            self.schedule(request, spider)
            self.slot.nextcall.schedule()
    
        def schedule(self, request, spider):
            self.signals.send_catch_log(signal=signals.request_scheduled,
                    request=request, spider=spider)
            if not self.slot.scheduler.enqueue_request(request):
                self.signals.send_catch_log(signal=signals.request_dropped,
                                            request=request, spider=spider)
    
        def download(self, request, spider):
            d = self._download(request, spider)
            d.addBoth(self._downloaded, self.slot, request, spider)
            return d
    
        def _downloaded(self, response, slot, request, spider):
            slot.remove_request(request)
            return self.download(response, spider) \
                    if isinstance(response, Request) else response
    
        def _download(self, request, spider):
            slot = self.slot
            slot.add_request(request)
            def _on_success(response):
                assert isinstance(response, (Response, Request))
                if isinstance(response, Response):
                    response.request = request # tie request to response received
                    logkws = self.logformatter.crawled(request, response, spider)
                    logger.log(*logformatter_adapter(logkws), extra={ 'spider': spider})
                    self.signals.send_catch_log(signal=signals.response_received, \
                        response=response, request=request, spider=spider)
                return response
    
            def _on_complete(_):
                slot.nextcall.schedule()
                return _
    
            dwld = self.downloader.fetch(request, spider)
            dwld.addCallbacks(_on_success)
            dwld.addBoth(_on_complete)
            return dwld
    
        @defer.inlineCallbacks
        def open_spider(self, spider, start_requests=(), close_if_idle=True):
            assert self.has_capacity(), "No free spider slot when opening %r" % \
                spider.name
            logger.info("Spider opened", extra={ 'spider': spider})
            nextcall = CallLaterOnce(self._next_request, spider)
            scheduler = self.scheduler_cls.from_crawler(self.crawler)
            start_requests = yield self.scraper.spidermw.process_start_requests(start_requests, spider)
            slot = Slot(start_requests, close_if_idle, nextcall, scheduler)
            self.slot = slot
            self.spider = spider
            yield scheduler.open(spider)
            yield self.scraper.open_spider(spider)
            self.crawler.stats.open_spider(spider)
            yield self.signals.send_catch_log_deferred(signals.spider_opened, spider=spider)
            slot.nextcall.schedule()
            slot.heartbeat.start(5)
    
        def _spider_idle(self, spider):
            """Called when a spider gets idle. This function is called when there are no remaining pages to download or schedule. It can be called multiple times. If some extension raises a DontCloseSpider exception (in the spider_idle signal handler) the spider is not closed until the next loop and this function is guaranteed to be called (at least) once again for this spider. """
            res = self.signals.send_catch_log(signal=signals.spider_idle, \
                spider=spider, dont_log=DontCloseSpider)
            if any(isinstance(x, Failure) and isinstance(x.value, DontCloseSpider) \
                    for _, x in res):
                return
    
            if self.spider_is_idle(spider):
                self.close_spider(spider, reason='finished')
    
        def close_spider(self, spider, reason='cancelled'):
            """Close (cancel) spider and clear all its outstanding requests"""
    
            slot = self.slot
            if slot.closing:
                return slot.closing
            logger.info("Closing spider (%(reason)s)",
                        { 'reason': reason},
                        extra={ 'spider': spider})
    
            dfd = slot.close()
    
            def log_failure(msg):
                def errback(failure):
                    logger.error(
                        msg,
                        exc_info=failure_to_exc_info(failure),
                        extra={ 'spider': spider}
                    )
                return errback
    
            dfd.addBoth(lambda _: self.downloader.close())
            dfd.addErrback(log_failure('Downloader close failure'))
    
            dfd.addBoth(lambda _: self.scraper.close_spider(spider))
            dfd.addErrback(log_failure('Scraper close failure'))
    
            dfd.addBoth(lambda _: slot.scheduler.close(reason))
            dfd.addErrback(log_failure('Scheduler close failure'))
    
            dfd.addBoth(lambda _: self.signals.send_catch_log_deferred(
                signal=signals.spider_closed, spider=spider, reason=reason))
            dfd.addErrback(log_failure('Error while sending spider_close signal'))
    
            dfd.addBoth(lambda _: self.crawler.stats.close_spider(spider, reason=reason))
            dfd.addErrback(log_failure('Stats close failure'))
    
            dfd.addBoth(lambda _: logger.info("Spider closed (%(reason)s)",
                                              { 'reason': reason},
                                              extra={ 'spider': spider}))
    
            dfd.addBoth(lambda _: setattr(self, 'slot', None))
            dfd.addErrback(log_failure('Error while unassigning slot'))
    
            dfd.addBoth(lambda _: setattr(self, 'spider', None))
            dfd.addErrback(log_failure('Error while unassigning spider'))
    
            dfd.addBoth(lambda _: self._spider_closed_callback(spider))
    
            return dfd
    
        def _close_all_spiders(self):
            dfds = [self.close_spider(s, reason='shutdown') for s in self.open_spiders]
            dlist = defer.DeferredList(dfds)
            return dlist
    
        @defer.inlineCallbacks
        def _finish_stopping_engine(self):
            yield self.signals.send_catch_log_deferred(signal=signals.engine_stopped)
            self._closewait.callback(None)