Understanding Message Queue Protocols
Message queue protocols define the rules, formats, and semantics for exchanging messages between producers and consumers through a message broker or directly in a peer-to-peer fashion. These protocols sit at the application layer and govern how messages are structured, addressed, routed, acknowledged, and persisted. The most widely adopted protocols include AMQP (Advanced Message Queuing Protocol), MQTT (Message Queue Telemetry Transport), STOMP (Simple Text Oriented Messaging Protocol), and the Kafka wire protocol. Each protocol was designed with specific use cases in mindâfrom enterprise-grade transactional messaging to lightweight IoT telemetry.
At their core, all message queue protocols share a common goal: decoupling applications so they can communicate asynchronously. This decoupling allows services to operate independently, absorb traffic spikes, and recover from failures without cascading effects. Understanding these protocols at a practical level means knowing not just their specifications, but how to implement clients, handle edge cases, and optimize for production environments.
What Message Queue Protocols Actually Define
đ Deploy your AI agent in 10 minutes
Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.
Try it free →A message queue protocol is fundamentally a contract between the client and the broker (or between peers). This contract spans multiple layers of communication:
Frame Format and Wire-Level Encoding
Every protocol defines precisely how messages are serialized onto the wire. AMQP uses a binary framing system with distinct frame typesâmethod frames, content header frames, and body framesâeach carrying specific segments of a complete message transfer. MQTT uses a compact binary header with a fixed header, variable header, and payload, optimized for minimal bandwidth consumption. STOMP uses plain text frames with a simple command/header/body structure delimited by null characters, making it trivially inspectable with tools like netcat or telnet.
Here's an example of how AMQP frames a message at the wire level:
// AMQP frame structure (simplified)
// Each frame: type (1 byte) + channel (2 bytes) + size (4 bytes) + payload
typedef struct {
uint8_t frame_type; // 1 = method, 2 = header, 3 = body, 4 = heartbeat
uint16_t channel; // channel number, 0 for global frames
uint32_t payload_size; // size of payload in bytes
uint8_t payload[]; // variable-length payload
} amqp_frame_t;
// A complete message delivery across the wire:
// Frame 1: Basic.Deliver method frame (type=1, channel=1)
// Frame 2: Content header frame with properties (type=2, channel=1)
// Frame 3: Body frame with actual payload (type=3, channel=1)
Connection Lifecycle and Heartbeating
Protocols specify how connections are established, maintained, and torn down. AMQP has a rich connection negotiation phase where clients and brokers agree on channel limits, frame sizes, and heartbeat intervals. MQTT's CONNECT packet carries a keepalive timer that both sides use to detect half-open connections. Understanding these lifecycle details is crucial because improper handling leads to resource leaksâorphaned connections, undelivered messages, and eventually broker overload.
Message Delivery Guarantees and Acknowledgments
Perhaps the most critical aspect of any message queue protocol is its delivery guarantee model. This is typically expressed through acknowledgment semantics:
- At-most-once delivery: Fire and forget. The message is sent once with no confirmation. If the network drops the packet, the message is lost.
- At-least-once delivery: The broker persists the message and requires explicit acknowledgment. If the acknowledgment is lost, the message may be redelivered, causing duplicates.
- Exactly-once delivery: The protocol guarantees no loss and no duplication through idempotent operations and transactional acknowledgments. This is the hardest to implement correctly.
The distinction between these models manifests directly in protocol operations. In AMQP, the basic.ack method with the multiple flag set to true acknowledges all messages up to a delivery tag, enabling batch acknowledgment. MQTT's QoS levels (0, 1, 2) map directly to these guarantees, with QoS 2 implementing a four-step handshake to ensure exactly-once delivery:
// MQTT QoS 2 exactly-once delivery handshake
// Step 1: Publisher sends PUBLISH with QoS=2, packetId=X
// Step 2: Broker stores message, sends PUBREC with packetId=X
// Step 3: Publisher sends PUBREL with packetId=X
// Step 4: Broker delivers to subscribers, sends PUBCOMP with packetId=X
// The message is only deleted from broker storage after PUBCOMP
// If the publisher disconnects after step 2, it MUST resend PUBREL on reconnect
Why Implementing Protocol Clients Matters
Most developers interact with message queues through high-level librariesâthe amqp gem, pika for Python, or the MQTT.js library. These libraries abstract away the protocol internals. However, understanding the protocol implementation matters deeply for several reasons:
Debugging Production Issues
When messages go missing, consumers stall, or brokers start rejecting connections, the symptoms often manifest at the protocol level. A consumer that stops sending heartbeats will be disconnected by the broker. A publisher sending messages to a non-existent exchange gets a basic.return frame backâbut only if the mandatory flag was set. Without protocol knowledge, these failures appear as mysterious timeouts or silent data loss.
Building Custom Protocol Adapters
In polyglot architectures, you often need protocol bridgesâtranslating MQTT from IoT devices into AMQP for backend services, or converting STOMP frames from a web frontend into Kafka protocol messages for stream processing. These bridges require intimate protocol knowledge to preserve delivery semantics across translation boundaries.
Performance Optimization at Scale
At high throughput, the overhead of protocol framing becomes significant. Understanding how frame sizes, batching, and multiplexing work allows you to tune client behavior. For instance, AMQP channels allow multiplexing hundreds of logical streams over a single TCP connection, avoiding the connection overhead per stream. MQTT's topic-based filtering happens at the broker levelâunderstanding wildcard matching rules prevents accidentally subscribing to thousands of irrelevant topics.
Edge Cases and Resilience Testing
Protocol specifications contain nuanced edge cases: what happens when a channel exceeds its prefetch limit? How does the broker handle a consumer that acknowledges a delivery tag that was never delivered? Implementing a protocol client forces you to handle these edge cases explicitly, producing systems that survive chaos engineering experiments.
Implementing an AMQP Client: A Practical Walkthrough
Let's walk through implementing a minimal but complete AMQP 0-9-1 client in Python that can connect, declare exchanges and queues, bind them, publish messages, and consume with acknowledgments. This will illuminate the protocol's inner workings.
Connection Establishment and Tuning
The first step is the TCP connection and the protocol header exchange. AMQP requires both sides to send a specific 8-byte header to confirm protocol version. After the header, the connection tuning phase negotiates operational parameters:
import socket
import struct
import threading
from enum import IntEnum
from dataclasses import dataclass
from typing import Optional, Callable
# AMQP 0-9-1 constants
AMQP_PROTOCOL_HEADER = b'\x41\x4d\x51\x50\x00\x00\x09\x01' # "AMQP" + 0, 0, 9, 1
FRAME_METHOD = 1
FRAME_HEADER = 2
FRAME_BODY = 3
FRAME_HEARTBEAT = 8
class ClassId(IntEnum):
CONNECTION = 10
CHANNEL = 20
EXCHANGE = 40
QUEUE = 50
BASIC = 60
class ConnectionMethod(IntEnum):
START = 10
START_OK = 11
TUNE = 30
TUNE_OK = 31
OPEN = 40
OPEN_OK = 41
CLOSE = 50
CLOSE_OK = 51
class BasicMethod(IntEnum):
PUBLISH = 40
CONSUME = 20
CONSUME_OK = 21
DELIVER = 60
ACK = 80
QOS = 10
@dataclass
class Frame:
frame_type: int
channel: int
payload: bytes
class AmqpClient:
def __init__(self, host: str, port: int, username: str, password: str):
self.host = host
self.port = port
self.username = username
self.password = password
self.sock: Optional[socket.socket] = None
self.channel_id = 1
self.frame_size_limit = 131072 # default 128KB
self.heartbeat_interval = 60
self.delivery_tag_counter = 0
self.consumers: dict = {}
self.lock = threading.Lock()
def connect(self):
"""Establish TCP connection and negotiate AMQP protocol"""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.settimeout(30)
self.sock.connect((self.host, self.port))
# Send protocol header
self.sock.sendall(AMQP_PROTOCOL_HEADER)
# Read server's protocol header (expect same version)
server_header = self._recv_exactly(8)
if server_header != AMQP_PROTOCOL_HEADER:
raise RuntimeError(f"Protocol mismatch: {server_header}")
# Read Connection.Start frame from broker
frame = self._read_frame()
start_payload = frame.payload
# Parse the Start method payload
# Format: version_major(1), version_minor(1),
# properties(shortstr), mechanisms(longstr), locales(longstr)
offset = 2 # skip version_major, version_minor
server_props_len = struct.unpack_from('>B', start_payload, offset)[0]
offset += 1 + server_props_len
mechanisms_len = struct.unpack_from('>I', start_payload, offset)[0]
offset += 4
mechanisms = start_payload[offset:offset + mechanisms_len].decode()
offset += mechanisms_len
locales_len = struct.unpack_from('>I', start_payload, offset)[0]
offset += 4
locales = start_payload[offset:offset + locales_len].decode()
print(f"Server mechanisms: {mechanisms}, locales: {locales}")
# Send Connection.Start-Ok with PLAIN authentication
# Format: client_props(shortstr), mechanism(shortstr), response(longstr), locale(shortstr)
auth_response = f"\x00{self.username}\x00{self.password}".encode()
client_props_field = self._encode_short_string('product=python-client')
mechanism_field = self._encode_short_string('PLAIN')
response_field = self._encode_long_string(auth_response.decode('latin-1'))
locale_field = self._encode_short_string('en_US')
start_ok_payload = client_props_field + mechanism_field + response_field + locale_field
self._write_method_frame(0, ClassId.CONNECTION, ConnectionMethod.START_OK, start_ok_payload)
# Read Connection.Tune
frame = self._read_frame()
tune_payload = frame.payload
# Parse: channel_max(2), frame_max(4), heartbeat(2)
channel_max, frame_max, heartbeat = struct.unpack_from('>HIH', tune_payload, 0)
if frame_max > 0 and frame_max < self.frame_size_limit:
self.frame_size_limit = frame_max
if heartbeat > 0:
self.heartbeat_interval = heartbeat
print(f"Tuned: channels={channel_max}, frame_max={frame_max}, heartbeat={heartbeat}")
# Send Connection.Tune-Ok echoing parameters we accept
tune_ok_payload = struct.pack('>HIH', 0, self.frame_size_limit, self.heartbeat_interval)
self._write_method_frame(0, ClassId.CONNECTION, ConnectionMethod.TUNE_OK, tune_ok_payload)
# Send Connection.Open with virtual host '/'
vhost_field = self._encode_short_string('/')
open_payload = vhost_field + b'\x00' # reserved bytes
self._write_method_frame(0, ClassId.CONNECTION, ConnectionMethod.OPEN, open_payload)
# Read Connection.Open-Ok
frame = self._read_frame()
print("Connection established successfully")
# Start heartbeat thread
threading.Thread(target=self._heartbeat_loop, daemon=True).start()
def _encode_short_string(self, s: str) -> bytes:
"""Encode AMQP short string: 1 byte length + data"""
data = s.encode('utf-8')
return struct.pack('>B', len(data)) + data
def _encode_long_string(self, s: str) -> bytes:
"""Encode AMQP long string: 4 byte length + data"""
data = s.encode('utf-8')
return struct.pack('>I', len(data)) + data
def _write_method_frame(self, channel: int, class_id: ClassId, method_id: int, args: bytes):
"""Write a method frame: class_id(2) + method_id(2) + arguments"""
method_payload = struct.pack('>HH', class_id.value, method_id) + args
self._write_frame(channel, FRAME_METHOD, method_payload)
def _write_frame(self, channel: int, frame_type: int, payload: bytes):
"""Write a raw AMQP frame"""
header = struct.pack('>BHI', frame_type, channel, len(payload))
frame = header + payload + b'\xCE' # frame-end byte 0xCE
with self.lock:
self.sock.sendall(frame)
def _read_frame(self) -> Frame:
"""Read a complete AMQP frame"""
header = self._recv_exactly(7)
frame_type, channel, payload_size = struct.unpack('>BHI', header)
payload = self._recv_exactly(payload_size)
frame_end = self._recv_exactly(1) # should be 0xCE
if frame_end != b'\xCE':
raise RuntimeError("Invalid frame end marker")
return Frame(frame_type, channel, payload)
def _recv_exactly(self, n: int) -> bytes:
"""Receive exactly n bytes, handling partial reads"""
data = b''
while len(data) < n:
chunk = self.sock.recv(n - len(data))
if not chunk:
raise ConnectionError("Socket closed")
data += chunk
return data
def _heartbeat_loop(self):
"""Send heartbeat frames at negotiated interval"""
import time
while True:
time.sleep(self.heartbeat_interval)
try:
with self.lock:
self.sock.sendall(struct.pack('>BHI', FRAME_HEARTBEAT, 0, 0) + b'\xCE')
except Exception:
break
Channel Opening and Resource Declaration
With the connection established, we open a channelâthe primary unit of multiplexing in AMQP. Channels are lightweight and allow concurrent operations without head-of-line blocking:
def open_channel(self) -> int:
"""Open a new channel"""
channel_id = self.channel_id
self.channel_id += 1
# Send Channel.Open
# Format: out_of_band(shortstr, reserved)
open_payload = self._encode_short_string('') # no out-of-band channel
self._write_method_frame(channel_id, ClassId.CHANNEL, 20, open_payload)
# Read Channel.Open-Ok
frame = self._read_frame()
# Response carries a reserved longstr
print(f"Channel {channel_id} opened")
return channel_id
def declare_exchange(self, channel: int, name: str, exchange_type: str = 'direct',
durable: bool = False, auto_delete: bool = False):
"""Declare an exchange"""
# Exchange.Declare method
# Format: reserved(2), exchange(shortstr), type(shortstr),
# passive(bit), durable(bit), reserved(3 bits), auto_delete(bit),
# internal(bit), reserved(1 bit), arguments(field-table)
reserved = struct.pack('>H', 0)
exchange_field = self._encode_short_string(name)
type_field = self._encode_short_string(exchange_type)
bits = 0
if durable: bits |= 0x02 # bit 1 is durable
if auto_delete: bits |= 0x10 # bit 4 is auto_delete
bits_byte = struct.pack('>B', bits)
args_table = b'\x00\x00\x00\x00' # empty field table
payload = reserved + exchange_field + type_field + bits_byte + args_table
self._write_method_frame(channel, ClassId.EXCHANGE, 10, payload)
# Read Exchange.Declare-Ok
frame = self._read_frame()
print(f"Exchange '{name}' declared")
def declare_queue(self, channel: int, name: str, durable: bool = False,
exclusive: bool = False, auto_delete: bool = False) -> str:
"""Declare a queue. Returns the queue name (server-generated if name is empty)"""
# Queue.Declare method
# Format: reserved(2), queue(shortstr), bits(byte), arguments(field-table)
reserved = struct.pack('>H', 0)
queue_field = self._encode_short_string(name)
bits = 0
if durable: bits |= 0x02
if exclusive: bits |= 0x04
if auto_delete: bits |= 0x08
bits_byte = struct.pack('>B', bits)
args_table = b'\x00\x00\x00\x00'
payload = reserved + queue_field + bits_byte + args_table
self._write_method_frame(channel, ClassId.QUEUE, 10, payload)
# Read Queue.Declare-Ok
frame = self._read_frame()
# Parse: queue_name(longstr), message_count(4), consumer_count(4)
queue_name_len = struct.unpack_from('>I', frame.payload, 0)[0]
queue_name = frame.payload[4:4 + queue_name_len].decode()
print(f"Queue '{queue_name}' declared")
return queue_name
def bind_queue(self, channel: int, queue_name: str, exchange: str, routing_key: str):
"""Bind a queue to an exchange with a routing key"""
# Queue.Bind method
# Format: reserved(2), queue(shortstr), exchange(shortstr), routing_key(shortstr),
# reserved(1), arguments(field-table)
reserved = struct.pack('>H', 0)
queue_field = self._encode_short_string(queue_name)
exchange_field = self._encode_short_string(exchange)
rk_field = self._encode_short_string(routing_key)
no_args_bit = b'\x00'
args_table = b'\x00\x00\x00\x00'
payload = reserved + queue_field + exchange_field + rk_field + no_args_bit + args_table
self._write_method_frame(channel, ClassId.QUEUE, 20, payload)
# Read Queue.Bind-Ok
frame = self._read_frame()
print(f"Queue '{queue_name}' bound to exchange '{exchange}' with routing key '{routing_key}'")
Publishing Messages with Delivery Guarantees
Publishing in AMQP involves a method frame (Basic.Publish), a content header frame carrying properties like content type and delivery mode, and one or more body frames. The mandatory flag causes the broker to return the message if it cannot be routed to any queue, while persistent delivery mode ensures the message survives broker restarts:
def publish(self, channel: int, exchange: str, routing_key: str, body: bytes,
content_type: str = 'application/octet-stream',
persistent: bool = True, mandatory: bool = False):
"""Publish a message to an exchange"""
# Build Basic.Publish method frame
# Format: reserved(2), exchange(shortstr), routing_key(shortstr),
# mandatory(bit), immediate(bit), reserved(6 bits)
reserved = struct.pack('>H', 0)
exchange_field = self._encode_short_string(exchange)
rk_field = self._encode_short_string(routing_key)
bits = 0
if mandatory: bits |= 0x01
bits_byte = struct.pack('>B', bits)
publish_payload = reserved + exchange_field + rk_field + bits_byte
self._write_method_frame(channel, ClassId.BASIC, BasicMethod.PUBLISH, publish_payload)
# Build Content Header frame
# Format: class_id(2), weight(2), body_size(8), property_flags(2),
# properties (variable based on flags)
class_id = struct.pack('>H', ClassId.BASIC.value)
weight = struct.pack('>H', 0) # always 0 in AMQP 0-9-1
body_size = struct.pack('>Q', len(body))
# Property flags: bitmask indicating which properties are present
# We'll set: content-type(bit 3), delivery-mode(bit 5)
property_flags = 0x0000
property_flags |= 0x0080 # bit 3 (0-indexed from MSB: bit 15-3=12) content-type present
property_flags |= 0x0200 # bit 5 (15-5=10) delivery-mode present
properties = b''
# Content-type: shortstr
properties += self._encode_short_string(content_type)
# Delivery-mode: octet (1=non-persistent, 2=persistent)
properties += struct.pack('>B', 2 if persistent else 1)
content_header = class_id + weight + body_size + struct.pack('>H', property_flags) + properties
self._write_frame(channel, FRAME_HEADER, content_header)
# Write body frames, respecting frame_size_limit
max_body_per_frame = self.frame_size_limit - 8 # account for overhead
body_offset = 0
while body_offset < len(body):
chunk = body[body_offset:body_offset + max_body_per_frame]
self._write_frame(channel, FRAME_BODY, chunk)
body_offset += len(chunk)
print(f"Published {len(body)} bytes to exchange '{exchange}' with routing key '{routing_key}'")
Consuming Messages with Acknowledgments
Consumption requires setting a QoS prefetch limit to control flow, issuing Basic.Consume, and then processing incoming Basic.Deliver frames. Each delivery carries a delivery tag that must be acknowledged to remove the message from the queue:
def set_qos(self, channel: int, prefetch_count: int = 1):
"""Set Quality of Service prefetch limit"""
# Basic.Qos method
# Format: prefetch_size(4), prefetch_count(2), global(bit)
payload = struct.pack('>IHB', 0, prefetch_count, 0) # prefetch_size=0 means no byte limit
self._write_method_frame(channel, ClassId.BASIC, BasicMethod.QOS, payload)
# Read Basic.Qos-Ok
frame = self._read_frame()
print(f"QoS set: prefetch_count={prefetch_count}")
def consume(self, channel: int, queue_name: str, consumer_tag: str,
callback: Callable[[bytes, int], None]):
"""Start consuming from a queue. callback receives (body, delivery_tag)"""
# Basic.Consume method
# Format: reserved(2), queue(shortstr), consumer_tag(shortstr),
# reserved_bits(byte), arguments(field-table)
reserved = struct.pack('>H', 0)
queue_field = self._encode_short_string(queue_name)
tag_field = self._encode_short_string(consumer_tag)
bits = struct.pack('>B', 0) # no no-local, no no-ack, no exclusive
args_table = b'\x00\x00\x00\x00'
payload = reserved + queue_field + tag_field + bits + args_table
self._write_method_frame(channel, ClassId.BASIC, BasicMethod.CONSUME, payload)
# Read Basic.Consume-Ok
frame = self._read_frame()
# Parse: consumer_tag(longstr)
tag_len = struct.unpack_from('>I', frame.payload, 0)[0]
returned_tag = frame.payload[4:4 + tag_len].decode()
print(f"Consumer '{returned_tag}' registered on queue '{queue_name}'")
self.consumers[consumer_tag] = callback
# Start delivery processing loop in background
threading.Thread(target=lambda: self._delivery_loop(channel, consumer_tag), daemon=True).start()
def _delivery_loop(self, channel: int, consumer_tag: str):
"""Process incoming deliveries"""
callback = self.consumers.get(consumer_tag)
while callback is not None:
try:
frame = self._read_frame()
if frame.frame_type == FRAME_METHOD:
# Parse method: class_id(2), method_id(2)
class_id, method_id = struct.unpack_from('>HH', frame.payload, 0)
if class_id == ClassId.BASIC.value and method_id == BasicMethod.DELIVER.value:
# Basic.Deliver format:
# consumer_tag(shortstr), delivery_tag(8), redelivered(bit),
# exchange(shortstr), routing_key(shortstr)
offset = 2 + 2 # skip class_id, method_id
tag_len = frame.payload[offset]
offset += 1
delivered_consumer = frame.payload[offset:offset + tag_len].decode()
offset += tag_len
delivery_tag = struct.unpack_from('>Q', frame.payload, offset)[0]
offset += 8
redelivered_bit = frame.payload[offset] & 0x01
offset += 1
exch_len = frame.payload[offset]
offset += 1
exchange = frame.payload[offset:offset + exch_len].decode()
offset += exch_len
rk_len = frame.payload[offset]
offset += 1
routing_key = frame.payload[offset:offset + rk_len].decode()
# Read content header frame
header_frame = self._read_frame()
# Parse body_size from header: class_id(2), weight(2), body_size(8)
body_size = struct.unpack_from('>Q', header_frame.payload, 4)[0]
# Read body frames and assemble message
body = b''
while len(body) < body_size:
body_frame = self._read_frame()
if body_frame.frame_type == FRAME_BODY:
body += body_frame.payload
callback(body, delivery_tag)
elif frame.frame_type == FRAME_HEARTBEAT:
continue # heartbeat frames are silently ignored
except Exception as e:
print(f"Delivery loop error: {e}")
break
print(f"Delivery loop for consumer '{consumer_tag}' stopped")
def acknowledge(self, channel: int, delivery_tag: int, multiple: bool = False):
"""Acknowledge a delivery"""
# Basic.Ack method
# Format: delivery_tag(8), multiple(bit)
bits = 0x01 if multiple else 0x00
payload = struct.pack('>QB', delivery_tag, bits)
self._write_method_frame(channel, ClassId.BASIC, BasicMethod.ACK, payload)
print(f"Acknowledged delivery_tag={delivery_tag}, multiple={multiple}")
Putting It All Together: A Complete Producer-Consumer Example
def main():
client = AmqpClient('localhost', 5672, 'guest', 'guest')
client.connect()
channel = client.open_channel()
# Declare a durable direct exchange
client.declare_exchange(channel, 'orders', 'direct', durable=True)
# Declare a durable queue
queue_name = client.declare_queue(channel, 'order-processing', durable=True)
# Bind queue to exchange with routing key
client.bind_queue(channel, queue_name, 'orders', 'order.created')
# Set QoS prefetch to 10 messages
client.set_qos(channel, prefetch_count=10)
# Define message handler
def handle_order(body: bytes, delivery_tag: int):
print(f"Processing order: {body.decode()}")
# Simulate processing
import time
time.sleep(0.1)
# Acknowledge the message
client.acknowledge(channel, delivery_tag)
# Start consuming
client.consume(channel, queue_name, 'order-consumer', handle_order)
# Publish some test messages
import time
for i in range(5):
message = f'{{"orderId": {i}, "item": "widget-{i}"}}'.encode()
client.publish(channel, 'orders', 'order.created', message,
content_type='application/json', persistent=True)
time.sleep(0.5)
# Keep main thread alive
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("Shutting down...")
client.close()
if __name__ == '__main__':
main()
This implementation handles the core AMQP protocol flow: connection negotiation, channel multiplexing, resource declaration, publishing with persistence, and consuming with acknowledgment. Each frame is correctly structured with the proper byte-level encoding. In production, you'd add reconnection logic, error handling for broker-sent basic.return frames, consumer cancellation notifications, and graceful channel/connection teardown with Close and Close-Ok exchanges.
Implementing an MQTT Client for IoT Scenarios
MQTT is deliberately simpler than AMQP. Its fixed header is just two bytes (control packet type + flags, followed by remaining length encoded as a variable-length integer). This minimalism makes it ideal for constrained devices. Let's implement a minimal MQTT 3.1.1 client that handles CONNECT, PUBLISH (with QoS 0 and 1), SUBSCRIBE, and incoming PUBLISH delivery:
import socket
import struct
import threading
from enum import IntEnum
class MqttPacketType(IntEnum):
CONNECT = 1
CONNACK = 2
PUBLISH = 3
PUBACK = 4
PUBREC = 5
PUBREL = 6
PUBCOMP = 7
SUBSCRIBE = 8
SUBACK = 9
UNSUBSCRIBE = 10
UNSUBACK = 11
PINGREQ = 12
PINGRESP = 13
DISCONNECT = 14
def encode_remaining_length(length: int) -> bytes:
"""Encode MQTT remaining length as variable-length integer (up to 4 bytes)"""
encoded = b''
while True:
digit = length % 128
length //= 128
if length > 0:
digit |= 0x80 # continuation bit
encoded += struct.pack('>B', digit)
if length == 0:
break
return encoded
def decode_remaining_length(data: bytes, offset: int) -> tuple[int, int]:
"""Decode MQTT remaining length, returning (value, bytes_consumed)"""
multiplier = 1
value = 0
consumed = 0
while True:
byte = data[offset + consumed]
value += (byte & 0x7F) * multiplier
consumed += 1
multiplier *= 128
if not (byte & 0x80):
break
return value, consumed
class MqttClient:
def __init__(self, client_id: str, host: str = 'localhost', port: int = 1883,
keepalive: int = 60):
self.client_id = client_id
self.host = host
self.port = port
self.keepalive = keepalive
self.sock: socket.socket | None = None
self.packet_id_counter = 0
self.subscriptions: dict = {}
self.message_handlers: dict[str, callable] = {} # topic -> handler
self._running = False
def connect(self, username: str = '', password: str = ''):
"""Connect to MQTT broker"""
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.host, self.port))
# Build CONNECT packet
# Fixed header: Packet type(4 bits) + flags(4 bits) = 0x10 for CONNECT
# Variable header: Protocol name("MQTT"), protocol level(4),
# connect flags, keepalive
# Payload: client ID, optionally username and password
protocol_name = b'MQTT' # must be exactly "MQTT" for 3.1.1
protocol_level = struct.pack('>B', 4) # 3.1.1 = level 4
# Connect flags byte:
# bit 7: username flag
# bit 6: password flag
# bit 5: will retain
# bit 4-3: will QoS
# bit 2: will flag
# bit 1: clean session
# bit 0: reserved
flags = 0x02 # clean session
if username: flags |= 0x80 # username flag
if password: flags |= 0x40 # password flag
flags_byte = struct.pack('>B', flags)
keepal