Are you tired of waiting for network requests to finish before moving on to the next task?
Asyncio is an updated version of asynchat, which simplifies asynchronous clients and servers by making it easier to handle protocols with arbitrary strings or variable-length elements. Instead of creating instances of class `asyncore.dispatcher` and `asynchat.async_chat`, asyncio allows you to define coroutines that can be run concurrently without blocking the main thread of execution.
Here’s an example script using asyncio:
import asyncio
from typing import TextIO, List
# Define a class for the EchoProtocol, which inherits from the asyncio protocol class
class EchoProtocol(asyncio.Protocol):
# Initialize the class with an empty buffer
def __init__(self) -> None:
self.buffer = b''
# Define a method for when a connection is made, takes in a transport object and an address
def connection_made(self, transport: asyncio.Transport, _addr) -> None:
# Print a message indicating a connection has been made
print('Connected by', _addr)
# Set the transport attribute to the given transport object
self.transport = transport
# Define a method for when a disconnection is requested, takes in a future object
def disconnection_requested(self, close_future: asyncio.Future) -> None:
# Print a message indicating the connection has been closed
print('Connection closed')
# Set the result of the future object to None
close_future.set_result(None)
# Define a method for when data is received, takes in a bytes object
def data_received(self, data: bytes) -> None:
# Add the received data to the buffer
self.buffer += data
# Loop through the buffer until a newline character is found
while b'\n' in self.buffer:
# Split the buffer at the first newline character, and assign the result to message and the remaining buffer to self.buffer
message, self.buffer = self.buffer.split(b'\n', 1)
# If the message is empty, continue to the next iteration of the loop
if not message: continue
# Print a message indicating the received message
print('Received:', message.decode())
# Call the sendline method with a "Hello, world!" message
self.sendline(b'Hello, world!')
# Define a method for sending a line of data, takes in a bytes object
def sendline(self, data: bytes) -> None:
# Write the given data and a newline character to the transport object
self.transport.write(data + b'\n')
# Define a class for the EchoServerProtocol, which inherits from the EchoProtocol class
class EchoServerProtocol(EchoProtocol):
# Initialize the class, calling the super class's __init__ method and setting a clients attribute to an empty list
def __init__(self) -> None:
super().__init__()
self.clients = []
# Define a method for when a connection is made, takes in a transport object and an address
def connection_made(self, transport: asyncio.Transport, _addr) -> None:
# Print a message indicating a connection has been made
print('Connected by', _addr)
# Set the transport attribute to the given transport object
self.transport = transport
# Append a tuple of the transport object and a new instance of the EchoProtocol class to the clients list
self.clients.append((transport, EchoProtocol()))
# Loop through all clients except the last one
for client in self.clients[:-1]:
# Call the sendline method with a "Welcome to the echo server!" message
client[1].sendline(b'Welcome to the echo server!\n')
# Define a method for when data is received, takes in a bytes object
def data_received(self, data: bytes) -> None:
# Decode the received data and assign it to the message variable
message = data.decode()
# If the message is "LIST"
if 'LIST' == message:
# Loop through the clients list, using enumerate to get the index and the tuple of transport object and EchoProtocol instance
for i, (_, protocol) in enumerate(self.clients):
# Print a message indicating the client number and the peername of the transport object
print(f"Client {i+1}: {protocol.transport.get_extra_info('peername')}")
# If the message is not "LIST"
else:
# Call the super class's data_received method with the received data
super().data_received(data)
# Define a method for sending a line of data, takes in a bytes object
def sendline(self, data: bytes) -> None:
# Loop through all clients
for client in self.clients:
# If the EchoProtocol instance is an instance of the EchoProtocol class
if isinstance(client[1], EchoProtocol):
# Write the given data and a newline character to the transport object
client[0].write(data + b'\n')
# Define a class for the EchoServer, which inherits from the asyncio protocol class
class EchoServer(asyncio.Protocol):
# Initialize the class with a loop attribute and a server attribute
def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
self.loop = loop
self.server = asyncio.start_server(EchoServerProtocol.factory(), '127.0.0.1', 8888)
# Define a method for running the server
def run(self) -> None:
# Run the server until it is complete
self.loop.run_until_complete(self.server)
# Close the loop
self.loop.close()
# Set the loop attribute to None
self.loop = None
# Close the server
self.server.close()
# If the script is being run directly
if __name__ == '__main__':
# Get the event loop
loop = asyncio.get_event_loop()
# Create an instance of the EchoServer class, passing in the loop
server = EchoServer(loop)
try:
# Run the server
server.run()
# If a keyboard interrupt is received
except KeyboardInterrupt:
# Do nothing
pass
In this example script, we’re creating a simple echo server that listens for incoming connections on port 8888 and sends back “Hello, world!” when it receives a message from the client. The `EchoServerProtocol` class is an instance of our custom protocol `EchoProtocol`, which handles incoming data and errors, while the `EchoServer` class manages multiple network channels simultaneously using asyncio’s event loop.
With this powerful feature at your fingertips, you can create high-performance network servers and clients that handle multiple requests concurrently without blocking the main thread of execution.