As a best-selling author, I invite you to explore my books on Amazon. Don't forget to follow me on Medium and show your support. Thank you! Your support means the world!
Network programming often seems like a world of obscure rules and invisible conversations between machines. I remember when I first looked at a packet capture, seeing all those hexadecimal numbers scrolling by, and wondering how anyone made sense of it. The truth is, it's just a structured conversation, and Python gives us the tools to both listen in and participate. Let's talk about how you can build these conversations yourself.
When you need total control, you work with raw sockets. This is like being given sheets of metal and told to build a car from scratch. You handle every bolt. You decide the exact shape of every piece of data that leaves your machine. It's powerful but detailed work.
import socket
import struct
# Let's say we're designing a simple heartbeat protocol.
# Our packet will have: [VERSION:1 byte][SEQUENCE:4 bytes][MESSAGE:variable]
def create_heartbeat(sequence_id, status_message):
version = 2 # Protocol version 2
# Pack the header: '!' for network byte order, 'B' for 1 byte, 'I' for 4-byte integer
header = struct.pack('!BI', version, sequence_id)
packet = header + status_message.encode('utf-8')
return packet
# To send it, we create a socket and specify we want raw IP.
def send_raw_heartbeat(dest_ip, sequence, message):
# Note: On many systems, creating raw sockets requires administrator privileges.
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
# In a real scenario, we'd build full IP and transport headers here.
# For simplicity, we're just sending our custom payload.
custom_packet = create_heartbeat(sequence, message)
try:
sock.sendto(custom_packet, (dest_ip, 0)) # Port 0 for raw
print(f"Sent heartbeat #{sequence}")
finally:
sock.close()
# On the receiving end, we need to listen and parse.
def listen_for_heartbeats(bind_ip):
# We'll listen for all IP packets. This captures everything coming to the interface.
conn = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_IP)
conn.bind((bind_ip, 0))
print(f"Listening for raw packets on {bind_ip}...")
while True:
raw_data, addr = conn.recvfrom(65535)
# The raw_data includes the IP header. Let's skip it (first 20 bytes).
if len(raw_data) > 24: # Ensure we have data beyond the IP header
potential_payload = raw_data[20:]
# Try to parse it as our heartbeat format
try:
version, seq = struct.unpack('!BI', potential_payload[:5])
message = potential_payload[5:].decode('utf-8', errors='ignore')
print(f"Heartbeat from {addr}: Seq={seq}, Status='{message}'")
except struct.error:
pass # Wasn't our protocol packet
The code above is a basic skeleton. In reality, you'd need to construct proper IP and UDP/TCP headers for the packet to be routable. Raw sockets are your gateway to tools like ping, traceroute, or custom network diagnostics. I once used a similar approach to diagnose a weird proprietary protocol a legacy piece of hardware was using, by watching the bytes it sent and then mimicking them.
Moving up a layer of abstraction, many protocols are like a formal dance. They have specific steps: introduce yourselves, verify identities, exchange information, say goodbye. Programming this as a series of if statements quickly becomes messy. A state machine keeps it clean.
Think of a state machine as a flowchart for your protocol. Your program can only be in one state at a time, like "Waiting for Hello" or "Sending Data." Events, like receiving a network message, cause it to move to a new state.
from enum import Enum, auto
class ChatProtocolState(Enum):
DISCONNECTED = auto()
CONNECTED = auto()
AWAITING_USERNAME = auto()
AWAITING_PASSWORD = auto()
IN_CHAT = auto()
ERROR = auto()
class SimpleChatServer:
def __init__(self):
self.state = ChatProtocolState.DISCONNECTED
self.client_username = None
def handle_event(self, event_type, data=None):
"""The core state machine logic."""
old_state = self.state
if self.state == ChatProtocolState.DISCONNECTED and event_type == "client_connected":
self.state = ChatProtocolState.CONNECTED
response = "HELLO Please enter your username:"
elif self.state == ChatProtocolState.CONNECTED and event_type == "message_received":
if data and data.startswith("USER "):
self.client_username = data[5:].strip()
self.state = ChatProtocolState.AWAITING_PASSWORD
response = f"Hello {self.client_username}. Please enter password:"
else:
response = "ERROR Expected 'USER <username>'"
self.state = ChatProtocolState.ERROR
elif self.state == ChatProtocolState.AWAITING_PASSWORD and event_type == "message_received":
# In reality, you'd verify the password.
if data and data.startswith("PASS "):
self.state = ChatProtocolState.IN_CHAT
response = "OK Login successful. Type your messages."
else:
response = "ERROR Bad password"
self.state = ChatProtocolState.ERROR
elif self.state == ChatProtocolState.IN_CHAT and event_type == "message_received":
response = f"BROADCAST [{self.client_username}]: {data}"
else:
# No valid transition for this event in the current state
response = "ERROR Invalid command or state"
print(f"State: {old_state} -> {self.state} | Event: {event_type}")
return response
# Simulating a conversation with the server
server = SimpleChatServer()
print(server.handle_event("client_connected"))
print(server.handle_event("message_received", "USER alice"))
print(server.handle_event("message_received", "PASS secret123"))
print(server.handle_event("message_received", "Hello everyone!"))
This structure makes the logic clear and easy to debug. You can look at the handle_event method and see exactly what should happen at each point in the conversation. I find this indispensable for anything more complex than a simple request-response.
Now, let's talk about binary protocols. Not everything is friendly text like HTTP or chat messages. Devices like sensors, PLCs, or game servers often communicate in pure bytes to save space and time. Python's struct module is your best friend here.
It lets you translate between Python values and C-style binary data. You define a "format string" that acts like a blueprint for the data.
import struct
# Example: A GPS tracker protocol.
# Packet format: [HEADER 'GP'(2B)][LATITUDE:float][LONGITUDE:float][SPEED: unsigned short][STATUS: byte]
def create_gps_packet(lat, lon, speed_kmh, status):
# Format: '!' network byte order, '2s' two-byte string, 'ff' two floats, 'H' unsigned short, 'B' unsigned byte
packet_data = struct.pack('!2sffHB', b'GP', lat, lon, speed_kmh, status)
return packet_data
def parse_gps_packet(binary_data):
try:
# Unpack according to the same format
header, latitude, longitude, speed, status = struct.unpack('!2sffHB', binary_data)
if header != b'GP':
raise ValueError("Invalid packet header")
return {
'latitude': latitude,
'longitude': longitude,
'speed_kmh': speed,
'status': {
'moving': bool(status & 0x01),
'gps_fix': bool(status & 0x02),
'low_battery': bool(status & 0x04)
}
}
except struct.error as e:
print(f"Failed to unpack data: {e}")
return None
# Let's pack and unpack some data
packet = create_gps_packet(40.7128, -74.0060, 65, 0b011) # Moving, with GPS fix
print(f"Raw packet bytes: {packet.hex()}")
parsed = parse_gps_packet(packet)
if parsed:
print(f"Parsed: Lat {parsed['latitude']}, Lon {parsed['longitude']}, Speed {parsed['speed_kmh']}")
print(f"Status: Moving={parsed['status']['moving']}")
This is how you talk to a lot of hardware. The format string syntax is powerful, letting you handle integers of different sizes, floats, and even specific byte alignments. Getting the format string right is 90% of the battle.
When your data structures become complex, constantly writing struct.pack and .unpack gets tedious and error-prone. This is where serialization libraries like Protocol Buffers (protobuf) shine. You define your data structure in a simple .proto file, and it generates Python classes for you. The resulting binary is compact, fast to parse, and automatically versionable.
First, you define your protocol in a .proto file, say sensor.proto:
syntax = "proto3";
message SensorReading {
string sensor_id = 1;
int64 timestamp_unix_ms = 2;
double value = 3;
enum Unit {
CELSIUS = 0;
FAHRENHEIT = 1;
PASCAL = 2;
VOLT = 3;
}
Unit unit = 4;
map<string, string> metadata = 5; // Extra key-value pairs
}
You then use the protoc compiler to generate sensor_pb2.py. Now, in your Python code:
# Assuming sensor_pb2 was generated by protoc
import sensor_pb2
def send_sensor_data():
# Creating a message is like using a familiar Python class.
reading = sensor_pb2.SensorReading()
reading.sensor_id = "thermometer_floor_2"
reading.timestamp_unix_ms = 1700000000000
reading.value = 22.5
reading.unit = sensor_pb2.SensorReading.CELSIUS
reading.metadata["location"] = "Room 205"
reading.metadata["calibration_date"] = "2023-11-01"
# Serialize to a compact binary string
binary_data = reading.SerializeToString()
print(f"Original object size: ~{len(str(reading))} chars")
print(f"Serialized size: {len(binary_data)} bytes")
# This binary_data is what you send over the network
return binary_data
def receive_sensor_data(binary_data):
# Deserializing is just as easy
new_reading = sensor_pb2.SensorReading()
new_reading.ParseFromString(binary_data)
print(f"Received from {new_reading.sensor_id}: {new_reading.value} {new_reading.Unit.Name(new_reading.unit)}")
print(f"Metadata: {dict(new_reading.metadata)}")
return new_reading
# Simulate a network round-trip
data_to_send = send_sensor_data()
print(f"Wire format (hex): {data_to_send.hex()[:50]}...")
received_object = receive_sensor_data(data_to_send)
The beauty is in the schema. If you later add a new field like accuracy_rating to the SensorReading message, old code that doesn't know about it can still parse the binary, ignoring the new field. This makes evolving your protocol over time much safer.
All these techniques involve talking over a network, which means managing connections. Opening and closing a socket for every single message is slow, like hanging up the phone after every sentence in a conversation. Connection pooling keeps a set of open, ready-to-use sockets.
A pool manages the lifecycle: it creates connections up to a limit, gives them out when you need them, and recycles them when you're done. It handles timeouts, broken connections, and cleanup.
import socket
import threading
from queue import Queue
import time
class SimpleConnectionPool:
def __init__(self, host, port, max_connections=5):
self.host = host
self.port = port
self.max_connections = max_connections
self._pool = Queue(maxsize=max_connections)
self._in_use = set()
self._lock = threading.Lock()
# Pre-create some idle connections
for _ in range(min(2, max_connections)):
self._create_idle_connection()
def _create_connection(self):
"""Create a fresh socket connection."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(3.0) # Timeout for operations
sock.connect((self.host, self.port))
return sock
def _create_idle_connection(self):
"""Add a new connection to the idle pool."""
try:
conn = self._create_connection()
self._pool.put(conn, block=False)
except (socket.error, Queue.Full):
pass
def get_connection(self):
"""Get a connection from the pool. May create a new one if under limit."""
with self._lock:
# 1. Try to get an idle connection
try:
conn = self._pool.get_nowait()
self._in_use.add(conn)
return conn
except Queue.Empty:
pass
# 2. Can we create a new one?
if len(self._in_use) < self.max_connections:
try:
conn = self._create_connection()
self._in_use.add(conn)
return conn
except socket.error as e:
raise ConnectionError(f"Failed to create connection: {e}")
else:
raise ConnectionError("Connection pool exhausted")
def release_connection(self, conn):
"""Return a connection to the pool after use."""
with self._lock:
if conn in self._in_use:
self._in_use.remove(conn)
# Test if the connection is still alive before recycling
try:
# Send a zero-byte probe (platform dependent)
conn.settimeout(0.1)
conn.send(b'')
self._pool.put(conn, block=False) # It's alive, reuse it
except (socket.timeout, socket.error):
# It's dead, close it. A new one will be created on demand.
conn.close()
else:
# This connection wasn't from our pool, just close it.
conn.close()
def close_all(self):
"""Close all connections in the pool."""
while not self._pool.empty():
try:
conn = self._pool.get_nowait()
conn.close()
except Queue.Empty:
break
for conn in list(self._in_use):
conn.close()
self._in_use.clear()
# Example usage simulating a client that makes rapid requests
def worker(pool, worker_id):
for i in range(3):
try:
conn = pool.get_connection()
# Simulate using the connection
request = f"GET /data/{worker_id}_{i} HTTP/1.0\\r\\n\\r\\n"
conn.send(request.encode())
# Read a small part of the response (simulated)
response = conn.recv(1024)
print(f"Worker {worker_id}, request {i}: Got {len(response)} bytes")
time.sleep(0.1) # Simulate processing
pool.release_connection(conn)
except Exception as e:
print(f"Worker {worker_id} failed: {e}")
# Simulate a multi-threaded scenario
pool = SimpleConnectionPool("httpbin.org", 80, max_connections=3)
threads = []
for w_id in range(5): # 5 workers, but only 3 connections max
t = threading.Thread(target=worker, args=(pool, w_id))
t.start()
threads.append(t)
for t in threads:
t.join()
pool.close_all()
print("Test complete.")
In a production system, you'd add more features like health checks, connection age limits, and better error handling. But this pattern is key for performance in database clients, HTTP clients, or any service you call repeatedly.
Finally, let's discuss a crucial but often invisible part of protocols: flow control. Imagine pouring water from a big jug into a small cup. You need to pause to let the cup empty, or you'll overflow it. Network flow control is the same idea. A fast sender can overwhelm a slow receiver.
TCP has this built-in, but if you're implementing a custom protocol over UDP or raw sockets, you need to manage it yourself. A simple method is a sliding window. The receiver tells the sender, "I have space for N messages." The sender can only send that many before waiting for an acknowledgement.
import time
import threading
from collections import deque
class SimpleWindowFlowControl:
def __init__(self, window_size=5):
self.window_size = window_size
self.sent_messages = {} # seq_num -> (data, timestamp)
self.acknowledged = set()
self.next_sequence = 0
self.last_ack_received = -1
self.lock = threading.Lock()
def can_send(self):
"""Check if we have room in the window to send a new message."""
with self.lock:
in_flight = len(self.sent_messages)
return in_flight < self.window_size
def send_message(self, data):
"""Send a message if the window allows. Returns sequence number or None."""
with self.lock:
if not self.can_send():
return None
seq = self.next_sequence
self.next_sequence += 1
self.sent_messages[seq] = (data, time.time())
print(f"[SENDER] Sent message #{seq}: '{data}'")
return seq
def receive_ack(self, ack_sequence_num):
"""Process an acknowledgement from the receiver."""
with self.lock:
print(f"[SENDER] Received ACK for #{ack_sequence_num}")
self.last_ack_received = max(self.last_ack_received, ack_sequence_num)
# Remove all messages up to and including this ACK
to_delete = [seq for seq in self.sent_messages if seq <= ack_sequence_num]
for seq in to_delete:
del self.sent_messages[seq]
def check_timeouts(self, timeout_seconds=2.0):
"""Resend any messages that haven't been ACKed in time."""
with self.lock:
now = time.time()
to_resend = []
for seq, (data, sent_time) in self.sent_messages.items():
if now - sent_time > timeout_seconds:
to_resend.append((seq, data))
# Update timestamp as we resend
self.sent_messages[seq] = (data, now)
for seq, data in to_resend:
print(f"[SENDER] Timeout, resending #{seq}: '{data}'")
return to_resend
# Simulated Receiver
def receiver_process(sender, delay=0.3, loss_rate=0.2):
"""Simulates a slower, sometimes lossy receiver."""
import random
expected_seq = 0
received_buffer = {}
def send_ack(seq):
time.sleep(delay) # Simulate network+processing delay
# Simulate packet loss for ACKs
if random.random() > loss_rate:
sender.receive_ack(seq)
else:
print(f"[RECEIVER] (Simulated ACK loss for #{seq})")
while True:
# In reality, this would get data from a socket.
# We'll simulate it by checking the sender's 'sent' list.
time.sleep(0.1)
with sender.lock:
# Look for the next message we expect
if expected_seq in sender.sent_messages:
data, _ = sender.sent_messages[expected_seq]
print(f"[RECEIVER] Got message #{expected_seq}: '{data}'")
# Store in buffer in case we get out-of-order messages
received_buffer[expected_seq] = data
# Acknowledge the highest consecutive sequence number we have
while expected_seq in received_buffer:
send_ack(expected_seq)
del received_buffer[expected_seq]
expected_seq += 1
# Test the flow control
sender = SimpleWindowFlowControl(window_size=3)
recv_thread = threading.Thread(target=receiver_process, args=(sender,), daemon=True)
recv_thread.start()
messages = ["Hello", "World", "Flow", "Control", "Is", "Important"]
for msg in messages:
while not sender.can_send():
print("[SENDER] Window full, waiting...")
time.sleep(0.5)
# Check for timeouts while waiting
sender.check_timeouts()
sender.send_message(msg)
time.sleep(0.2) # Small pause between sends
# Let the test run for a bit
time.sleep(5)
print("\\nSimulation ending.")
This sliding window mechanism is the heart of reliable data transfer. It ensures the receiver isn't overwhelmed and that no data is lost, even on an unreliable network. Implementing it teaches you a lot about how core internet protocols work under the hood.
Each of these eight techniques—raw sockets, state machines, binary parsing with struct, serialization with Protobuf, connection pooling, and flow control—is a tool. You might use one or combine several, depending on whether you're building a high-speed game server, talking to an industrial sensor, or creating a robust messaging queue. The key is understanding the conversation you need to have and then picking the right words and grammar, in the form of bytes and logic, to make it happen reliably. Start simple, test each part, and you'll find that the invisible world of network protocols becomes something you can not only see but actually build.
📘 Checkout my latest ebook for free on my channel!
Be sure to like, share, comment, and subscribe to the channel!
101 Books
101 Books is an AI-driven publishing company co-founded by author Aarav Joshi. By leveraging advanced AI technology, we keep our publishing costs incredibly low—some books are priced as low as $4—making quality knowledge accessible to everyone.
Check out our book Golang Clean Code available on Amazon.
Stay tuned for updates and exciting news. When shopping for books, search for Aarav Joshi to find more of our titles. Use the provided link to enjoy special discounts!
Our Creations
Be sure to check out our creations:
Investor Central | Investor Central Spanish | Investor Central German | Smart Living | Epochs & Echoes | Puzzling Mysteries | Hindutva | Elite Dev | Java Elite Dev | Golang Elite Dev | Python Elite Dev | JS Elite Dev | JS Schools
We are on Medium
Tech Koala Insights | Epochs & Echoes World | Investor Central Medium | Puzzling Mysteries Medium | Science & Epochs Medium | Modern Hindutva
Top comments (0)