What to await when creating new listener with start_server?

[DavesCodeMusings]

I'm trying to determine why my start_server() returns before the connection is ready. I'm thinking I have the awaits wrong, but I can't seem to find the correct way to do it.

Here is the function:

    async def pasv(self, _, writer):
        """
        Start a new data listener on one of the high numbered ports and
        report back to the client.

        Args:
            _ (discard): command does not take parameters
            writer (stream): the FTP client's control connection

        Returns:
            boolean: always True
        """
        host_octets = self.host.replace(".", ",")
        port = self.get_pasv_port()
        port_octet_high = port // 256
        port_octet_low = port % 256
        if self.debug:
            print(f"Starting data listener on port: {self.host}:{port}")
        loop = get_event_loop()
        data_listener = start_server(self.on_data_connect, self.host, port, 1)
        await loop.create_task(data_listener)
        await self.send_response(
            227,
            f"Entering passive mode ={host_octets},{port_octet_high},{port_octet_low}",
            writer,
        )
        return True

This is part of a larger asyncio FTP server project I'm building. See the end for complete code thus far.

There are no errors when the pasv() code runs, but it returns before the data_listener is ready. I know this, because a typical FTP client listing files will first request PASV to establish the data listener, then request LIST after getting the 227 return code. And when it does, the client says: Can't connect to xxxxxx: Connection refused

My current kludge of a fix is to wait half a second in any function that requires the data connection to be available. Obviously, this is a bad solution, but I'm kind of at a loss for where to go with this.

According to the documentation start_server() is a coroutine and loop.create_task() is not. This makes me think I should await start_server() and not create_task() (like I've shown below), but when I do I get an exception TypeError: coroutine expected raised by loop_create()

data_listener = await start_server(self.on_data_connect, self.host, port, 1)
loop.create_task(data_listener)

Removing await from start_server (like below) fixes the exception, but results in the data connection not being ready for the next FTP command (even with my kludge of waiting before listing files.) I suspect this is the case, because on the client side I still get a connection refused error.

data_listener = start_server(self.on_data_connect, self.host, port, 1)
loop.create_task(data_listener)

The only combination of awaiting that seems to work is what I have below, but this seems counter to what the documention says is a coroutine and what is not. And, I still have my problem of having to await sleep_ms() before I list files or do anything else that relies on the data connection being up.

data_listener = start_server(self.on_data_connect, self.host, port, 1)
await loop.create_task(data_listener)

I've been banging my head on this for a while, trying to find a solution, but coming up empty. I'm hoping someone will have a better idea of what's going on and how I might fix it so I don't have to use the silly kludge of a delay before sending anything over the data connection.

Thanks much!

(Edited out complete code listing to make things more readable. It's at https://github.com/DavesCodeMusings/ftpdlite/ if anyone cares.)

[peterhinch]

asyncio.start_server() returns a Server instance. It runs a background task which accepts connections as they occur, calling the callback each time this occurs. So the correct call pattern is:

my_server = await asyncio.start_server(cb, host, port)

The my_server instance should be retained in case you need to issue my_server.close() but, aside from that, all connections are handled by the callback which gets reader and writer streams for that particular connections.

Note that the event loop code is obsolescent. See the tutorial.

Host addresses should be from socket.getaddrinfo().

[DavesCodeMusings]

Excellent! Thank you, @peterhinch I was able to remove the sleep_ms(500) for the functions that require the data connection and no more connection refused error on the client side.

For anyone viewing this later on, here's a summary of the changes I made in the pasv() function:

#        loop = get_event_loop()
        self.data_listener = await start_server(self.on_data_connect, self.host, port, 1)
#        await loop.create_task(data_listener)

Eliminated the loop bits and awaited the start_server()