asyncio project. What am I missing?

I've been working on a client for this chat server but I am running into a bit of a challenge. The server uses Python's 3.4RC1 asyncio module.

Behavior:

My client connects. My second client connects. Either can send messages to the server BUT, the server is not broadcasting them as it should in a normal public chat room.

User1: Hello. Presses Enter.

User2 does not see it.

User2: Anyone there? Presses Enter.

User2 sees User1: Hello. and User2: Anyone there?

Just... strange. Not sure what I'm missing.

Here are the files. Give it a try.

Server:

from socket import socket, SO_REUSEADDR, SOL_SOCKET
from asyncio import Task, coroutine, get_event_loop

class Peer(object):
    def __init__(self, server, sock, name):
        self.loop = server.loop
        self.name = name
        self._sock = sock
        self._server = server
        Task(self._peer_handler())

    def send(self, data):
        return self.loop.sock_send(self._sock, data.encode('utf-8'))

    @coroutine
    def _peer_handler(self):
        try:
            yield from self._peer_loop()
        except IOError:
            pass
        finally:
            self._server.remove(self)

    @coroutine
    def _peer_loop(self):
        while True:
            buf = yield from self.loop.sock_recv(self._sock, 1024)
            if buf == b'':
                break
            self._server.broadcast('%s: %s' % (self.name, buf.decode('utf-8')))

class Server(object):
    def __init__(self, loop, port):
        self.loop = loop
        self._serv_sock = socket()
        self._serv_sock.setblocking(0)
        self._serv_sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
        self._serv_sock.bind(('',port))
        self._serv_sock.listen(5)
        self._peers = []
        Task(self._server())

    def remove(self, peer):
        self._peers.remove(peer)
        self.broadcast('Peer %s quit!' % (peer.name,))

    def broadcast(self, message):
        for peer in self._peers:
            peer.send(message)

    @coroutine
    def _server(self):
        while True:
            peer_sock, peer_name = yield from self.loop.sock_accept(self._serv_sock)
            peer_sock.setblocking(0)
            peer = Peer(self, peer_sock, peer_name)
            self._peers.append(peer)
            self.broadcast('Peer %s connected!' % (peer.name,))

def main():
    loop = get_event_loop()
    Server(loop, 1234)
    loop.run_forever()

if __name__ == '__main__':
    main()

Client:

# import socket
from socket import *
# form socket import socket, bind, listen, recv, send

HOST = 'localhost' #localhost / 192.168.1.1
# LAN - 192.168.1.1
PORT = 1234
s = socket(AF_INET, SOCK_STREAM)# 98% of all socket programming will use AF_INET and SOCK_STREAM
s.connect((HOST, PORT))

while True:
    message = input("Your Message: ")
    encoded_msg = message.encode('utf-8')
    s.send(encoded_msg)
    print('Awaiting Reply..')
    reply = s.recv(1024)
    decoded_reply = reply.decode('utf-8')
    decoded_reply = repr(decoded_reply)
    print('Received ', decoded_reply)

s.close()

Here's the non threaded server code I wrote. works great but ONLY between 2 people. How could this code be updated to broadcast every message received to all clients connected?

# import socket
from socket import *
# form socket import socket, bind, listen, recv, send

HOST = 'localhost' #localhost / 192.168.1.1
# LAN - 192.168.1.1
PORT = 1234
s = socket(AF_INET, SOCK_STREAM) # 98% of all socket programming will use AF_INET and SOCK_STREAM
s.bind((HOST, PORT))
s.listen(5) # how many connections it can receive at one time
conn, addr = s.accept() # accept the connection
print('Connected by', addr) # print the address of the person connected

while True:
    data = conn.recv(1024)
    decoded_data = data.decode('utf-8')
    data = repr(decoded_data)
    print('Received ', decoded_data)
    reply = input("Reply: ")
    encoded_reply = reply.encode('utf-8')
    conn.sendall(encoded_reply)
    print('Server Started')
conn.close()

Okay, let’s think about what your client does. You ask for a message to send, blocking for user input. Then you send that message and receive whatever there is at the server. Afterwards, you block again, waiting for another message.

So when client A sends a text, client B is likely blocking for user input. As such, B won’t actually check if the server sent anything. It will only display what’s there after you have sent something.

Obviously, in a chat, you don’t want to block on user input. You want to continue receiving new messages from the server even if the user isn’t sending messages. So you need to separate those, and run both asynchronously.

I haven’t really done much with asyncio yet, so I don’t really know if this can be nicely done with it, but you essentially just need to put the reading and sending into two separate concurrent tasks, e.g. using threads or concurrent.futures.

A quick example of what you could do, using threading:

from socket import *
from threading import Thread

HOST = 'localhost'
PORT = 1234
s = socket(AF_INET, SOCK_STREAM)
s.connect((HOST, PORT))

def keepReading ():
    try:
        while True:
            reply = s.recv(1024).decode()
            print('Received ', reply)
    except ConnectionAbortedError:
        pass

t = Thread(target=keepReading)
t.start()

try:
    while True:
        message = input('')
        s.send(message.encode())
except EOFError:
    pass
finally:
    s.close()