python - run - ¿Cuál es la forma correcta de obtener de una corriente?

python asynchronous programming

Question

Tengo un objeto de Connection que se utiliza para contener las secuencias de lectura y escritura de asyncio conexiones asyncio :

class Connection(object):
def __init__(self, stream_in, stream_out):
object.__init__(self)
self.__in = stream_in
self.__out = stream_out
def read(self, n_bytes : int = -1):
return self.__in.read(n_bytes)
def write(self, bytes_ : bytes):
self.__out.write(bytes_)
yield from self.__out.drain()

En el lado del servidor, connected crea un objeto de Connection cada vez que un cliente se conecta y luego lee 4 bytes.

@asyncio.coroutine
def new_conection(stream_in, stream_out):
conn = Connection(stream_in, stream_out)
data = yield from conn.read(4)
print(data)

Y en el lado del cliente, se escriben 4 bytes.

@asyncio.coroutine
def client(loop):
...
conn = Connection(stream_in, stream_out)
yield from conn.write(b''test'')

Esto funciona casi como se esperaba, pero tengo que yield from cada llamada de read y write . He intentado yield from desde dentro de Connection :

def read(self, n_bytes : int = -1):
data = yield from self.__in.read(n_bytes)
return data

Pero en lugar de obtener datos, obtengo una salida como

<generator object StreamReader.read at 0x1109983b8>

Si llamo read y write desde múltiples lugares, preferiría no repetir el yield from s todas las veces; más bien mantenerlos dentro de la Connection . Mi objetivo final es reducir mi función new_conection a esto:

@asyncio.coroutine
def new_conection(stream_in, stream_out):
conn = Connection(stream_in, stream_out)
print(conn.read(4))

Answer 1

Debido a que StreamReader.read es una guía , sus únicas opciones para llamarlo son a) envolviéndolo en una Task o Future y ejecutándolo a través de un bucle de eventos, b) esperándolo de la definición de coroutine definida con async def , o c) usando el yield from con él desde una coroutine definida como una función decorada con @asyncio.coroutine .

Dado que se llama a Connection.read desde un bucle de eventos (a través de la nueva_conexión de new_connection ), no puede volver a usar ese bucle de eventos para ejecutar una Task o Future para StreamReader.read : los bucles de eventos no se pueden iniciar mientras ya se están ejecutando . Tendría que detener el bucle de eventos (desastroso y, probablemente, no es posible hacerlo correctamente) o crear un nuevo bucle de eventos (desordenado y frustrado para el uso de coroutines). Ninguno de los dos es deseable, por lo que Connection.read debe ser una función básica o async .

Las otras dos opciones ( await en una async def asincrónica o el yield from en una función @asyncio.coroutine en @asyncio.coroutine ) son en su mayoría equivalentes. La única diferencia es que se agregaron async def y await en Python 3.5 , por lo que para 3.4, la única opción es usar el yield from y @asyncio.coroutine (no existían versiones anteriores y asyncio antes de 3.4, por lo que otras versiones son irrelevantes). Personalmente, prefiero usar async def y await , porque definir coroutines con async def es más claro que con el decorador.

En resumen: new_connection que Connection.read y new_connection sean coroutines (usando la palabra clave decorator o async ), y use await (o yield from ) cuando llame a otros coroutines ( await conn.read(4) en new_connection , y await self.__in.read(n_bytes) en Connection.read ).

Answer 2

Encontré que una parte del código fuente de StreamReader en la línea 620 es en realidad un ejemplo perfecto del uso de la función.

En mi respuesta anterior, pasé por alto el hecho de que self.__in.read(n_bytes) no es solo una coroutine (que debería haber sabido, considerando que era del módulo de asyncio XD) sino que produce un resultado en línea. Así que, de hecho, es un generador, y tendrá que ceder de él.

Al tomar prestado este bucle del código fuente, su función de lectura debería tener este aspecto:

def read(self, n_bytes : int = -1):
data = bytearray() #or whatever object you are looking for
while 1:
block = yield from self.__in.read(n_bytes)
if not block:
break
data += block
return data

Debido a que self.__in.read(n_bytes) es un generador, debe continuar produciendo hasta que produzca un resultado vacío para indicar el final de la lectura. Ahora su función de lectura debería devolver datos en lugar de un generador. No tendrá que ceder de esta versión de conn.read() .