Why HYQP?

Quick Start

Run the broker, then connect a subscriber and a publisher. Three terminals, five minutes.

1 · Run the broker

With Docker (no JDK needed):

# Pull and run the broker (TCP 4444, state persisted in a volume)
docker pull 78549/hyqp:v2.0
docker run -d --name hyqp -p 4444:4444 -v hyqp-data:/app/data 78549/hyqp:v2.0

Or from source (Java 21):

javac -d target/classes $(find src/main/java -name "*.java")
java -cp target/classes com.hyqp.broker.BrokerMain   # default port 4444

2 · Subscribe (only hot readings)

java -cp target/classes com.hyqp.broker.client.Subscriber dash sensors/temp '{"value":{"$gt":30}}'

3 · Publish

java -cp target/classes com.hyqp.broker.client.Publisher sensor-1
> CREATE sensors/temp {"type":"object","required":["value"],
    "properties":{"value":{"type":"number","minimum":-50,"maximum":100}}}
> PUBLISH sensors/temp {"value":22}    # not delivered (22 <= 30)
> PUBLISH sensors/temp {"value":75}    # delivered to dash
> PUBLISH sensors/temp {"value":200}   # ERROR: exceeds schema maximum 100

Protocol Reference

Every frame is a single UTF-8 line terminated by \n, over a long-lived TCP connection.

Client → Broker

Topic matching

Payload predicate operators

Content types v2.0

Tutorials

Short, focused walkthroughs of each HYQP capability.

Regex topic routing

Subscribe to every numbered production line at once, ignoring non-numeric zones:

SUBSCRIBE factory/regex('line-[0-9]+')/temp
# matches factory/line-1/temp, factory/line-42/temp -- NOT factory/zone-A/temp

One command to many topics (symmetric publish)

PUBLISH factory/regex('line-[0-9]+')/control {"cmd":"recalibrate"}
# the broker expands the pattern and delivers to every matching concrete topic

Schema governance

CREATE sensors/temp {"type":"object","required":["value"],
  "properties":{"value":{"type":"number","minimum":-50,"maximum":100}}}
# a later PUBLISH of {"value":150} is rejected before any subscriber sees it

Dynamic filter (no reconnect)

SUBSCRIBE sensors/temp FILTER {"value":{"$gt":50}}   # set
SUBSCRIBE sensors/temp FILTER {"value":{"$gt":90}}   # tighten, same connection
SUBSCRIBE sensors/temp                              # remove filter

Governance & discovery v2.0

OWNER sensors/temp          # -> OK Owner of sensors/temp: admin
PUBLISHERS sensors/temp     # -> OK Publishers of sensors/temp: sensor-1
SUBSCRIBERS sensors/temp    # -> OK Subscribers of sensors/temp: dash
DESCRIBE sensors/temp       # -> OK Topic ... | owner=admin | schema=yes | ...
LIST sensors/               # -> OK Topics: sensors/humidity, sensors/temp

Multi-format payloads v2.0

PUBLISH TYPE xml   sensors/cfg     <cfg rate="5s"/>
PUBLISH RETAIN TYPE octet line-1/firmware  iVBORw0KGgo...   # base64 bytes
PUBLISH           sensors/temp    {"value":22.5}              # json (default)

Tutorial 1 — Topic model & matching, in depth

A topic is a slash-separated path. A filter (used in SUBSCRIBE and in wildcard PUBLISH) may mix four kinds of segment. The table shows what each filter matches against a set of concrete topics.

Tutorial 2 — The ten payload operators, by example

A predicate is a JSON object: each field maps to one or more operators; all conditions are ANDed. The broker evaluates it before delivery, so non-matching messages never reach the subscriber. Each row shows a filter, a payload, and whether it is delivered.

Combine conditions across fields (logical AND). There is no explicit OR — express it as two subscriptions on the same topic:

SUBSCRIBE sensors/temp FILTER {"value":{"$gt":20,"$lt":80},"unit":{"$eq":"celsius"}}
# OR (value>90) OR (value<0): two subscriptions
SUBSCRIBE sensors/temp FILTER {"value":{"$gt":90}}
SUBSCRIBE sensors/temp FILTER {"value":{"$lt":0}}

Tutorial 3 — Schema governance, every constraint

A schema is attached at CREATE time and enforced on every PUBLISH, before fan-out. Supported constraints: type, required, nested properties, minimum/maximum, minLength/maxLength, enum, and items.

CREATE sensors/temp {"type":"object","required":["value"],
  "properties":{
    "value":{"type":"number","minimum":-50,"maximum":100},
    "unit" :{"type":"string","enum":["celsius","fahrenheit"]},
    "tags" :{"type":"array","items":{"type":"string","minLength":1}}
  }}

# On rejection the publisher gets a descriptive error:
ERROR Schema validation failed: $.value: value 150.0 exceeds maximum 100.0

Tutorial 4 — Quality of Service flows

QoS 0 — at most once (fire-and-forget)

PUBLISH sensors/temp {"value":22}
# subscriber gets:  MESSAGE sensors/temp {"value":22}

QoS 1 — at least once (PUBACK)

PUBLISH QOS1 sensors/temp {"value":22}
# broker -> subscriber:  MESSAGE 7 QOS1 sensors/temp {"value":22}
# subscriber -> broker:  PUBACK 7
# 10s retransmit window, up to 3 retries if no PUBACK

QoS 2 — exactly once (4-step handshake)

PUBLISH QOS2 sensors/temp {"value":22}
# broker -> sub:  MESSAGE 8 QOS2 sensors/temp {"value":22}
# sub   -> broker: PUBREC 8
# broker -> sub:  PUBREL 8
# sub   -> broker: PUBCOMP 8   (delivered exactly once)

Tutorial 5 — Retained messages

PUBLISH RETAIN sensors/temp {"value":22}     # stored as last-known value
# any LATER subscriber receives it immediately on subscribe:
SUBSCRIBE sensors/temp                       # -> MESSAGE sensors/temp {"value":22}
PUBLISH RETAIN sensors/temp                   # empty body clears the retained value

Tutorial 6 — Persistent sessions

CONNECT edge-1 PERSISTENT
SUBSCRIBE factory/line-1/temp FILTER {"celsius":{"$gt":50}}
DISCONNECT
# ...later, the same client reconnects:
CONNECT edge-1 PERSISTENT
# -> OK Connected as edge-1 (session restored)  -- subscriptions & filters are back

Tutorial 7 — Governance & discovery, every verb v2.0

CREATE sensors/temp {...schema...}    # caller becomes the owner
OWNER sensors/temp                    # -> OK Owner of sensors/temp: admin
PUBLISHERS sensors/temp               # -> OK Publishers of sensors/temp: sensor-1, sensor-2
SUBSCRIBERS sensors/temp              # -> OK Subscribers of sensors/temp: dash, alarm
DESCRIBE sensors/temp                 # -> OK Topic sensors/temp | owner=admin | created=... | schema=yes | contentType=json | publishers=2 | subscribers=2
LIST sensors/                         # -> OK Topics: sensors/humidity, sensors/temp

Tutorial 8 — Multi-format payloads, round-trip v2.0

# Delivery echoes the type so the subscriber knows how to decode:
MESSAGE TYPE octet line-1/firmware iVBORw0KGgo...
# json deliveries omit the tag, so plain subscribers are unaffected:
MESSAGE sensors/temp {"value":22.5}

Tutorial 9 — Responses & error handling

Python SDK

Zero-dependency client for dashboards, scripts and gateways. Runs on CPython 3.8+ and Raspberry Pi.

pip install -e sdk/python        # or set PYTHONPATH=sdk/python

Publisher

from hyqp import Publisher

with Publisher(host="localhost", port=4444, client_id="sensor-1") as pub:
    pub.create_topic("sensors/temp", schema={
        "type":"object", "required":["value"],
        "properties":{"value":{"type":"number","minimum":-50,"maximum":100}}})
    pub.publish("sensors/temp", {"value": 23.5}, qos=1)

Subscriber with a value filter

from hyqp import Subscriber

def on_msg(topic, payload, qos, mid):
    print(topic, payload)      # dict for JSON, str for text/xml, bytes for octet/pdf

sub = Subscriber(host="localhost", port=4444, client_id="dash", on_message=on_msg)
sub.connect()
sub.subscribe("sensors/+", filter={"value":{"$gt":30}})
sub.loop_forever()

Multi-format & discovery v2.0

pub.publish_text("line-1/note", "calibrated 09:00")
pub.publish_xml("line-1/cfg", "<cfg ver='4.2.1'/>")
pub.publish_bytes("line-1/firmware", open("fw.bin","rb").read())   # base64 on the wire

pub.owner("sensors/temp")        # -> 'sensor-1'
pub.subscribers("sensors/temp")  # -> ['dash']
pub.describe("sensors/temp")     # -> {'owner':'sensor-1','schema':'yes',...}
pub.list_topics("sensors/")      # -> ['sensors/temp', ...]

API reference (facades)

Scenario A — Continuous sensor publisher

import time, random
from hyqp import Publisher

pub = Publisher(host="localhost", port=4444, client_id="sensor-01")
pub.connect()
try:
    pub.create_topic("sensors/temp", schema={
        "type":"object","required":["value"],
        "properties":{"value":{"type":"number","minimum":-50,"maximum":150}}})
except Exception:
    pass   # topic already exists

while True:
    pub.publish("sensors/temp", {"value": round(random.uniform(15,45),1)}, qos=1)
    time.sleep(2)

Scenario B — Per-topic callbacks + default handler

from hyqp import Subscriber

def on_temp(topic, payload, qos, mid): print("temp", payload["value"])
def on_any(topic, payload, qos, mid):  print("other", topic, payload)

sub = Subscriber(client_id="dash", on_message=on_any)   # default handler
sub.connect()
sub.subscribe("sensors/temp", callback=on_temp)            # per-topic handler
sub.subscribe("sensors/humidity")                        # falls back to on_any
sub.loop_forever()

Scenario C — Dynamic filter during an incident

sub.subscribe("sensors/temp")                              # see everything
sub.subscribe("sensors/temp", filter={"value":{"$gt":80}})  # incident: only hot
sub.subscribe("sensors/temp")                              # back to normal (filter removed)
# the connection and any QoS state are preserved across all three

Scenario D — Persistent session + lifecycle callbacks

def on_connect(client, resp):    print("connected:", resp)
def on_disconnect(client, why):  print("lost:", why)

sub = Subscriber(client_id="edge-7", persistent=True,
                 on_message=lambda t,p,q,m: print(t,p),
                 on_connect=on_connect, on_disconnect=on_disconnect)
sub.connect()                       # "(session restored)" on reconnect
sub.subscribe("factory/regex('line-[0-9]+')/temp", filter={"celsius":{"$gt":50}})
sub.loop_forever()

Scenario E — Receiving multi-format payloads

def on_msg(topic, payload, qos, mid):
    # payload type reflects the content type automatically:
    if isinstance(payload, dict):    handle_json(payload)                # json
    elif isinstance(payload, bytes): open("fw.bin","wb").write(payload) # octet/pdf
    else:                            handle_text(payload)                # text/xml (str)

sub = Subscriber(client_id="fw-watch", on_message=on_msg); sub.connect()
sub.subscribe("line-1/#")

Scenario F — Error handling

from hyqp import Publisher, HYQPError
try:
    pub.publish("sensors/temp", {"value": 999})   # violates schema maximum
except HYQPError as e:
    print("rejected:", e)   # ERROR Schema validation failed: $.value: ...

Arduino / ESP32 SDK

A C++ client for ESP32/ESP8266 endpoints — zero dependencies beyond the Arduino WiFi stack.

Copy sdk/arduino/HYQP into your Arduino libraries/ folder (or zip-import it), then:

Publisher sketch

#include <WiFi.h>
#include <HYQP.h>

HYQPClient client("192.168.1.10", 4444, "esp32-1");

void setup(){
  WiFi.begin("ssid", "pass");
  while(WiFi.status()!=WL_CONNECTED) delay(300);
  client.connect();
  client.createTopic("sensors/temp", "{\"type\":\"object\",\"required\":[\"value\"]}");
  client.publish("sensors/temp", "{\"value\":23.5}", 1);   // QoS 1
}
void loop(){ client.loop(); delay(10); }

Subscriber with callback & filter

void onMsg(const char* topic, const char* payload, uint8_t qos, int id){
  Serial.printf("%s -> %s\n", topic, payload);
}
// in setup():
client.onMessage(onMsg);
client.subscribe("sensors/temp", "{\"value\":{\"$gt\":30}}");

Multi-format & discovery v2.0

uint8_t blob[] = {0x89,0x50,0x4E,0x47};
client.publishBytes("line-1/firmware", blob, sizeof(blob));  // base64
client.publishXml("line-1/cfg", "<cfg/>");
String owner = client.owner("sensors/temp");
String meta  = client.describe("sensors/temp");

API reference

Scenario A — Robust publisher with WiFi auto-reconnect

#include <WiFi.h>
#include <HYQP.h>

HYQPClient client("192.168.1.10", 4444, "esp32-line1");

void ensureConnected(){
  if(WiFi.status()!=WL_CONNECTED){ WiFi.reconnect(); return; }
  if(!client.isConnected()) client.connect(true);   // persistent session
}

void setup(){
  Serial.begin(115200);
  WiFi.begin("ssid", "pass");
  while(WiFi.status()!=WL_CONNECTED) delay(300);
  client.connect(true);
  client.createTopic("factory/line-1/temp",
    "{\"type\":\"object\",\"required\":[\"value\"],"
    "\"properties\":{\"value\":{\"type\":\"number\",\"minimum\":-50,\"maximum\":150}}}");
}

unsigned long last=0;
void loop(){
  ensureConnected();
  client.loop();
  if(millis()-last > 2000){
    last=millis();
    char buf[48]; snprintf(buf,sizeof(buf), "{\"value\":%.1f}", readTemp());
    client.publish("factory/line-1/temp", buf, 1, true);  // QoS1 + RETAIN
  }
}

Scenario B — Subscriber with a value filter

void onTemp(const char* topic, const char* payload, uint8_t qos, int id){
  Serial.printf("[%s] %s\n", topic, payload);
}
// in setup(), after connect():
client.subscribe("factory/regex('line-[0-9]+')/temp",
                 "{\"value\":{\"$gt\":50}}", onTemp);

Scenario C — Sending and decoding binary

// send a small binary blob (base64 on the wire)
uint8_t img[] = {0x89,0x50,0x4E,0x47,0x0D,0x0A};
client.publishBytes("line-1/snapshot", img, sizeof(img));

// decode an incoming base64 (octet) payload back to bytes
uint8_t out[256];
size_t n = HYQPClient::base64Decode(payload, out, sizeof(out));

Scenario D — Runtime discovery from the device

Serial.println(client.owner("factory/line-1/temp"));        // "plant-ops"
Serial.println(client.subscribers("factory/line-1/temp"));  // "dash, alarm"
Serial.println(client.describe("factory/line-1/temp"));     // full metadata line
Serial.println(client.listTopics("factory/"));             // comma-separated topics

IoT Wiring

From a sensor pin to a governed topic — on ESP32 and on Raspberry Pi.

ESP32 + DHT22 (temperature/humidity)

// read sensor, then publish a governed JSON reading
float t = dht.readTemperature();
char buf[48]; snprintf(buf,sizeof(buf), "{\"value\":%.1f,\"unit\":\"celsius\"}", t);
client.publish("factory/line-1/temp", buf, 1, false);   // QoS 1

Raspberry Pi gateway

1. Flash Raspberry Pi OS and enable SSH.
2. Copy the SDK: scp -r sdk/python pi@raspberrypi.local:~/hyqp-sdk
3. SSH in and run a subscriber that bridges to a dashboard:

   PYTHONPATH=~/hyqp-sdk python3 sub_example.py

4. Auto-start at boot with a systemd unit (see the Python SDK guide PDF for the unit file).

Building Complete Scenarios

Wire publishers and subscribers into an end-to-end flow. Example: a smart-factory line.

1 · Owner creates governed topics

from hyqp import Publisher
ops = Publisher(client_id="plant-ops"); ops.connect()
for n in range(1,4):
    ops.create_topic(f"factory/line-{n}/temp", schema={
        "type":"object","required":["celsius"],
        "properties":{"celsius":{"type":"number","minimum":-50,"maximum":200}}})

2 · Dashboard subscribes to all lines, only hot readings

dash = Subscriber(client_id="dashboard", on_message=on_hot); dash.connect()
dash.subscribe("factory/regex('line-[0-9]+')/temp", filter={"celsius":{"$gt":50}})

3 · Lines publish; one command reaches all of them

line1.publish("factory/line-1/temp", {"celsius":75})         # -> dashboard
ops.publish("factory/regex('line-[0-9]+')/control", {"cmd":"recalibrate"})

4 · Operator audits the topic graph at runtime

ops.subscribers("factory/line-1/temp")   # who is listening?
ops.describe("factory/line-1/temp")      # owner, schema, content type, counts

Docker

The broker is published on Docker Hub as 78549/hyqp.

Pull & run

docker pull 78549/hyqp:v2.0
docker run -d --name hyqp -p 4444:4444 -v hyqp-data:/app/data 78549/hyqp:v2.0

docker-compose (broker + MQTT/AMQP for benchmarks)

docker compose up -d        # starts hyqp + mosquitto + rabbitmq

Image facts

• Two-stage build: JDK 21 to compile → slim JRE 21 to run (virtual threads enabled).
• Runs as a non-root user; no third-party layers beyond the JRE.
• Mount a volume at /app/data to persist retained messages & sessions across restarts.

Download & Resources

At a glance