Testando recursos do MQTT 5: user properties, expiry e topic aliases

npm install mqtt@5
pip install "paho-mqtt>=2.0"

import mqtt from 'mqtt';
import assert from 'node:assert/strict';
import { randomUUID } from 'node:crypto';

const URL = 'mqtt://broker.connectmq.com:1883';
const opts = { protocolVersion: 5, username: process.env.MQTT_USER, password: process.env.MQTT_PASS };
const base = `testmqtt/mqtt5/${randomUUID().slice(0, 8)}`;
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

const nextMessage = (client, timeoutMs = 5000) =>
  new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('timeout')), timeoutMs);
    client.once('message', (topic, payload, packet) => {
      clearTimeout(timer);
      resolve({ topic, payload, packet });
    });
  });

const sub = await mqtt.connectAsync(URL, opts);
const pub = await mqtt.connectAsync(URL, opts);

// 1. User properties, content type, correlation data
await sub.subscribeAsync(`${base}/props`, { qos: 1 });
let incoming = nextMessage(sub);
await pub.publishAsync(`${base}/props`, '{"temp":21.5}', {
  qos: 1,
  properties: {
    userProperties: { source: 'ci', 'trace-id': 'abc123' },
    contentType: 'application/json',
    responseTopic: `${base}/reply`,
    correlationData: Buffer.from('req-42'),
  },
});
const { packet } = await incoming;
assert.equal(packet.properties.userProperties.source, 'ci');
assert.equal(packet.properties.userProperties['trace-id'], 'abc123');
assert.equal(packet.properties.contentType, 'application/json');
assert.equal(packet.properties.correlationData.toString(), 'req-42');
console.log('PASS user properties / content type / correlation data');

// 2. Retained message expiry
await pub.publishAsync(`${base}/expiry`, 'short-lived', {
  qos: 1, retain: true, properties: { messageExpiryInterval: 3 },
});
await sleep(5000);
const late = await mqtt.connectAsync(URL, opts);
incoming = nextMessage(late, 3000);
await late.subscribeAsync(`${base}/expiry`, { qos: 1 });
await assert.rejects(incoming, /timeout/, 'expired retained message was delivered');
console.log('PASS retained message expired');

// 3. Topic alias: register alias 1, then publish with an empty topic
const aliasClient = mqtt.connect(URL, opts);
const connack = await new Promise((resolve, reject) => {
  aliasClient.once('connect', resolve);
  aliasClient.once('error', reject);
});
const aliasMax = connack.properties?.topicAliasMaximum ?? 0;
if (aliasMax === 0) {
  console.log('SKIP topic alias: broker advertises TopicAliasMaximum 0');
} else {
  const long = `${base}/alias/site-01/building-a/floor-3/sensor-temperature`;
  await sub.subscribeAsync(long, { qos: 1 });
  incoming = nextMessage(sub);
  await aliasClient.publishAsync(long, 'first', { qos: 1, properties: { topicAlias: 1 } });
  assert.equal((await incoming).topic, long);
  incoming = nextMessage(sub);
  await aliasClient.publishAsync('', 'second', { qos: 1, properties: { topicAlias: 1 } });
  assert.equal((await incoming).topic, long, 'broker did not resolve the alias');
  console.log(`PASS topic alias (broker allows ${aliasMax})`);
}

// 4. No local: a client must not receive its own publish
await pub.subscribeAsync(`${base}/nolocal`, { qos: 1, nl: true });
incoming = nextMessage(pub, 2000);
await pub.publishAsync(`${base}/nolocal`, 'echo?', { qos: 1 });
await assert.rejects(incoming, /timeout/, 'received own message despite no-local');
console.log('PASS no local');

await Promise.all([sub, pub, late, aliasClient].map((c) => c.endAsync()));

import time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

HOST, PORT = "broker.connectmq.com", 1883
TOPIC = f"testmqtt/mqtt5/{uuid.uuid4().hex[:8]}/queued"
CLIENT_ID = f"mqtt5-session-{uuid.uuid4().hex[:8]}"
received = []

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code, "| session present:", flags.session_present)

def on_message(client, userdata, msg):
    expiry = getattr(msg.properties, "MessageExpiryInterval", None)
    user = getattr(msg.properties, "UserProperty", None)
    print(f"got {msg.payload.decode()!r}  expiry left: {expiry}  user properties: {user}")
    received.append(msg.payload.decode())

def session_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                    protocol=mqtt.MQTTv5)
    c.on_connect, c.on_message = on_connect, on_message
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 120   # keep the session 2 minutes after disconnect
    c.connect(HOST, PORT, clean_start=False, properties=props)
    c.loop_start()
    time.sleep(1)
    return c

# 1. Create the session and subscription, then go offline
s = session_client()
s.subscribe(TOPIC, qos=1)
time.sleep(1)
s.disconnect(); s.loop_stop()

# 2. Publish while the subscriber is offline
pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
pub.connect(HOST, PORT); pub.loop_start()
for i, expiry in enumerate((60, 2)):
    props = Properties(PacketTypes.PUBLISH)
    props.MessageExpiryInterval = expiry
    props.UserProperty = ("test", f"msg-{i}")
    pub.publish(TOPIC, f"msg-{i}", qos=1, properties=props).wait_for_publish()
pub.disconnect(); pub.loop_stop()

time.sleep(5)  # longer than the 2 s expiry

# 3. Reconnect: msg-0 arrives with less than 60 s left, msg-1 has expired
s = session_client()
time.sleep(3)
s.disconnect(); s.loop_stop()
assert received == ["msg-0"], received
print("PASS session kept, unexpired message delivered, expired message dropped")