Message Broker Instructions

Guidelines for building TCP-based pub/sub message brokers for inter-process communication between Python applications and AutoHotkey scripts.

Architecture Overview

┌─────────────────────────────────────────┐
│         Message Broker Server           │
│         (localhost:15234)               │
│  Topic Router | Message Queue | Health  │
└─────────────────────────────────────────┘
        ▲               ▲               ▲
        │ TCP           │ TCP           │ TCP
┌───────┴───────┐ ┌─────┴─────┐ ┌───────┴───────┐
│  Python App   │ │  AHK App  │ │  Qt App       │
│ BrokerClient  │ │ TCP Socket│ │QtBrokerClient │
└───────────────┘ └───────────┘ └───────────────┘

Protocol Specification

Transport

Protocol: TCP
Default Port: 15234
Host: localhost

Frame Format

┌─────────────────┬────────────────────────┐
│ Length (4 bytes)│ JSON Payload (UTF-8)   │
│ (big-endian)    │                        │
└─────────────────┴────────────────────────┘

JSON-RPC 2.0 Messages

Request:

{"jsonrpc": "2.0", "method": "subscribe", "params": {"patterns": ["jobs.*"]}, "id": 1}

Response:

{"jsonrpc": "2.0", "result": {"subscribed": ["jobs.*"]}, "id": 1}

Notification (no response):

{"jsonrpc": "2.0", "method": "message", "params": {"topic": "jobs.activated", "payload": {...}, "sender": "app1"}}

Methods

Method	Description	Params
`connect`	Initialize connection	`client_id`, `client_type`, `version`
`disconnect`	Clean disconnect	-
`subscribe`	Subscribe to patterns	`patterns[]`
`unsubscribe`	Unsubscribe	`patterns[]`
`publish`	Publish message	`topic`, `payload`, `qos`, `ttl`
`ping`	Heartbeat	-
`status`	Get broker status	-

Broker Server

Configuration

from dataclasses import dataclass

@dataclass
class BrokerConfig:
    host: str = "localhost"
    port: int = 15234
    max_connections: int = 100
    heartbeat_interval: float = 5.0
    heartbeat_timeout: float = 30.0
    max_message_size: int = 1024 * 1024  # 1 MB
    max_queue_size: int = 10000

Basic Server

import asyncio
from message_broker import MessageBroker, BrokerConfig

async def main():
    config = BrokerConfig(host="localhost", port=15234)
    broker = MessageBroker(config)
    await broker.start()
    
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await broker.stop()

asyncio.run(main())

System Topics

Topic	Payload	Description
`system.broker.ready`	`{host, port}`	Broker started
`system.broker.shutdown`	`{reason}`	Broker stopping
`system.client.connected`	`{client_id}`	Client connected
`system.client.disconnected`	`{client_id}`	Client disconnected

Python Async Client

from message_broker import BrokerClient, Message

async def main():
    client = BrokerClient("my_app", host="localhost", port=15234)
    await client.connect()
    
    # Subscribe
    async def handler(msg: Message):
        print(f"{msg.topic}: {msg.payload}")
    
    await client.subscribe(["jobs.*", "scripts.+"], handler)
    
    # Publish
    await client.publish("jobs.activated", {"job_id": "123"})
    
    # Request/Response
    response = await client.request("scripts.status", {"id": "saw1"}, timeout=5.0)
    
    await client.disconnect()

Client Properties

client.client_id: str
client.is_connected: bool

await client.connect() -> bool
await client.disconnect()
await client.subscribe(patterns, callback) -> list[str]
await client.unsubscribe(patterns) -> list[str]
await client.publish(topic, payload, qos=0, ttl=0) -> str
await client.request(topic, payload, timeout=5.0) -> dict

Qt Integration (QtBrokerClient)

Thread-safe client with Qt signals for PySide6/PyQt applications.

from PySide6.QtWidgets import QMainWindow
from PySide6.QtCore import Slot
from message_broker.qt_client import QtBrokerClient

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        
        self._broker = QtBrokerClient("main_app", parent=self)
        
        # Connect signals
        self._broker.connected.connect(self._on_connected)
        self._broker.disconnected.connect(self._on_disconnected)
        self._broker.message_received.connect(self._on_message)
        self._broker.error_occurred.connect(self._on_error)
        
        self._broker.connect_to_broker()
    
    @Slot()
    def _on_connected(self):
        self._broker.subscribe(["jobs.*"])
    
    @Slot(str, dict, str, str)
    def _on_message(self, topic, payload, sender, msg_id):
        print(f"{topic} from {sender}: {payload}")
    
    def publish_event(self, job_id):
        self._broker.publish("jobs.activated", {"job_id": job_id})
    
    def closeEvent(self, event):
        self._broker.disconnect_from_broker()
        super().closeEvent(event)

QtBrokerClient Signals

Signal	Parameters	Description
`connected`	-	Connected
`disconnected`	-	Disconnected
`connection_failed`	`str`	Connection error
`message_received`	`str, dict, str, str`	topic, payload, sender, id
`subscribed`	`list[str]`	Patterns subscribed
`published`	`str`	Message ID
`error_occurred`	`str`	Error message

Topic Patterns

Naming Convention

<domain>.<entity>.<action>

Examples:

jobs.activated
jobs.12345.updated
scripts.saw1.started
system.broker.ready

Pattern Matching

Pattern	Matches	Description
`jobs.activated`	Exact match	Single topic
`jobs.*`	`jobs.activated`, `jobs.created`	Single level wildcard
`jobs.+`	`jobs.123`, `jobs.abc`	Single level (≥1 char)
`scripts.#`	`scripts.saw1.started`, `scripts.saw1.hotkey.pressed`	Multi-level wildcard
`scripts.+.started`	`scripts.saw1.started`	Middle wildcard

Reserved Topics

Pattern	Purpose
`system.*`	Broker internal events
`_internal.*`	Internal operations

Common Patterns

Job Activation Flow

# Publisher (main app)
await broker.publish("jobs.activated", {
    "job_id": job_id,
    "job_number": "ABC123",
    "timestamp": datetime.now().isoformat(),
})

# Subscriber (other apps)
async def on_message(msg):
    if msg.topic == "jobs.activated":
        job_id = msg.payload["job_id"]
        update_ui(job_id)

await client.subscribe(["jobs.*"], on_message)

Script Status Updates

# Runner publishes lifecycle
await broker.publish(f"scripts.{script_id}.started", {"pid": pid})
await broker.publish(f"scripts.{script_id}.stopped", {"exit_code": 0})

# Monitor subscribes
await client.subscribe(["scripts.+.started", "scripts.+.stopped"], handler)

Request/Response (RPC)

# Client requests
response = await client.request("scripts.status", {"script_id": "saw1"}, timeout=5.0)

# Service handles
async def handle_request(msg):
    script_id = msg.payload["script_id"]
    return {"running": script_id in running_scripts}

AHK Integration

Direct TCP (AHK v2)

class BrokerClient {
    __New(clientId, host := "localhost", port := 15234) {
        this.clientId := clientId
        this.host := host
        this.port := port
        this.requestId := 0
    }
    
    Connect() {
        this.socket := ComObject("MSWinsock.Winsock.1")
        this.socket.RemoteHost := this.host
        this.socket.RemotePort := this.port
        this.socket.Connect()
        
        ; Send connect request
        this.requestId++
        request := '{"jsonrpc":"2.0","method":"connect","params":{"client_id":"' . this.clientId . '"},"id":' . this.requestId . '}'
        this._Send(request)
    }
    
    Publish(topic, payload) {
        this.requestId++
        request := '{"jsonrpc":"2.0","method":"publish","params":{"topic":"' . topic . '","payload":' . payload . '},"id":' . this.requestId . '}'
        this._Send(request)
    }
    
    _Send(data) {
        ; Length prefix (4 bytes big-endian)
        len := StrLen(data)
        header := Chr(len >> 24 & 0xFF) . Chr(len >> 16 & 0xFF) . Chr(len >> 8 & 0xFF) . Chr(len & 0xFF)
        this.socket.SendData(header . data)
    }
}

Python Bridge (Recommended)

from message_broker import BrokerClient
from message_broker.ahk_bridge import AHKBridge

async def main():
    client = BrokerClient("ahk_bridge")
    await client.connect()
    
    bridge = AHKBridge(broker_client=client)
    await bridge.start()
    await bridge.run_script("scripts/hotkeys.ahk")

Error Codes

Code	Name	Description
-32700	PARSE_ERROR	Invalid JSON
-32600	INVALID_REQUEST	Not valid JSON-RPC
-32601	METHOD_NOT_FOUND	Unknown method
-32602	INVALID_PARAMS	Invalid parameters
-32603	INTERNAL_ERROR	Internal broker error
-32001	NOT_CONNECTED	Client not connected
-32002	TIMEOUT	Request timed out

Testing

Mock Client

from unittest.mock import AsyncMock, MagicMock

@pytest.fixture
def mock_broker():
    client = MagicMock()
    client.is_connected = True
    client.connect = AsyncMock(return_value=True)
    client.publish = AsyncMock(return_value="msg-123")
    client.subscribe = AsyncMock(return_value=["jobs.*"])
    return client

Integration Test

@pytest.fixture
async def broker():
    broker = MessageBroker()
    await broker.start()
    yield broker
    await broker.stop()

async def test_pub_sub(broker):
    client = BrokerClient("test")
    await client.connect()
    
    received = []
    await client.subscribe(["test.*"], lambda m: received.append(m))
    await client.publish("test.msg", {"data": "hello"})
    
    await asyncio.sleep(0.1)
    assert received[0].payload["data"] == "hello"

Troubleshooting

Connection Refused

Check broker is running
Verify port is correct
Check firewall settings

Messages Not Received

Verify topic matches subscription pattern
Check subscription was active before publish
Enable debug logging: logging.getLogger("message_broker").setLevel(logging.DEBUG)

Qt Client Freezing

Never use await in Qt main thread
Use QtBrokerClient which handles threading

Debug Tools

# Enable message logging
config = BrokerConfig(log_messages=True)

# Get broker status
status = await client.request("status", {})
print(f"Clients: {status['client_count']}")

# Subscribe to all for debugging
await client.subscribe(["#"], debug_handler)

Configuration Defaults

DEFAULT_HOST = "localhost"
DEFAULT_PORT = 15234
DEFAULT_HEARTBEAT_INTERVAL = 5.0
DEFAULT_HEARTBEAT_TIMEOUT = 30.0
DEFAULT_MAX_MESSAGE_SIZE = 1024 * 1024

message-broker