스트림
******

**소스 코드:** Lib/asyncio/streams.py

======================================================================

스트림은 네트워크 연결로 작업하기 위해, async/await에서 사용할 수 있는
고수준 프리미티브입니다. 스트림은 콜백이나 저수준 프로토콜과 트랜스포
트를 사용하지 않고 데이터를 송수신할 수 있게 합니다.

다음은 asyncio 스트림을 사용하여 작성된 TCP 메아리 클라이언트의 예입니
다:

   import asyncio

   async def tcp_echo_client(message):
       reader, writer = await asyncio.open_connection(
           '127.0.0.1', 8888)

       print(f'Send: {message!r}')
       writer.write(message.encode())
       await writer.drain()

       data = await reader.read(100)
       print(f'Received: {data.decode()!r}')

       print('Close the connection')
       writer.close()
       await writer.wait_closed()

   asyncio.run(tcp_echo_client('Hello World!'))

아래의 예제 절도 참조하십시오.

-[ 스트림 함수 ]-

다음 최상위 asyncio 함수를 사용하여 스트림을 만들고 작업할 수 있습니다
.:

async asyncio.open_connection(host=None, port=None, *, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, happy_eyeballs_delay=None, interleave=None)

   네트워크 연결을 만들고 "(reader, writer)" 객체 쌍을 반환합니다.

   반환된 *reader* 와 *writer* 객체는 "StreamReader" 와 "StreamWriter"
   클래스의 인스턴스입니다.

   *limit*는 반환된 "StreamReader" 인스턴스가 사용하는 버퍼 크기 한계
   를 결정합니다. 기본적으로 *limit*는 64KiB로 설정됩니다.

   나머지 인자는 "loop.create_connection()"로 직접 전달됩니다.

   참고:

     The *sock* argument transfers ownership of the socket to the
     "StreamWriter" created. To close the socket, call its "close()"
     method.

   버전 3.7에서 변경: *ssl_handshake_timeout* 매개 변수를 추가했습니다
   .

   버전 3.8에서 변경: Added the *happy_eyeballs_delay* and
   *interleave* parameters.

   버전 3.10에서 변경: Removed the *loop* parameter.

   버전 3.11에서 변경: *ssl_shutdown_timeout* 매개 변수를 추가했습니다
   .

async asyncio.start_server(client_connected_cb, host=None, port=None, *, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True)

   소켓 서버를 시작합니다.

   새 클라이언트 연결이 만들어질 때마다 *client_connected_cb* 콜백이
   호출됩니다. 이 콜백은 두 개의 인자로 "(reader, writer)" 쌍을 받는데
   , "StreamReader" 와 "StreamWriter" 클래스의 인스턴스입니다.

   *client_connected_cb*는 일반 콜러블이나 코루틴 함수 일 수 있습니다;
   코루틴 함수면, 자동으로 "Task"로 예약됩니다.

   *limit*는 반환된 "StreamReader" 인스턴스가 사용하는 버퍼 크기 한계
   를 결정합니다. 기본적으로 *limit*는 64KiB로 설정됩니다.

   나머지 인자는 "loop.create_server()"로 직접 전달됩니다.

   참고:

     The *sock* argument transfers ownership of the socket to the
     server created. To close the socket, call the server's "close()"
     method.

   버전 3.7에서 변경: *ssl_handshake_timeout* 와 *start_serving* 매개
   변수를 추가했습니다.

   버전 3.10에서 변경: Removed the *loop* parameter.

   버전 3.11에서 변경: *ssl_shutdown_timeout* 매개 변수를 추가했습니다
   .

-[ 유닉스 소켓 ]-

asyncio.open_unix_connection(path=None, *, limit=None, ssl=None, sock=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)
:async:

      유닉스 소켓 연결을 만들고 "(reader, writer)" 쌍을 반환합니다.

      "open_connection()"과 비슷하지만, 유닉스 소켓에서 작동합니다.

      "loop.create_unix_connection()"의 설명서도 참조하십시오.

      참고:

        The *sock* argument transfers ownership of the socket to the
        "StreamWriter" created. To close the socket, call its
        "close()" method.

      가용성: Unix.

      버전 3.7에서 변경: *ssl_handshake_timeout* 매개 변수를 추가했습
      니다. *path* 매개 변수는 이제 *경로류 객체*가 될 수 있습니다.

      버전 3.10에서 변경: Removed the *loop* parameter.

   버전 3.11에서 변경: *ssl_shutdown_timeout* 매개 변수를 추가했습니다
   .

async asyncio.start_unix_server(client_connected_cb, path=None, *, limit=None, sock=None, backlog=100, ssl=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True)

   유닉스 소켓 서버를 시작합니다.

   "start_server()"와 비슷하지만, 유닉스 소켓에서 작동합니다.

   "loop.create_unix_server()"의 설명서도 참조하십시오.

   참고:

     The *sock* argument transfers ownership of the socket to the
     server created. To close the socket, call the server's "close()"
     method.

   가용성: Unix.

   버전 3.7에서 변경: *ssl_handshake_timeout* 와 *start_serving* 매개
   변수를 추가했습니다. *path* 매개 변수는 이제 *경로류 객체*가 될 수
   있습니다.

   버전 3.10에서 변경: Removed the *loop* parameter.

   버전 3.11에서 변경: *ssl_shutdown_timeout* 매개 변수를 추가했습니다
   .


StreamReader
============

class asyncio.StreamReader

   IO 스트림에서 데이터를 읽는 API를 제공하는 판독기(reader) 객체를 나
   타냅니다. *비동기 이터러블*로서, 객체는 "async for" 문을 지원합니다
   .

   *StreamReader* 객체를 직접 인스턴스로 만드는 것은 권장되지 않습니다
   . 대신 "open_connection()"과 "start_server()"를 사용하십시오.

   feed_eof()

      Acknowledge the EOF.

   async read(n=-1)

      Read up to *n* bytes from the stream.

      *n*이 제공되지 않거나 "-1"로 설정되면, EOF까지 읽은 다음, 모든
      읽어들인 "bytes"를 반환합니다. EOF를 수신했고 내부 버퍼가 비어
      있으면, 빈 "bytes" 객체를 반환합니다.

      If *n* is "0", return an empty "bytes" object immediately.

      If *n* is positive, return at most *n* available "bytes" as soon
      as at least 1 byte is available in the internal buffer. If EOF
      is received before any byte is read, return an empty "bytes"
      object.

   async readline()

      한 줄을 읽습니다. 여기서 "줄"은 "\n"로 끝나는 바이트의 시퀀스입
      니다.

      EOF를 수신했고, "\n"를 찾을 수 없으면, 이 메서드는 부분적으로 읽
      은 데이터를 반환합니다.

      EOF를 수신했고, 내부 버퍼가 비어 있으면 빈 "bytes" 객체를 반환합
      니다.

   async readexactly(n)

      정확히 *n* 바이트를 읽습니다.

      *n* 바이트를 읽기 전에 EOF에 도달하면, "IncompleteReadError"를
      일으킵니다. 부분적으로 읽은 데이터를 가져오려면
      "IncompleteReadError.partial" 어트리뷰트를 사용하십시오.

   async readuntil(separator=b'\n')

      *separator*가 발견될 때까지 스트림에서 데이터를 읽습니다.

      성공하면, 데이터와 separator가 내부 버퍼에서 제거됩니다 (소비됩
      니다). 반환된 데이터에는 끝에 separator가 포함됩니다.

      읽은 데이터의 양이 구성된 스트림 제한을 초과하면
      "LimitOverrunError" 예외가 발생하고, 데이터는 내부 버퍼에 그대로
      남아 있으며 다시 읽을 수 있습니다.

      완전한 separator가 발견되기 전에 EOF에 도달하면
      "IncompleteReadError" 예외가 발생하고, 내부 버퍼가 재설정됩니다.
      "IncompleteReadError.partial" 어트리뷰트에는 separator 일부가 포
      함될 수 있습니다.

      Added in version 3.5.2.

   at_eof()

      버퍼가 비어 있고 "feed_eof()"가 호출되었으면 "True"를 반환합니다
      .


StreamWriter
============

class asyncio.StreamWriter

   IO 스트림에 데이터를 쓰는 API를 제공하는 기록기(writer) 객체를 나타
   냅니다.

   *StreamWriter* 객체를 직접 인스턴스로 만드는 것은 권장되지 않습니다
   . 대신 "open_connection()"과 "start_server()"를 사용하십시오.

   write(data)

      이 메서드는 하부 소켓에 *data*를 즉시 기록하려고 시도합니다. 실
      패하면, data는 보낼 수 있을 때까지 내부 쓰기 버퍼에 계류됩니다.

      이 메서드는 "drain()" 메서드와 함께 사용해야 합니다:

         stream.write(data)
         await stream.drain()

   writelines(data)

      이 메서드는 하부 소켓에 바이트열의 리스트(또는 임의의 이터러블)
      를 즉시 기록합니다. 실패하면, data는 보낼 수 있을 때까지 내부 쓰
      기 버퍼에 계류됩니다.

      이 메서드는 "drain()" 메서드와 함께 사용해야 합니다:

         stream.writelines(lines)
         await stream.drain()

   close()

      이 메서드는 스트림과 하부 소켓을 닫습니다.

      이 메서드는 필수는 아니지만 "wait_closed()" 메서드와 함께 사용해
      야 합니다:

         stream.close()
         await stream.wait_closed()

   can_write_eof()

      하부 트랜스포트가 "write_eof()" 메서드를 지원하면 "True"를 반환
      하고, 그렇지 않으면 "False"를 반환합니다.

   write_eof()

      버퍼링 된 쓰기 데이터가 플러시 된 후에 스트림의 쓰기 끝을 닫습니
      다.

   transport

      하부 asyncio 트랜스포트를 돌려줍니다.

   get_extra_info(name, default=None)

      선택적 트랜스포트 정보에 액세스합니다; 자세한 내용은
      "BaseTransport.get_extra_info()"를 참조하십시오.

   async drain()

      스트림에 기록을 다시 시작하는 것이 적절할 때까지 기다립니다. 예:

         writer.write(data)
         await writer.drain()

      이것은 하부 IO 쓰기 버퍼와 상호 작용하는 흐름 제어 메서드입니다.
      버퍼의 크기가 높은 수위에 도달하면, 버퍼 크기가 낮은 수위까지 내
      려가서 쓰기가 다시 시작될 수 있을 때까지 *drain()*은 블록합니다.
      기다릴 것이 없으면, "drain()"은 즉시 반환합니다.

   async start_tls(sslcontext, *, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)

      Upgrade an existing stream-based connection to TLS.

      Parameters:

      * *sslcontext*: a configured instance of "SSLContext".

      * *server_hostname*: sets or overrides the host name that the
        target server's certificate will be matched against.

      * *ssl_handshake_timeout* is the time in seconds to wait for the
        TLS handshake to complete before aborting the connection.
        "60.0" seconds if "None" (default).

      * *ssl_shutdown_timeout* is the time in seconds to wait for the
        SSL shutdown to complete before aborting the connection.
        "30.0" seconds if "None" (default).

      Added in version 3.11.

      버전 3.12에서 변경: *ssl_shutdown_timeout* 매개 변수를 추가했습
      니다.

   is_closing()

      스트림이 닫혔거나 닫히고 있으면 "True"를 반환합니다.

      Added in version 3.7.

   async wait_closed()

      스트림이 닫힐 때까지 기다립니다.

      가령 프로그램 종료 전에 모든 데이터가 플러시되었음을 확인하기위
      해, 하부 연결이 닫힐 때까지 기다리려면 "close()" 뒤에 호출해야
      합니다.

      Added in version 3.7.


예제
====


스트림을 사용하는 TCP 메아리 클라이언트
---------------------------------------

"asyncio.open_connection()" 함수를 사용하는 TCP 메아리 클라이언트:

   import asyncio

   async def tcp_echo_client(message):
       reader, writer = await asyncio.open_connection(
           '127.0.0.1', 8888)

       print(f'Send: {message!r}')
       writer.write(message.encode())
       await writer.drain()

       data = await reader.read(100)
       print(f'Received: {data.decode()!r}')

       print('Close the connection')
       writer.close()
       await writer.wait_closed()

   asyncio.run(tcp_echo_client('Hello World!'))

더 보기: TCP 메아리 클라이언트 프로토콜 예제는 저수준
      "loop.create_connection()" 메서드를 사용합니다.


스트림을 사용하는 TCP 메아리 서버
---------------------------------

"asyncio.start_server()" 함수를 사용하는 TCP 메아리 서버:

   import asyncio

   async def handle_echo(reader, writer):
       data = await reader.read(100)
       message = data.decode()
       addr = writer.get_extra_info('peername')

       print(f"Received {message!r} from {addr!r}")

       print(f"Send: {message!r}")
       writer.write(data)
       await writer.drain()

       print("Close the connection")
       writer.close()
       await writer.wait_closed()

   async def main():
       server = await asyncio.start_server(
           handle_echo, '127.0.0.1', 8888)

       addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
       print(f'Serving on {addrs}')

       async with server:
           await server.serve_forever()

   asyncio.run(main())

더 보기: TCP 메아리 서버 프로토콜 예제는 "loop.create_server()" 메서드를 사용
      합니다.


HTTP 헤더 가져오기
------------------

명령 줄로 전달된 URL의 HTTP 헤더를 조회하는 간단한 예제:

   import asyncio
   import urllib.parse
   import sys

   async def print_http_headers(url):
       url = urllib.parse.urlsplit(url)
       if url.scheme == 'https':
           reader, writer = await asyncio.open_connection(
               url.hostname, 443, ssl=True)
       else:
           reader, writer = await asyncio.open_connection(
               url.hostname, 80)

       query = (
           f"HEAD {url.path or '/'} HTTP/1.0\r\n"
           f"Host: {url.hostname}\r\n"
           f"\r\n"
       )

       writer.write(query.encode('latin-1'))
       while True:
           line = await reader.readline()
           if not line:
               break

           line = line.decode('latin1').rstrip()
           if line:
               print(f'HTTP header> {line}')

       # 바디를 무시하고, 소켓을 닫습니다
       writer.close()
       await writer.wait_closed()

   url = sys.argv[1]
   asyncio.run(print_http_headers(url))

사용법:

   python example.py http://example.com/path/page.html

또는 HTTPS를 사용하면:

   python example.py https://example.com/path/page.html


스트림을 사용하여 데이터를 기다리는 열린 소켓 등록
--------------------------------------------------

"open_connection()" 함수를 사용하여 소켓이 데이터를 수신할 때까지 기다
리는 코루틴:

   import asyncio
   import socket

   async def wait_for_data():
       # 저수준 API에 액세스하기 위해 현재 이벤트 루프에 대한 참조를 가져옵니다.
       loop = asyncio.get_running_loop()

       # 연결된 소켓 쌍을 만듭니다.
       rsock, wsock = socket.socketpair()

       # 데이터를 기다리는 열린 소켓을 등록합니다.
       reader, writer = await asyncio.open_connection(sock=rsock)

       # 네트워크로부터의 데이터 수신을 시뮬레이션합니다
       loop.call_soon(wsock.send, 'abc'.encode())

       # 데이터를 기다립니다
       data = await reader.read(100)

       # 데이터를 받았습니다, 할 일을 마쳤습니다: 소켓을 닫습니다.
       print("Received:", data.decode())
       writer.close()
       await writer.wait_closed()

       # 두 번째 소켓을 닫습니다
       wsock.close()

   asyncio.run(wait_for_data())

더 보기:

  프로토콜을 사용하여 데이터를 기다리는 열린 소켓 등록 예제는 저수준
  프로토콜과 "loop.create_connection()" 메서드를 사용합니다.

  파일 기술자에서 읽기 이벤트를 관찰하기 예제는 저수준
  "loop.add_reader()" 메서드를 사용하여 파일 기술자를 관찰합니다.
