Asynchronous Server/Client with Python [0x03]: Client Data & Server Broadcast

In part 0x03, we will be making some modifications to our server. By the end of this post, our server should be able to store basic information about each client that connects to it and it should be able to broadcast messages back out to each of the clients.

Client Data

Before we start coding, let’s consider what client data we may want to track:

• StreamReader – Used for reading messages from a specific client
• StreamWriter – Used for writing messages to a specific client
• IP – IP associated with a client
• Port – Port associated with a client
• Nickname – Nickname that will be used for labeling messages sent/received by a client

We will plan to store all of this information in it’s own class named Client. Alongside these properties, we will also plan to include a function for the client class that will retrieve messages from the StreamReader for the server. We will touch more on the specifics of this later in the code section of this post.

The nickname attribute for our client will expand on our client commands. In this case, we will plan for our client to be able to change it’s nickname via a /nick <nickname> command. We will implement this in the same function that handles our client’s quit command.

Broadcasting Messages

Our implementation of broadcasting messages will be simple. In order to implement message broadcasting, our server will need to be modified to keep track of all connected clients. We will plan to keep track of our clients inside the new server property clients.

In our code following this section, you will see that our clients property is just a dictionary of clients that we can iterate over for broadcasting messages.

Code

	import asyncio
	from aioconsole import ainput
	class Client:
	def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
	self.__reader: asyncio.StreamReader = reader
	self.__writer: asyncio.StreamWriter = writer
	self.__ip: str = writer.get_extra_info('peername')[0]
	self.__port: int = writer.get_extra_info('peername')[1]
	self.nickname: str = str(writer.get_extra_info('peername'))
	def __str__(self):
	'''
	Outputs client information as '<nickname> <ip>:<port>'
	'''
	return f"{self.nickname} {self.ip}:{self.port}"
	@property
	def reader(self):
	'''
	Gets the StreamReader associated with a client.
	'''
	return self.__reader
	@property
	def writer(self):
	'''
	Gets the StreamWriter associated with a client.
	'''
	return self.__writer
	@property
	def ip(self):
	'''
	Gets the ip associated with a client
	'''
	return self.__ip
	@property
	def port(self):
	'''
	Gets the port associated with a client
	'''
	return self.__port
	async def get_message(self):
	'''
	Retrieves the incoming message from a client and returns it in string format.
	Parameters
	———-
	client : Client
	The client who's message will be received.
	———-
	Returns
	——-
	str
	The incoming client's message as a string.
	——-
	'''
	return str((await self.reader.read(255)).decode('utf8'))

view raw 0x03_Client.py hosted with ❤ by GitHub

• Lines 6-11: We will be initializing our client with the properties mentioned above when we discussed the client data we wanted to track
• Lines 13-17: In Python, a class can define a __str__ function which defines what is outputted when a class object is directly printed. In this case, we will simply use this to print our client’s nickname, ip, and port.
• Lines 19-45: Define our getter properties for our private variables.
• Lines 47-61: Our function for retrieving messages from a client. Our server will use this to receive and process messages sent by each of our clients.

	import asyncio
	import sys
	import logging
	import pathlib
	import os
	from Client import Client
	from datetime import datetime
	class Server:
	def __init__(self, ip: str, port: int, loop: asyncio.AbstractEventLoop):
	'''
	Parameters
	———-
	ip : str
	IP that the server will be using
	port : int
	Port that the server will be using
	———-
	'''
	self.__ip: str = ip
	self.__port: int = port
	self.__loop: asyncio.AbstractEventLoop = loop
	self.__logger: logging.Logger = self.initialize_logger()
	self.__clients: dict[asyncio.Task, Client] = {}
	self.logger.info(f"Server Initialized with {self.ip}:{self.port}")
	@property
	def ip(self):
	return self.__ip
	@property
	def port(self):
	return self.__port
	@property
	def loop(self):
	return self.__loop
	@property
	def logger(self):
	return self.__logger
	@property
	def clients(self):
	return self.__clients
	def initialize_logger(self):
	'''
	Initializes a logger and generates a log file in ./logs.
	Returns
	——-
	logging.Logger
	Used for writing logs of varying levels to the console and log file.
	——-
	'''
	path = pathlib.Path(os.path.join(os.getcwd(), "logs"))
	path.mkdir(parents=True, exist_ok=True)
	logger = logging.getLogger('Server')
	logger.setLevel(logging.DEBUG)
	ch = logging.StreamHandler()
	fh = logging.FileHandler(
	filename=f'logs/{datetime.now().strftime("%Y-%m-%d-%H-%M-%S")}_server.log'
	)
	ch.setLevel(logging.INFO)
	fh.setLevel(logging.DEBUG)
	formatter = logging.Formatter(
	'[%(asctime)s] – %(levelname)s – %(message)s'
	)
	ch.setFormatter(formatter)
	fh.setFormatter(formatter)
	logger.addHandler(ch)
	logger.addHandler(fh)
	return logger
	def start_server(self):
	'''
	Starts the server on IP and PORT.
	'''
	try:
	self.server = asyncio.start_server(
	self.accept_client, self.ip, self.port
	)
	self.loop.run_until_complete(self.server)
	self.loop.run_forever()
	except Exception as e:
	self.logger.error(e)
	except KeyboardInterrupt:
	self.logger.warning("Keyboard Interrupt Detected. Shutting down!")
	self.shutdown_server()
	def accept_client(self, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter):
	'''
	Callback that is used when server accepts clients
	Parameters
	———-
	client_reader : asyncio.StreamReader
	StreamReader generated by asyncio.start_server.
	client_writer : asyncio.StreamWriter
	StreamWriter generated by asyncio.start_server.
	———-
	'''
	client = Client(client_reader, client_writer)
	task = asyncio.Task(self.handle_client(client))
	self.clients[task] = client
	client_ip = client_writer.get_extra_info('peername')[0]
	client_port = client_writer.get_extra_info('peername')[1]
	self.logger.info(f"New Connection: {client_ip}:{client_port}")
	task.add_done_callback(self.disconnect_client)
	async def handle_client(self, client: Client):
	'''
	Handles incoming messages from client
	Parameters
	———-
	client_reader : asyncio.StreamReader
	StreamReader generated by asyncio.start_server.
	client_writer : asyncio.StreamWriter
	StreamWriter generated by asyncio.start_server.
	———-
	'''
	while True:
	client_message = await client.get_message()
	if client_message.startswith("quit"):
	break
	elif client_message.startswith("/"):
	self.handle_client_command(client, client_message)
	else:
	self.broadcast_message(
	f"{client.nickname}: {client_message}".encode('utf8'))
	self.logger.info(f"{client_message}")
	await client.writer.drain()
	self.logger.info("Client Disconnected!")
	def handle_client_command(self, client: Client, client_message: str):
	client_message = client_message.replace("\n", "").replace("\r", "")
	if client_message.startswith("/nick"):
	split_client_message = client_message.split(" ")
	if len(split_client_message) >= 2:
	client.nickname = split_client_message[1]
	client.writer.write(
	f"Nickname changed to {client.nickname}\n".encode('utf8'))
	return
	client.writer.write("Invalid Command\n".encode('utf8'))
	def broadcast_message(self, message: bytes, exclusion_list: list = []):
	'''
	Parameters
	———-
	message : bytes
	A message consisting of utf8 bytes to broadcast to all clients.
	OPTIONAL exclusion_list : list[Client]
	A list of clients to exclude from receiving the provided message.
	———-
	'''
	for client in self.clients.values():
	if client not in exclusion_list:
	client.writer.write(message)
	def disconnect_client(self, task: asyncio.Task):
	'''
	Disconnects and broadcasts to the other clients that a client has been disconnected.
	Parameters
	———-
	task : asyncio.Task
	The task object associated with the client generated during self.accept_client()
	———-
	'''
	client = self.clients[task]
	self.broadcast_message(
	f"{client.nickname} has left!".encode('utf8'), [client])
	del self.clients[task]
	client.writer.write('quit'.encode('utf8'))
	client.writer.close()
	self.logger.info("End Connection")
	def shutdown_server(self):
	'''
	Shuts down server.
	'''
	self.logger.info("Shutting down server!")
	for client in self.clients.values():
	client.writer.write('quit'.encode('utf8'))
	self.loop.stop()
	if __name__ == '__main__':
	if len(sys.argv) < 3:
	sys.exit(f"Usage: {sys.argv[0]} HOST_IP PORT")
	loop = asyncio.get_event_loop()
	server = Server(sys.argv[1], sys.argv[2], loop)
	server.start_server()

view raw 0x03_Server.py hosted with ❤ by GitHub

For this, I will only be describing the lines of code that are new. If you’re not familiar with another line of code, it’s very likely that it has already been touched on in a previous blog post on this series. Feel free to go back to posts 0x01 or 0x02 for more info!

• Line 6: Import our new Client class. This line assumes that Client.py exists within the same directory.
• Line 25: Initialize our new clients property. This generates an empty dictionary that we will use as clients join/disconnect from our server
• Lines 45-47: Define our new Server property for our private clients variable.
• Line 97: We added this line to ensure that if/when our Server shuts down, all of our clients are properly disconnected.
• Line 111: Initialize a new client as soon as it connects to our server with it’s generated StreamWriter and StreamReader.
• Line 112: Direct our client to operate on our self.handle_client() function which will handle the reading/writing of our client’s messages/commands. Notice that we modified our function so that it accepts a client object instead of the raw StreamWriter and StreamReader objects.
• Line 113: Add our newly created client to our list so that our Server can keep track of it.
• Line 119: This is a new capability provided by asyncio that will run this callback when our handle_client task ends. In this case, we want to execute our server’s new self.disconnect_client() function to ensure that our client is properly disconnected.
• Line 121-148: Our handle_client() function will be used to receive and process any incoming messages from our client. In here, I added the capability for our server to change our client’s nickname. I also went ahead and sent messages through our new broadcast_message() function so that all connected clients can receive messages from other clients.
• Lines 150-161: handle_client_command() will be called if our client ever sends a message preceded by the / character. This will indicate to our server that our client might want to send execute a command. If a valid command is sent to our server, it will be processed accordingly, otherwise the user will be informed that their command is invalid.
• Lines 163-175: broadcast_message() iterates over our server’s clients property and sends a message out to all clients that our server is tracking.
• Lines 177-194: disconnect_client() will disconnect a client from our server. When a client disconnects, our server will broadcast to all connected clients that someone has disconnected.
• Lines 196-203: shutdown_server() will send out quit messages to our clients. As of right now, this doesn’t actually do anything but we may use this later in the future if we decide to create our own python clients instead of using netcat. At the end of this function, we will then finally shutdown our asyncio loop.

Testing our Chatroom

We can test our python chatroom locally by starting up our server with the same command from the last few sections:
python3 Server.py 127.0.0.1 55556

Conclusion

In this post, we modified our Server.py so that it can track our connected clients and broadcast messages to it’s connected clients. Along with these updates, we also updated our server so that it could handle commands from our clients. Hopefully this section provided you with enough room for you to expand on it how you see fit for your own needs.

In the next section of this series, we will focus on expanding our Client.py so that we can stop using netcat.

Link to 0x04 can be found here

As always, if you liked what you read and you’d like to support me, please consider buying me some coffee! Every cup of coffee sent my way helps me stay awake after my full-time job so that I can produce more high quality blog posts! Any and all support would be greatly appreciated!

Sauce, References, & Documentation

Github: https://github.com/D3ISM3/Python-Asynchronous-Server-Client
Part 0x01 Starting Our Server: https://testingonprod.com/2021/10/10/asynchronous-server-client-with-python-0x01-starting-our-server/
Part 0x02 Logging: https://testingonprod.com/2021/10/17/asynchronous-server-client-with-python-0x02-logging/