我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.ioloop.IOLoop.close()。
def test_add_callback_while_closing(self): # Issue #635: add_callback() should raise a clean exception # if called while another thread is closing the IOLoop. closing = threading.Event() def target(): other_ioloop.add_callback(other_ioloop.stop) other_ioloop.start() closing.set() other_ioloop.close(all_fds=True) other_ioloop = IOLoop() thread = threading.Thread(target=target) thread.start() closing.wait() for i in range(1000): try: other_ioloop.add_callback(lambda: None) except RuntimeError as e: self.assertEqual("IOLoop is closing", str(e)) break
def test_read_while_writeable(self): # Ensure that write events don't come in while we're waiting for # a read and haven't asked for writeability. (the reverse is # difficult to test for) client, server = socket.socketpair() try: def handler(fd, events): self.assertEqual(events, IOLoop.READ) self.stop() self.io_loop.add_handler(client.fileno(), handler, IOLoop.READ) self.io_loop.add_timeout(self.io_loop.time() + 0.01, functools.partial(server.send, b'asdf')) self.wait() self.io_loop.remove_handler(client.fileno()) finally: client.close() server.close()
def test_handler_callback_file_object(self): """The handler callback receives the same fd object it passed in.""" server_sock, port = bind_unused_port() fds = [] def handle_connection(fd, events): fds.append(fd) conn, addr = server_sock.accept() conn.close() self.stop() self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ) with contextlib.closing(socket.socket()) as client_sock: client_sock.connect(('127.0.0.1', port)) self.wait() self.io_loop.remove_handler(server_sock) self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ) with contextlib.closing(socket.socket()) as client_sock: client_sock.connect(('127.0.0.1', port)) self.wait() self.assertIs(fds[0], server_sock) self.assertEqual(fds[1], server_sock.fileno()) self.io_loop.remove_handler(server_sock.fileno()) server_sock.close()
def test_add_callback_while_closing(self): # Issue #635: add_callback() should raise a clean exception # if called while another thread is closing the IOLoop. if IOLoop.configured_class().__name__.endswith('AsyncIOLoop'): raise unittest.SkipTest("AsyncIOMainLoop shutdown not thread safe") closing = threading.Event() def target(): other_ioloop.add_callback(other_ioloop.stop) other_ioloop.start() closing.set() other_ioloop.close(all_fds=True) other_ioloop = IOLoop() thread = threading.Thread(target=target) thread.start() closing.wait() for i in range(1000): try: other_ioloop.add_callback(lambda: None) except RuntimeError as e: self.assertEqual("IOLoop is closing", str(e)) break
def test_multiple_add(self): sock, port = bind_unused_port() try: self.io_loop.add_handler(sock.fileno(), lambda fd, events: None, IOLoop.READ) # Attempting to add the same handler twice fails # (with a platform-dependent exception) self.assertRaises(Exception, self.io_loop.add_handler, sock.fileno(), lambda fd, events: None, IOLoop.READ) finally: self.io_loop.remove_handler(sock.fileno()) sock.close()
def test_remove_without_add(self): # remove_handler should not throw an exception if called on an fd # was never added. sock, port = bind_unused_port() try: self.io_loop.remove_handler(sock.fileno()) finally: sock.close()
def test_add_callback_from_signal_other_thread(self): # Very crude test, just to make sure that we cover this case. # This also happens to be the first test where we run an IOLoop in # a non-main thread. other_ioloop = IOLoop() thread = threading.Thread(target=other_ioloop.start) thread.start() other_ioloop.add_callback_from_signal(other_ioloop.stop) thread.join() other_ioloop.close()
def test_mixed_fd_fileobj(self): server_sock, port = bind_unused_port() def f(fd, events): pass self.io_loop.add_handler(server_sock, f, IOLoop.READ) with self.assertRaises(Exception): # The exact error is unspecified - some implementations use # IOError, others use ValueError. self.io_loop.add_handler(server_sock.fileno(), f, IOLoop.READ) self.io_loop.remove_handler(server_sock.fileno()) server_sock.close()
def test_remove_handler_from_handler(self): # Create two sockets with simultaneous read events. client, server = socket.socketpair() try: client.send(b'abc') server.send(b'abc') # After reading from one fd, remove the other from the IOLoop. chunks = [] def handle_read(fd, events): chunks.append(fd.recv(1024)) if fd is client: self.io_loop.remove_handler(server) else: self.io_loop.remove_handler(client) self.io_loop.add_handler(client, handle_read, self.io_loop.READ) self.io_loop.add_handler(server, handle_read, self.io_loop.READ) self.io_loop.call_later(0.03, self.stop) self.wait() # Only one fd was read; the other was cleanly removed. self.assertEqual(chunks, [b'abc']) finally: client.close() server.close() # Deliberately not a subclass of AsyncTestCase so the IOLoop isn't # automatically set as current.
def tearDown(self): if self.io_loop is not None: self.io_loop.close()
def test_default_current(self): self.io_loop = IOLoop() # The first IOLoop with default arguments is made current. self.assertIs(self.io_loop, IOLoop.current()) # A second IOLoop can be created but is not made current. io_loop2 = IOLoop() self.assertIs(self.io_loop, IOLoop.current()) io_loop2.close()
def tearDown(self): self.io_loop.close()
def test_close_file_object(self): """When a file object is used instead of a numeric file descriptor, the object should be closed (by IOLoop.close(all_fds=True), not just the fd. """ # Use a socket since they are supported by IOLoop on all platforms. # Unfortunately, sockets don't support the .closed attribute for # inspecting their close status, so we must use a wrapper. class SocketWrapper(object): def __init__(self, sockobj): self.sockobj = sockobj self.closed = False def fileno(self): return self.sockobj.fileno() def close(self): self.closed = True self.sockobj.close() sockobj, port = bind_unused_port() socket_wrapper = SocketWrapper(sockobj) io_loop = IOLoop() io_loop.add_handler(socket_wrapper, lambda fd, events: None, IOLoop.READ) io_loop.close(all_fds=True) self.assertTrue(socket_wrapper.closed)
