bpo-32418: Add get_loop() method on Server, AbstractServer classes (#4997)

* Add abstract get_loop() method to Server, AbstractServer classes.
* Add test cases for get_loop() method in Server, AbstractServer classes
* Add documentation for get_loop() method
parent fc35932a
...@@ -913,6 +913,12 @@ Server ...@@ -913,6 +913,12 @@ Server
The server is closed asynchronously, use the :meth:`wait_closed` The server is closed asynchronously, use the :meth:`wait_closed`
coroutine to wait until the server is closed. coroutine to wait until the server is closed.
.. method:: get_loop()
Gives the event loop associated with the server object.
.. versionadded:: 3.7
.. coroutinemethod:: wait_closed() .. coroutinemethod:: wait_closed()
Wait until the :meth:`close` method completes. Wait until the :meth:`close` method completes.
......
...@@ -185,6 +185,9 @@ class Server(events.AbstractServer): ...@@ -185,6 +185,9 @@ class Server(events.AbstractServer):
if self._active_count == 0: if self._active_count == 0:
self._wakeup() self._wakeup()
def get_loop(self):
return self._loop
def _wakeup(self): def _wakeup(self):
waiters = self._waiters waiters = self._waiters
self._waiters = None self._waiters = None
......
...@@ -155,6 +155,10 @@ class AbstractServer: ...@@ -155,6 +155,10 @@ class AbstractServer:
"""Coroutine to wait until service is closed.""" """Coroutine to wait until service is closed."""
return NotImplemented return NotImplemented
def get_loop(self):
""" Get the event loop the Server object is attached to."""
return NotImplemented
class AbstractEventLoop: class AbstractEventLoop:
"""Abstract event loop.""" """Abstract event loop."""
......
...@@ -2826,6 +2826,19 @@ else: ...@@ -2826,6 +2826,19 @@ else:
get_running_loop_impl = events._c_get_running_loop get_running_loop_impl = events._c_get_running_loop
get_event_loop_impl = events._c_get_event_loop get_event_loop_impl = events._c_get_event_loop
class TestServer(unittest.TestCase):
def test_get_loop(self):
loop = asyncio.new_event_loop()
proto = MyProto(loop)
server = loop.run_until_complete(loop.create_server(lambda: proto, '0.0.0.0', 0))
self.assertEqual(server.get_loop(), loop)
loop.close()
class TestAbstractServer(unittest.TestCase):
def test_get_loop(self):
self.assertEqual(events.AbstractServer().get_loop(), NotImplemented)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()
Add get_loop() method to Server and AbstractServer classes.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment