ストリーム
**********

**ソースコード:** 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=65536, 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* は 64 KiB に設定
   されます。

   残りの引数は直接 "loop.create_connection()" に渡されます。

   注釈:

     *sock* 引数を渡すと、ソケットの所有権は作成された "StreamWriter"
     に移ります。 ソケットを閉じるには、その "close()" メソッドを呼び
     出してください。

   バージョン 3.7 で変更: *ssl_handshake_timeout* 引数が追加されました
   。

   バージョン 3.8 で変更: *happy_eyeballs_delay* と *interleave* が追
   加されました。

   バージョン 3.10 で変更: *loop* パラメータが削除されました。

   バージョン 3.11 で変更: *ssl_shutdown_timeout* パラメータが追加され
   ました。

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

   ソケットサーバーを起動します。

   *client_connected_cb* コールバックは新しいクライアントコネクション
   が確立されるたびに呼び出されます。このコールバックは "StreamReader"
   と "StreamWriter" クラスのインスタンスのペア "(reader, writer)" を2
   つの引数として受け取ります。

   *client_connected_cb* には単純な呼び出し可能オブジェクトか、または
   コルーチン関数 を指定します; コルーチン関数が指定された場合、コール
   バックの呼び出しは自動的に "Task" としてスケジュールされます。

   *limit* は戻り値の "StreamReader" インスタンスが利用するバッファの
   サイズの上限値を設定します。デフォルトでは *limit* は 64 KiB に設定
   されます。

   残りの引数は直接 "loop.create_server()" に渡されます。

   注釈:

     *sock* 引数を渡すと、ソケットの所有権は作成されたサーバーに移りま
     す。 ソケットを閉じるには、サーバーの "close()" メソッドを呼び出
     してください。

   バージョン 3.7 で変更: *ssl_handshake_timeout* と *start_serving*
   パラメータが追加されました。

   バージョン 3.10 で変更: *loop* パラメータが削除されました。

   バージョン 3.11 で変更: *ssl_shutdown_timeout* パラメータが追加され
   ました。

   バージョン 3.13 で変更: *keep_alive* パラメーターが追加されました。

-[ Unix ソケット ]-

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

   Unix ソケットコネクションを確立し、 "(reader, writer)" のオブジェク
   トのペアを返します。

   この関数は "open_connection()" と似ていますが Unix ソケットに対して
   動作します。

   "loop.create_unix_connection()" のドキュメントも参照してください。

   注釈:

     *sock* 引数を渡すと、ソケットの所有権は作成された "StreamWriter"
     に移ります。 ソケットを閉じるには、その "close()" メソッドを呼び
     出してください。

   利用可能な環境: Unix.

   バージョン 3.7 で変更: *ssl_handshake_timeout* パラメーターが追加さ
   れました。 *path* パラメーターに *path-like object* を指定できるよ
   うになりました。

   バージョン 3.10 で変更: *loop* パラメータが削除されました。

   バージョン 3.11 で変更: *ssl_shutdown_timeout* パラメータが追加され
   ました。

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

   Unix のソケットサーバーを起動します。

   "start_server()" と似ていますが Unix ソケットに対して動作します。

   *cleanup_socket* が真の場合、サーバーを閉じると Unix ソケットをファ
   イルシステムから自動的に削除します。 ただし、サーバーを作成した後に
   ソケットが置き換えられていれば削除しません。

   "loop.create_unix_server()" のドキュメントも参照してください。

   注釈:

     *sock* 引数を渡すと、ソケットの所有権は作成されたサーバーに移りま
     す。 ソケットを閉じるには、サーバーの "close()" メソッドを呼び出
     してください。

   利用可能な環境: Unix.

   バージョン 3.7 で変更: *ssl_handshake_timeout* と *start_serving*
   パラメーターが追加されました。 *path* パラメーターに *path-like
   object* を指定できるようになりました。

   バージョン 3.10 で変更: *loop* パラメータが削除されました。

   バージョン 3.11 で変更: *ssl_shutdown_timeout* パラメータが追加され
   ました。

   バージョン 3.13 で変更: *cleanup_socket* パラメーターが追加されまし
   た。


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

class asyncio.StreamReader

   I/O ストリームからデータを読み込む API を提供するリーダーオブジェク
   トです。 *非同期イテラブル* なので、"async for" 文をサポートします
   。

   *StreamReader* オブジェクトを直接インスタンス化することは推奨されま
   せん; 代わりに "open_connection()" や "start_server()" を使ってくだ
   さい。

   feed_eof()

      EOF に達したことをストリームに通知します。

   async read(n=-1)

      ストリームから最大 *n* バイト読み込みます。

      *n* が指定されないか "-1" が指定されていた場合 EOF になるまで読
      み込み、読み込まれた全ての "bytes" を返します。EOF を受信し、か
      つ内部バッファーが空の場合、空の "bytes" オブジェクトを返します
      。

      *n* が "0" なら、ただちに空の "bytes" オブジェクトを返します。

      *n* が正なら、内部バッファで最低でも 1 バイトが利用可能になり次
      第、利用可能な最大 *n* "bytes" を返します。1 バイトも読み込まな
      いうちに EOF を受信したなら、空の "bytes" オブジェクトを返します
      。

   async readline()

      1 行読み込みます。 "行" とは、"\n" で終了するバイト列のシーケン
      スです。

      EOF を受信し、かつ "\n" が見つからない場合、このメソッドは部分的
      に読み込んだデータを返します。

      EOF を受信し、かつ内部バッファーが空の場合、空の "bytes" オブジ
      ェクトを返します。

   async readexactly(n)

      厳密に *n* バイトのデータを読み出します。

      *n* バイトを読み出す前に EOF に達した場合 "IncompleteReadError"
      を送出します。部分的に読み出したデータを取得するには
      "IncompleteReadError.partial" 属性を使ってください。

   async readuntil(separator=b'\n')

      *separator* が見つかるまでストリームからデータを読み出します。

      成功時には、データと区切り文字は内部バッファから削除されます (消
      費されます)。返されるデータの最後には区切り文字が含まれます。

      読み出したデータの量が設定したストリームの上限を超えると
      "LimitOverrunError" 例外が送出されます。このときデータは内部バッ
      ファーに残され、再度読み出すことができます。

      完全な区切り文字が見つかる前に EOF に達すると
      "IncompleteReadError"  例外が送出され、内部バッファーがリセット
      されます。このとき "IncompleteReadError.partial" 属性は区切り文
      字の一部を含むかもしれません。

      *separator* には、区切り文字のタプルも指定できます。 この場合、
      いずれかの区切り文字を末尾に持つ、できる限り短いデータを返します
      。 "LimitOverrunError" の判定では、該当する最短の区切り文字がマ
      ッチしたものとして扱います。

      Added in version 3.5.2.

      バージョン 3.13 で変更: *separator* パラメーターに、区切り文字の
      "tuple" を指定できるようになりました。

   at_eof()

      バッファーが空で "feed_eof()" が呼ばれていた場合 "True" を返しま
      す。


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

class asyncio.StreamWriter

   IO ストリームにデータを書き込むための API を提供するライターオブジ
   ェクトを表します。

   *StreamWriter* オブジェクトを直接インスタンス化することは推奨されま
   せん; 代わりに "open_connection()" や "start_server()" を使ってくだ
   さい。

   write(data)

      このメソッドは、背後にあるソケットにデータ *data* を即座に書き込
      みます。書き込みに失敗した場合、データは送信可能になるまで内部の
      書き込みバッファーに格納されて待機します。

      *data* バッファには、"bytes"、"bytearray"、または C 形式で連続し
      た 1 次元の "memoryview" オブジェクトを指定してください。

      このメソッドは "drain()" メソッドと組み合わせて使うべきです:

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

   writelines(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()" はすぐに返
      ります。

      注釈:

        When the write buffer is below the high watermark, "drain()"
        returns immediately without yielding to the event loop.  As a
        result, code which repeatedly calls "write()" followed by
        "await drain()" may prevent other tasks from running.  To
        prevent blocking behavior, yield to the event loop explicitly
        with "await asyncio.sleep(0)" (see "asyncio.sleep()").

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

      既存のストリームベースの接続を TLS にアップグレードします。

      引数:

      * *sslcontext*: 構成済みの "SSLContext" インスタンスです。

      * *server_hostname*: 対象のサーバーの証明書との照合に使われるホ
        スト名を設定または上書きします。

      * *ssl_handshake_timeout* は、接続を中断する前に TLS ハンドシェ
        イクの完了を待つ時間を秒単位で指定します。 "None" (デフォルト)
        の場合は "60.0" 秒です。

      * *ssl_shutdown_timeout* は、接続を中断する前に SSL の終了処理を
        待つ時間を秒単位で指定します。 "None" (デフォルト) の場合は
        "30.0" 秒です。

      Added in version 3.11.

      バージョン 3.12 で変更: *ssl_shutdown_timeout* パラメータが追加
      されました。

      バージョン 3.15 で変更: "sslcontext.check_hostname" が "True" で
      、"server_hostname" が指定されていない場合に "ValueError" を送出
      するようになりました。

   is_closing()

      ストリームがクローズされたか、またはクローズ処理中の場合に
      "True" を返します。

      Added in version 3.7.

   async wait_closed()

      ストリームがクローズされるまで待機します。

      "close()" の後に呼び出して、下層の接続が閉じられるまで待ってくだ
      さい。 これにより、例えばプログラムを終了する前に、すべてのデー
      タがフラッシュされたことを確認できます。

      Added in version 3.7.


使用例
======


ストリームを使った TCP Echo クライアント
----------------------------------------

"asyncio.open_connection()" 関数を使った TCP Echo クライアントです:

   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 Echo サーバー
------------------------------------

"asyncio.start_server()" 関数を使った TCP Echo サーバーです:

   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()" メソッドを使ってファイル記述子を監視しています。
