Extend StreamResponse documentation

[ttm56p]

Long story short

I need to stream content from given endpoint read from other source.
For that purpose I'm leveraging aiohttp.web.StreamResponse class.
Streaming work well, but can't be finalized when stream is closed from client side.

Documentation does not have example how to handle case - when client interrupted the stream.
I have found StreamResponse.task field. But it not seems to be usable for this purpose.

So I'm asking aiohttp team to extend documentation and show how to deal with this.

Expected behaviour

• Documentation describe how to handle StreamResponse interrupts from client side.
• Endpoint handler ends properly in snippet below.

Actual behaviour

• Documentation does not describe how to handle StreamResponse interrupts from client side.
• Endpoint handler enters in infinite loop in snippet below.

Steps to reproduce

1. Run snippet bellow.
2. Send request to the endpoint.

Snippet

from aiohttp import web


async def infinite_stream(request):
    stream = web.StreamResponse()
    await stream.prepare(request)
    while True:
        if stream.task.done() or stream.task.cancelled():
            break
        await stream.write(b'Hello Aiohttp!')
    # <<<==== Lines below are never executed
    await stream.write_eof()
    return stream


def main():
    app = web.Application()
    app.router.add_get('/infinite-stream', infinite_stream)
    web.run_app(app, host='127.0.0.1', port=8080)


if __name__ == '__main__':
    main()

Command to send request

wget http://127.0.0.1:8080/infinite-stream

Your environment

Ubuntu 17.04, Proxy are not used, Python 3.6, aiohttp 2.2.5, pip 9.0.1

[asvetlov]

Does #2257 satisfy your needs?

[ttm56p]

I think my case is not discovered in this version.
I saw topics describing how to protect code from CancelledError exception, raised manually when server is shutting down and connections to redis/postgres are closing by calling cancel method.
In my case I just stream some content which may be a file or content from response received from other service(similar to what proxy does).

Also I saw this:

:term:web-handler execution could be canceled on every await if client drops connection without reading entire response's BODY.

This is not happen in my case. Maybe I missed an option which enables this behavior or callback fired on request finalization.
Only thing I see when client interrupt connection is:

socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.
socket.send() raised exception.

[ttm56p]

Anyway I have managed to detect when connection is interrupted by client.

from aiohttp import web


async def infinite_stream(request):
    stream = web.StreamResponse()
    await stream.prepare(request)
    while True:
        if request.transport.is_closing(): # <=== This line does the job
            break
        await stream.write(b'Hello Aiohttp!')
    await stream.write_eof()
    return stream

def main():
    app = web.Application()
    app.router.add_get('/infinite-stream', infinite_stream)
    web.run_app(app, host='127.0.0.1', port=8080)


if __name__ == '__main__':
    main()

[asvetlov]

I suggest catching an exception instead of checking transport property: in future the transport might change behavior, HTTP/2 provides new abstraction levels and request/response pair is coupled with a HTTP/2 stream, not the whole connection.

Anyway, if you do think the docs should contain a suggestion for your test case -- please feel free to make a Pull Request.

[ttm56p]

I suggest catching an exception instead of checking transport property

I understand this and agree, but this exception is not raised in my case. It should be raised out of await stream.write(b'Hello Aiohttp!') call. But nothing is happen there.

Anyway, if you do think the docs should contain a suggestion for your test case -- please feel free to make a Pull Request.

I think this is possible when proper solution will be found.

[asvetlov]

I appreciate any help but frankly speaking the issue is not in my priority list.
But I agree that exception should be raised by await resp.write() call.
Did you catch asyncio.CancelledError or Exception?
aiohttp behavior in this case is not clean: sometimes it cancels a task but under other conditions raises other errors.
@fafhrd91 changed exceptions hierarchy and behavior in aiohttp 2.0, I don't remember all details honestly.

[ttm56p]

I appreciate any help but frankly speaking the issue is not in my priority list.

Wish you luck there.

Did you catch asyncio.CancelledError or Exception?

I have used BaseException to be sure nothing is missing.
I suppose this await resp.write() endup here - https://github.com/python/cpython/blob/master/Lib/asyncio/selector_events.py#L750
So I'm curious how CancelledError can be raised at all.

[fafhrd91]

I think, here is what happen.

1. client closes socket
2. asyncio schedule connection_lost call, (I added _conn_lost variable to transport)
3. your code runs in same event loop iteration and writes to transport, but transport is not marked as closed yet. aiohttp's connection_lost is not called yet, so aiohttp does not know about disconnection.

I am not sure if we can solve this without changing asyncio itself.

[asvetlov]

@ttm56p just curious: are its HTTPS streams?

[asvetlov]

Closing the issue after a year of inactivity

[lock]

This thread has been automatically locked since there has not been
any recent activity after it was closed. Please open a new issue for
related bugs.

If you feel like there's important points made in this discussion,
please include those exceprts into that new issue.