BUG: client.stream / .aiter_bytes never clears buffer

[gczuczy]

Hello,

I'm trying to read the twitter apiv2 filtered streaming endpoint, and I've noticed that httpx never clears the chunks received from aiter_bytes (haven't tested other aiter_ methods). My code is pretty much from the example here: https://www.python-httpx.org/async/#streaming-responses

Code:

#!/usr/bin/env python3

import json
import asyncio
import os

import httpx

from pprint import pprint

baseurl = 'https://api.twitter.com/2'
# prod
bearer = os.environ['TWITTER_BEARER']

def geturl(*args):
    global baseurl
    return '{b}/{e}'.format(b=baseurl,
                            e = '/'.join(args))

async def run(bearer):
    headers = {'Authorization': 'Bearer {b}'.format(b=bearer)}
    #pprint(headers)
    async with httpx.AsyncClient(timeout=None) as client:
        # and now listen to the stream
        params = {'tweet.fields': 'id,text,created_at,entities',
                  'expansions': 'author_id',
                  'user.fields': 'id,name,username'}
        async with client.stream('GET', geturl('tweets', 'search', 'stream'),
                                 headers = headers, params = params) as resp:
            buff = ''
            async for chunk in resp.aiter_bytes():
                pprint([' --- Got a chunk ---'])
                buff += chunk.decode('utf-8')
                lines = buff.split("\r\n")
                linebuff = ''
                linecnt = 0
                for line in lines:
                    if len(line)==0: continue
                    linebuff += line
                    try:
                        data = json.loads(linebuff)
                    except Exception as e:
                        pprint(['Exception while json load', str(e)])
                        continue
                    linebuff = ''
                    linecnt += 1
                    pprint(['linecnt', linecnt])
                    #pprint(['data', data])
                    pass
                buff += linebuff
                pass

            pass
        pass
    pprint('run exitting')
    loop = asyncio.get_event_loop()
    loop.stop()
    return

loop = asyncio.get_event_loop()
loop.create_task(run(bearer))
asyncio.get_event_loop().run_forever()

Output:

$ TWITTER_BEARER='blah' ~/bug-httpx.py
[' --- Got a chunk ---']
['linecnt', 1]
[' --- Got a chunk ---']
['linecnt', 1]
['linecnt', 2]
[' --- Got a chunk ---']
['linecnt', 1]
['linecnt', 2]
['linecnt', 3]
[' --- Got a chunk ---']
['linecnt', 1]
['linecnt', 2]
['linecnt', 3]
['linecnt', 4]
[' --- Got a chunk ---']
['linecnt', 1]
['linecnt', 2]
['linecnt', 3]
['linecnt', 4]
['linecnt', 5]
[' --- Got a chunk ---']
['linecnt', 1]
['linecnt', 2]
['linecnt', 3]
['linecnt', 4]
['linecnt', 5]
['linecnt', 6]

As you can see at every chunk there's simply one more line. When the buffer gets full, I'm getting disconnected.
The documentation says simply this is the way to read a http stream, however when it's getting iterated over, it's not really an iteration, it's just the same chunk over and over again being extended.

Should I open an issue?

[gczuczy]

Uhm, resetting the buff was at the wrong place. Sorry :)
sorry for rubber ducking here.

[tomchristie]

Indeed. 😀