I wrote a server-client application in Python where the server sends a never-ending stream of messages to connecting clients.
I was in need for such an architecture since I want to retrieve and process data from a public streaming endpoint, however the company limits the number of active connections to one per user.
With this setup, I can start my server ONCE and then forward the data to any number of connecting clients. This allows me to process data in different ways simultaneously (e.g. one client might write items to a DB, while another calculates some metrics etc.). Further, it allows me to develop a new module and connect it once it is ready for deployment, without interfering with my production modules.
Some details:
I send individual variable-length items by prefixing them with their length and then read this number of bytes of on the client side.
To avoid that a buggy client brings down the server, I have a capped message queue for each client and the client gets disconnected once the queue reaches a defined limit.
It's my first time writing network code in Python (and in general), so I'm looking for ANY advice or suggestions.
Thanks!
Server:
import socket
import struct
import time
from threading import Thread
from queue import Queue
MAX_CLIENTS = 16
MAX_CLIENT_QSIZE = 600
HEADER_FORMAT = struct.Struct('!I')
class _ClientHandler(Thread):
def __init__(self, sock, address, server):
Thread.__init__(self, daemon = True)
self.sock = sock
self.address = address
self.server = server # Reference to server
self.queue = Queue() # Message queue
def _send_item(self, item):
"""Send item to client. Blocks thread."""
self.sock.sendall(HEADER_FORMAT.pack(len(item)))
self.sock.sendall(item.encode('utf-8'))
def run(self):
"""Pop items from queue and sent to client."""
while True:
try:
item = self.queue.get()
self._send_item(item)
except BrokenPipeError:
self.server.delete_client(self.ident)
except OSError:
self.server.delete_client(self.ident)
except:
raise
class StreamServer(Thread):
def __init__(self, address):
Thread.__init__(self, daemon = True)
self.address = address
self._clients = dict()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
"""Disconnect clients and remove client threads when leaving context."""
for ident in self._clients.keys():
self.delete_client(ident)
def queue_item(self, item):
"""Push item to every client queue."""
for ident, client in self._clients.items():
if client.queue.qsize() >= MAX_CLIENT_QSIZE:
print('Client', client.sock.getsockname(), 'reached maximum qsize')
self.delete_client(ident)
client.queue.put(item)
def get_qsizes(self):
"""Get list of client queue sizes."""
return [client.queue.qsize() for _, client in self._clients.items()]
def delete_client(self, ident):
"""Close client socket and delete client thread."""
self._clients[ident].sock.close()
del self._clients[ident] # Delete only client thread reference (garbage collection?)
def run(self):
"""Listen for incomming client connections."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(self.address)
sock.listen(4)
print('Listening at', sock.getsockname())
while True:
try:
# Refuse connections once client limit reached
while len(self._clients) >= MAX_CLIENTS:
time.sleep(60)
client_sock, client_addr = sock.accept()
client_sock.shutdown(socket.SHUT_RD)
client = _ClientHandler(client_sock, client_addr, self)
print('Connection from', client_sock.getsockname())
client.start()
self._clients[client.ident] = client
except:
# ...
# What exceptions to expect?
raise
if __name__ == '__main__':
with StreamServer(('localhost', 1060)) as server:
server.start() # Start server thread
last_time = time.time()
for item in range(1,1000):
server.queue_item(str(item))
time.sleep(1)
if time.time() - last_time > 10.0:
print('Client queue sizes:', server.get_qsizes())
last_time = time.time()
Client:
import socket
import struct
HEADER_FORMAT = struct.Struct('!I')
class StreamClient():
def __init__(self, address):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect(address)
self.sock.shutdown(socket.SHUT_WR)
print('Connected to', self.sock.getsockname())
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
self.sock.close()
def _recvall(self, size):
data = b''
while size:
block = self.sock.recv(size)
if not block:
raise EOFError('Socket closed during message transmission')
data += block
size -= len(block)
return data
def __iter__(self):
while True:
try:
(item_size,) = HEADER_FORMAT.unpack(self._recvall(HEADER_FORMAT.size))
yield self._recvall(item_size).decode('utf-8')
except EOFError:
return
except:
# ...
# What exceptions to expect?
raise
if __name__ == '__main__':
with StreamClient(('localhost', 1060)) as stream:
for item in stream:
print(item)
