Method _read_write_loop

test/test_ssltransport.py:298–337

(
        self,
        client_sock: socket.socket,
        server_sock: socket.socket,
        chunks: int = 65536,
    )

        self._start_server(proxy_handler)

    def _read_write_loop(
        self,
        client_sock: socket.socket,
        server_sock: socket.socket,
        chunks: int = 65536,
    ) -> None:
        inputs = [client_sock, server_sock]
        output = [client_sock, server_sock]

        while inputs:
            readable, writable, exception = select.select(inputs, output, inputs)

            if exception:
                # Error occurred with either of the sockets, time to
                # wrap up, parent func will close sockets.
                break

            for s in readable:
                read_socket, write_socket = None, None
                if s == client_sock:
                    read_socket = client_sock
                    write_socket = server_sock
                else:
                    read_socket = server_sock
                    write_socket = client_sock

                # Ensure buffer is not full before writing
                if write_socket in writable:
                    try:
                        b = read_socket.recv(chunks)
                        if len(b) == 0:
                            # One of the sockets has EOFed, we return to close
                            # both.
                            return
                        write_socket.send(b)
                    except ssl.SSLEOFError:
                        # It's possible, depending on shutdown order, that we'll
                        # try to use a socket that was closed between select
                        # calls.
                        return


class TlsInTlsTestCase(SocketDummyServerTestCase):