Prompt for user input using python asyncio.create_server instance

Question

I'm learning about python 3 asyncio library, and I've run into a small issue. I'm trying to adapt the EchoServer example from the python docs to prompt for user input rather than just echo what the client sends.

I thought it would be as easy as just adding a call to input(), but of course input() will block until there is user input which causes the problems.

Ideally I would like to continue receiving data from the client even when the server has nothing to "say". Somewhat like a chat client where each connection is chatting with the server. I'd like to be able to switch to-and-from each individual connection and send input as needed from stdin. Almost like a P2P chat client.

Consider the following modified EchoServer code:

import asyncio

class EchoServerClientProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        reply = input()
        print('Send: {!r}'.format(reply))
        self.transport.write(reply.encode())

        #print('Close the client socket')
        #self.transport.close()

loop = asyncio.get_event_loop()
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

How would I go about getting input form stdin on the server side and specify which connection to send it to while still received inputs from the connected clients?

Answer 1

You can use loop.add_reader schedule a callback to run when data is available on sys.stdin, and then use an asyncio.Queue to pass the stdin data received to your data_received method:

import sys
import asyncio


def got_stdin_data(q):
    asyncio.ensure_future(q.put(sys.stdin.readline()))

class EchoServerClientProtocol(asyncio.Protocol):
   def connection_made(self, transport):
       peername = transport.get_extra_info('peername')
       print('Connection from {}'.format(peername))
       self.transport = transport

   def data_received(self, data):
       message = data.decode()
       print('Data received: {!r}'.format(message))
       fut = asyncio.ensure_future(q.get())
       fut.add_done_callback(self.write_reply)

   def write_reply(self, fut):
       reply = fut.result()
       print('Send: {!r}'.format(reply))
       self.transport.write(reply.encode())

       #print('Close the client socket')
       #self.transport.close()

q = asyncio.Queue()
loop = asyncio.get_event_loop()
loop.add_reader(sys.stdin, got_stdin_data, q)
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)

# Serve requests until CTRL+c is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

The only tricky bit is how we call the Queue.put/Queue.get methods; both of them are coroutines, which can't be called using yield from in the callback or the Protocol instance methods. Instead, we just schedule them with the event loop using asyncio.ensure_future, and then use the add_done_callback method to handle the reply we retrieve from the get() call.

Note: asyncio.ensure_future was introduced in Python 3.4.4. Prior to that, the method was called asyncio.async. Additionally, Python 3.7 introduced asyncio.create_task, which is now the preferred method.

Answer 2

The code in the other answers are good. besides, python3.7+ comes, I think it could be more simple:

   class AInput():
      def __init__(self):
          self.q = asyncio.Queue()

      async def out(self, prompt='input:'):
          print(prompt)
          await self.q.put(sys.stdin.readline().strip())

      async def input(self):
          tasks = [self.out(), self.q.get()]
          res = await asyncio.gather(*tasks)
          return res[1]


  async def demo():
      ain = AInput()
      txt = await ain.input()
      print(f"got: {txt}")

  if __name__ == "__main__":
      asyncio.run(demo())