streams_connect.py

import typing

_T = typing.NewType('_T')
_StreamFactory = typing.Callable[[asyncio.RawStream], _T]

def connect(host=None, port=None, *,
            limit=_DEFAULT_LIMIT,
            ssl=None, family=0, proto=0,
            flags=0, sock=None, local_addr=None,
            server_hostname=None,
            ssl_handshake_timeout=None,
            happy_eyeballs_delay=None, interleave=None, 
            factory=_make_stream: StreamFactory) -> _T:
    """Connect to TCP socket on *host* : *port* address to send and receive data.

    *limit* determines the buffer size limit used by the returned `Stream`
    instance. By default the *limit* is set to 64 KiB.

    The rest of the arguments are passed directly to `loop.create_connection()`.
    """
    # Design note:
    # Don't use decorator approach but explicit non-async
    # function to fail fast and explicitly
    # if passed arguments don't match the function signature
    return _ContextManagerHelper(_connect(host, port, limit,
                                          ssl, family, proto,
                                          flags, sock, local_addr,
                                          server_hostname,
                                          ssl_handshake_timeout,
                                          happy_eyeballs_delay,
                                          interleave,
                                          factory=factory))

async def _connect(host, port,
                  limit,
                  ssl, family, proto,
                  flags, sock, local_addr,
                  server_hostname,
                  ssl_handshake_timeout,
                  happy_eyeballs_delay, interleave,
                  factory):
    loop = events.get_running_loop()
    raw_stream = Stream(mode=StreamMode.READWRITE,
                    limit=limit,
                    loop=loop,
                    _asyncio_internal=True)

    await loop.create_connection(
        lambda: _StreamProtocol(raw_stream, loop=loop,
                                _asyncio_internal=True),
        host, port,
        ssl=ssl, family=family, proto=proto,
        flags=flags, sock=sock, local_addr=local_addr,
        server_hostname=server_hostname,
        ssl_handshake_timeout=ssl_handshake_timeout,
        happy_eyeballs_delay=happy_eyeballs_delay, interleave=interleave)
    stream = await factory(raw_stream)
    return stream

async def _make_stream(raw_stream: RawStream) -> Stream:
    return Stream(raw_stream)