Python 3.12 non-blocking TCP Client

Question

I am trying to listen to a TCP server I have no control over (it is a local windows app broadcasting every 500ms on a fixed socket).

I can set up a reader to get a random length packet in blocking mode

import asyncio, telnetlib3, time
TCP_SERVER_ADDRESS = "127.0.0.7"
TCP_SERVER_PORT = 8000

async def get_one_line(reader):
    reply1 = []
    while True:
        c = await reader.read(1)
        if not c:
            break
        if c in ['\r', '\n']:
            break
        reply1.append(c)
    return reply1

async def main():
    reader, writer = await telnetlib3.open_connection(TCP_SERVER_ADDRESS, TCP_SERVER_PORT)
    writer.write("$SYS,INFO")
    count = 0
    while True:
        reply = []
        reply = await get_one_line(reader)
        if reply:
            print('reply:', ''.join(reply))
            del reply[:]
        print(count)
        count += 1
        if count > 10:
            break
        time.sleep(0.1)

asyncio.run(main())

and it prints "count" once per TCP packet as expected (sorry about the XXX, not my data)

reply: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
2
reply: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
3
reply: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
4
:
reply: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
10

When I tried to wrap get_one_line() in an async function, I wrapped myself firmly around the axle. If I remove the "stop after 10", then it just prints count without receiving anything. If I quit after 10, then all kinds of errors (I assume) because the task was left running.

import asyncio, telnetlib3, time
TCP_SERVER_ADDRESS = "127.0.0.7"
TCP_SERVER_PORT = 8000

async def get_one_line(reader):
    reply1 = []
    while True:
        c = await reader.read(1)
        if not c:
            break

        if c in ['\r', '\n']:
            break

        reply1.append(c)
    return reply1

async def get_nonblocking(reader, callback):
    reply2 = await get_one_line(reader)
    callback(reply2)

async def main():
    reader, writer = await telnetlib3.open_connection(TCP_SERVER_ADDRESS, TCP_SERVER_PORT)
    writer.write("$SYS,INFO")
    count = 0
    while True:
        reply = []
        getTask = asyncio.create_task(get_nonblocking(reader,reply.append))
        if reply:
            print('reply:', ''.join(reply))
            del reply[:]
        print(count)
        count += 1
        if count > 10:
            getTask.cancel()
            break
        time.sleep(0.1)

asyncio.run(main())

I assume it's that I do not know how to use asyncio and 2 days of reading have not helped. Any help is appreciated.

Answer 1

You say "What I am trying to accomplish it to receive TCP packets while doing other things."

Look at this simpler example:

from random import random
from asyncio import sleep, get_event_loop, create_task


async def lines():
    # this is like your network code, yielding lines at somewhat unpredictable intervals
    n = 0
    while True:
        n += 1
        yield f'{n}\n'
        await sleep(random() * 3)


async def count_lines(reader):
    # this rountine receives lines, until it has 5 and then terminates
    count = 0
    async for line in reader:
        count += 1
        print(count, line, end='')
        if count >= 5:
            break


async def main():
    # set up the 'reader'
    reader = lines()
    # create a task for `count_lines`
    task = create_task(count_lines(reader))
    # while that task is going, do other stuff
    while not task.done():
        print('doing stuff in main...')
        await sleep(1)
    print('done!')


get_event_loop().run_until_complete(main())

Example output:

doing stuff in main...
1 1
doing stuff in main...
doing stuff in main...
2 2
doing stuff in main...
3 3
doing stuff in main...
doing stuff in main...
doing stuff in main...
4 4
doing stuff in main...
5 5
done!

Taking your code, and structuring it similarly:

from asyncio import sleep, get_event_loop, create_task
from telnetlib3 import open_connection

TCP_SERVER_ADDRESS = "127.0.0.7"
TCP_SERVER_PORT = 8000


async def get_one_line(reader):
    # apparently, you want these lines as a list of characters
    reply1 = []
    while True:
        c = await reader.read(1)
        # this seems like a relevant case - isn't this the end?
        if not c:
            break
        if c in ['\r', '\n']:
            break
        reply1.append(c)
    return reply1


async def read_from(reader):
    count = 0
    while True:
        reply = []  # this is pointless, the next line overwrites it
        reply = await get_one_line(reader)
        if reply:
            print('reply:', ''.join(reply))
            del reply[:]  # also pointless, you're just going to overwrite it
        count += 1
        if count >= 10:
            break


async def main():
    reader, writer = await telnetlib3.open_connection(TCP_SERVER_ADDRESS, TCP_SERVER_PORT)
    writer.write("$SYS,INFO")  # supposedly, this kicks off the server - can't test
    task = create_task(read_from(reader))

    while not task.done():
        print('doing stuff in main...')
        await sleep(1)
        
    print('done!')


get_event_loop().run_until_complete(main())

I can't test this code, since you didn't provide code for the server, but it follows the same pattern.

However, it seems like this would be a bit more straightforward way to go about it:

from asyncio import sleep, get_event_loop, create_task
from telnetlib3 import open_connection

TCP_SERVER_ADDRESS = "127.0.0.7"
TCP_SERVER_PORT = 8000


async def get_one_line(reader):
    line = ''
    while True:
        c = await reader.read(1)
        if not c:
            break  # we're done
        line += c.decode()
        if c in ['\r', '\n']:
            yield line
            line = ''


async def read_from(reader):
    count = 0
    async for line in get_one_line(reader):
        print('reply:', line)
        # can still count, in case we want to stop after 10
        if (count := count + 1) >= 10:
            break


async def main():
    reader, writer = await telnetlib3.open_connection(TCP_SERVER_ADDRESS, TCP_SERVER_PORT)
    writer.write("$SYS,INFO")  # supposedly, this kicks off the server - can't test
    task = create_task(read_from(reader))

    while not task.done():
        print('doing stuff in main...')
        await sleep(1)

    print('done!')


get_event_loop().run_until_complete(main())

It should stop after 10 lines have been received, or if the server no longer responds, whatever happens first. There's no need for while True endless loops in both - the get_one_line() generator will keep generating lines as long as we let it. And count_lines will keep reading from it until the counter expires. If you removed the counter, it would just run forever until the task was stopped.

Meanwhile, main() can go do whatever, as long as there's an await in the loop somewhere frequently enough, so that control can pass to the reader.