[FIXED] How to communicate between a server and two clients using Python's Asynchio?

Issue

I'm working on a basic Python application that utilizies an asynchio server and several clients, but I'm having some issue communicating to the second client.

client.py - Polls directory for new files and sends the contents to the server

# Currently just sends the contents of a known file to the server.
# The polling portion I plan on adding later
import asyncio

async def send_file(filename, server_host, server_port):
    reader, writer = await asyncio.open_connection(server_host, server_port)
    writer.write(filename.encode())
    with open(filename, 'rb') as file:
        while True:
            data = file.read(1024)
            if not data:
                break
            writer.write(data)

    writer.close()

if __name__ == '__main__':
    server_host = '127.0.0.1'  # Change to the server's IP address or hostname.
    server_port = 8888

    filename = 'temp.txt'  # Change to the file you want to send.

    asyncio.run(send_file(filename, server_host, server_port))

server.py- Waits for data from Client #1 and writes it to file. Alert Client #2 when done

import asyncio

async def handle_client(reader, writer):
    try:
        data = await reader.read(1024)
        if not data:
            return
        # Writes data from Client #1
        filename = data.decode().strip()
        print(f"Receiving file: {filename}")
        with open(filename, 'wb') as file:
            while True:
                data = await reader.read(1024)
                if not data:
                    break
                file.write(data)
        print(f"Received file: {filename}")
        
        # Send message to the receiver that the server is done writing
        response = "Server has written data from client.py"
        writer.write(response.encode())
        await writer.drain()
        
    except asyncio.CancelledError:
        pass
    finally:
        writer.close()

async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

if __name__ == '__main__':
    asyncio.run(main())

receiver.py - Print's done status from server

import asyncio

async def receive_data():
    reader, writer = await asyncio.open_connection('127.0.0.1', 8888)

    response = await reader.read(100)
    message = response.decode()
    print(f"Received: {message}")

    print("Closing the connection")
    writer.close()
    await writer.wait_closed()
    
async def main():
    await receive_data()

if __name__ == '__main__':
    server_host = '127.0.0.1'  # Change to the server's IP address or hostname.
    server_port = 8888

    filename = 'temp.txt'  # Change to the file you want to send.

    asyncio.run(main())

I'm able to get client.py and server.py to successfully communicate to the point that the contents of temp.txt located in the client directory is basically copied over to the directory of server.py, but my issue is that receiver.py never prints the done message from the server. I think it may have to do with how I'm setting up the connections, but I'm not sure since I'm fairly new to asynchronous tasks.

Any help is greatly appreciated, thanks!

Solution

Each program will connect as a different client to your server program - which means the "response" message will be written in the connection that is held by "client.py", and no bytes at all will transit in the separate connection, held by "receiver.py".

Asyncio "does its thing" so that each client program connecting to the server will make a different call to "handle_client", which will run concurrently to any other calls. So, in your code, this second, concurrent call, will be made in server.py to handle_client, which will likely just wait forever for some data to arrive on that connection, as your "receiver.py" sends nothing - (or just finish because there is no data - not sure there).

If you need two different programs to communicate with the server, the server will have to 'know' to which of them they are talking, and will need some internal mechanism to transfer its internal data across different "tasks" - each task handling a connection.

You can either start separate servers, in different ports numbers, so that programs with different roles connect each to the appropriate port number, or, add some in band information to your ad-hoc protocol, so that the server knows to identify a "receiver" program, rather than a client.

So, I changed your code to do that: if the filename sent is "receiver", the server handles it separately, in another function that will just send the "completed" message. I also included a minimal counter mechanism, so that any number of "receiver.py" programs can connect to the server at the same time, and when the first "client" program connects and send a file, all receivers are notified.

(client.py has a minimal modification, just sending a newline after the filename. "receiver.py" also was only changed to send "receiver" as the filename, so the server will treat it accordingly)

server.py

import asyncio
from queue import Empty


receivers = set()
receiver_msg_queue = asyncio.Queue()

async def handle_client(reader, writer):
    global receivers
    try:
        # read only one line of text as the file name (readline(), not "read(1024)"
        data = await reader.readline()
        if not data:
            return
        # Writes data from Client #1
        filename = data.decode().strip()

        # if the connection is from a receiver program, handle it in a different function
        # altogether:
        if filename == "receiver":
            receivers.add(asyncio.current_task())
            print("receiver connected")
            await handle_receiver(writer)
            return
        print(f"Receiving file: {filename}")
        # I added a sufix to the file, so I can run things on the same directory:
        with open(filename + ".copy", 'wb') as file:
            while True:
                data = await reader.read(1024)
                if not data:
                    break
                file.write(data)
        print(f"Received file: {filename}")

        # Send message to the receiver that the server is done writing
        response = "Server has written data from client.py"
        receiver_msg_queue.put_nowait((response, set()))
        writer.write(response.encode())
        await writer.drain()

    except asyncio.CancelledError:
        pass
    finally:
        writer.close()

async def handle_receiver(writer):
    """will be called once, concurrently, for each receiver program connected"""
    try:
        this_receiver = asyncio.current_task()
        while True:
            message, processed_at = await receiver_msg_queue.get()
            if this_receiver in processed_at: # this receiver already processed this message,
                # post message back
                receiver_msg_queue.put_nowait((message, processed_at))
                # wait a bit in order for other receives to conclude
                await asyncio.sleep(0.005)
                continue
            # act on this message
            writer.write(message.encode())
            # add receiver to processed_list
            processed_at.add(this_receiver)  # NB: this is fine for asyncio. Multithreading would need 
                                             # to use a "Lock" for this update.
            if processed_at == receivers: # we are the last receiver to process the message
                continue  # do not repost message to queue
            receiver_msg_queue.put_nowait((message, processed_at))
            # wait a bit in order for other receives to conclude processing
            await asyncio.sleep(0.005)

    except asyncio.CancelledError:
        writer.close()


async def main():
    server = await asyncio.start_server(
        handle_client, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

if __name__ == '__main__':
    asyncio.run(main())

client.py

...

async def send_file(filename, server_host, server_port):
    reader, writer = await asyncio.open_connection(server_host, server_port)
    writer.write(filename.encode() + b"\r\n")
    ...

...

receiver.py

...
    reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
    writer.write(b"receiver\r\n")
    response = await reader.readline()
...

Answered By - jsbueno