Call asyncio.get_event_loop when installing the asyncio reactor

scrapy/utils/reactor.py

@@ -60,8 +60,9 @@ def install_reactor(reactor_path, event_loop_path=None):
             if event_loop_path is not None:
                 event_loop_class = load_object(event_loop_path)
                 event_loop = event_loop_class()
+                asyncio.set_event_loop(event_loop)
             else:
-                event_loop = asyncio.new_event_loop()
+                event_loop = asyncio.get_event_loop()
             asyncioreactor.install(eventloop=event_loop)
     else:
         *module, _ = reactor_path.split(".")

tests/CrawlerProcess/asyncio_deferred_signal.py

@@ -0,0 +1,45 @@
+import asyncio
+import sys
+
+import scrapy
+
+from scrapy.crawler import CrawlerProcess
+from twisted.internet.defer import Deferred
+
+
+class UppercasePipeline:
+    async def _open_spider(self, spider):
+        spider.logger.info("async pipeline opened!")
+        await asyncio.sleep(0.1)
+
+    def open_spider(self, spider):
+        loop = asyncio.get_event_loop()
+        return Deferred.fromFuture(loop.create_task(self._open_spider(spider)))
+
+    def process_item(self, item, spider):
+        return {"url": item["url"].upper()}
+
+
+class UrlSpider(scrapy.Spider):
+    name = "url_spider"
+    start_urls = ["data:,"]
+    custom_settings = {
+        "ITEM_PIPELINES": {UppercasePipeline: 100},
+    }
+
+    def parse(self, response):
+        yield {"url": response.url}
+
+
+if __name__ == "__main__":
+    try:
+        ASYNCIO_EVENT_LOOP = sys.argv[1]
+    except IndexError:
+        ASYNCIO_EVENT_LOOP = None
+
+    process = CrawlerProcess(settings={
+        "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
+        "ASYNCIO_EVENT_LOOP": ASYNCIO_EVENT_LOOP,
+    })
+    process.crawl(UrlSpider)
+    process.start()

tests/test_crawler.py

@@ -364,6 +364,22 @@ class CrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase):
         self.assertIn("Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log)
         self.assertIn("Using asyncio event loop: uvloop.Loop", log)
 
+    @mark.skipif(sys.implementation.name == "pypy", reason="uvloop does not support pypy properly")
+    @mark.skipif(platform.system() == "Windows", reason="uvloop does not support Windows")
+    def test_custom_loop_asyncio_deferred_signal(self):
+        log = self.run_script("asyncio_deferred_signal.py", "uvloop.Loop")
+        self.assertIn("Spider closed (finished)", log)
+        self.assertIn("Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log)
+        self.assertIn("Using asyncio event loop: uvloop.Loop", log)
+        self.assertIn("async pipeline opened!", log)
+
+    def test_default_loop_asyncio_deferred_signal(self):
+        log = self.run_script("asyncio_deferred_signal.py")
+        self.assertIn("Spider closed (finished)", log)
+        self.assertIn("Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log)
+        self.assertNotIn("Using asyncio event loop: uvloop.Loop", log)
+        self.assertIn("async pipeline opened!", log)
+
 
 class CrawlerRunnerSubprocess(ScriptRunnerMixin, unittest.TestCase):
     script_dir = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'CrawlerRunner')