IoT Communication Protocols: MQTT, CoAP, AMQP & WebSocket

# docker-compose.yml
services:
  mosquitto:
    image: eclipse-mosquitto:2
    container_name: mqtt-broker
    ports:
      - "1883:1883"   # MQTT
      - "9001:9001"   # WebSocket (for browser clients)
    volumes:
      - ./mosquitto/config:/mosquitto/config
      - ./mosquitto/data:/mosquitto/data
      - ./mosquitto/log:/mosquitto/log
    restart: unless-stopped

# mosquitto/config/mosquitto.conf
listener 1883
protocol mqtt
 
listener 9001
protocol websockets
 
# Authentication (disable for development only)
allow_anonymous true
 
# Persistence
persistence true
persistence_location /mosquitto/data/
 
# Logging
log_dest file /mosquitto/log/mosquitto.log
log_type all

# Start the broker
docker compose up -d
 
# Verify it's running
docker compose logs mosquitto
# Output: mosquitto version 2.x.x running

# Topic naming convention: {organization}/{location}/{device-type}/{device-id}/{data-type}
factory/building-a/sensors/temp-042/readings
factory/building-a/sensors/temp-042/status
factory/building-a/actuators/fan-01/commands
factory/building-b/sensors/humidity-003/readings

import mqtt from "mqtt";
 
const client = mqtt.connect("mqtt://localhost:1883");
 
client.on("connect", () => {
  // Single-level wildcard (+): matches exactly ONE level
  client.subscribe("factory/building-a/sensors/+/readings");
  // Matches: factory/building-a/sensors/temp-042/readings ✅
  // Matches: factory/building-a/sensors/humidity-003/readings ✅
  // No match: factory/building-b/sensors/temp-042/readings ❌
 
  // Multi-level wildcard (#): matches EVERYTHING below
  client.subscribe("factory/building-a/#");
  // Matches: factory/building-a/sensors/temp-042/readings ✅
  // Matches: factory/building-a/actuators/fan-01/commands ✅
  // Matches: factory/building-a/alerts ✅
  // No match: factory/building-b/anything ❌
});
 
client.on("message", (topic, message) => {
  const segments = topic.split("/");
  const building = segments[1];
  const deviceType = segments[2];
  const deviceId = segments[3];
 
  console.log(`[${building}/${deviceId}] ${message.toString()}`);
});

import mqtt from "mqtt";
 
const client = mqtt.connect("mqtt://localhost:1883", {
  clientId: "sensor-node-042",
  clean: false,         // Persistent session — broker remembers subscriptions
  reconnectPeriod: 5000,
});
 
client.on("connect", () => {
  // QoS 0: GPS coordinates every second — missing one is fine
  setInterval(() => {
    client.publish("fleet/truck-07/gps", JSON.stringify({
      lat: 37.7749 + Math.random() * 0.01,
      lng: -122.4194 + Math.random() * 0.01,
      speed: 55 + Math.random() * 10,
      ts: Date.now(),
    }), { qos: 0 });
  }, 1000);
 
  // QoS 1: Temperature reading every 30s — we want all of them
  setInterval(() => {
    client.publish("factory/zone-1/temp/s042", JSON.stringify({
      value: 22.5 + Math.random() * 5,
      unit: "celsius",
      ts: Date.now(),
    }), { qos: 1 });
  }, 30000);
 
  // QoS 2: Actuator command — must execute exactly once
  client.publish("factory/zone-1/actuators/valve-01/cmd", JSON.stringify({
    action: "open",
    percentage: 75,
    requestId: crypto.randomUUID(),
  }), { qos: 2 });
});

// Device publishes its status as a retained message
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
  firmware: "2.1.0",
  ip: "192.168.1.42",
  uptime: 86400,
}), {
  qos: 1,
  retain: true, // Broker stores this — new subscribers get it immediately
});
 
// A dashboard connecting 5 minutes later immediately gets the status
// No need to wait for the next status update

const client = mqtt.connect("mqtt://localhost:1883", {
  clientId: "sensor-042",
  // Last Will: published by broker if this client disconnects ungracefully
  will: {
    topic: "devices/sensor-042/status",
    payload: JSON.stringify({
      state: "offline",
      lastSeen: Date.now(),
    }),
    qos: 1,
    retain: true, // Overwrite the "online" retained message
  },
});
 
// On connect, publish "online" status
client.on("connect", () => {
  client.publish("devices/sensor-042/status", JSON.stringify({
    state: "online",
    lastSeen: Date.now(),
  }), { qos: 1, retain: true });
});
 
// If the device crashes or loses network:
// 1. TCP keepalive times out (default 60s)
// 2. Broker publishes the LWT message
// 3. Dashboard sees "offline" status
// 4. Alert service triggers notification

// MQTT v5.0 features in action
const client = mqtt.connect("mqtt://localhost:1883", {
  protocolVersion: 5,
  clientId: "sensor-042",
});
 
client.on("connect", () => {
  // Message expiry: alert expires after 5 minutes
  client.publish("alerts/high-temp", JSON.stringify({
    device: "sensor-042",
    temp: 45.2,
  }), {
    qos: 1,
    properties: {
      messageExpiryInterval: 300, // 5 minutes in seconds
    },
  });
 
  // Request/Response: ask a device for its config
  client.publish("devices/sensor-042/config/get", "", {
    qos: 1,
    properties: {
      responseTopic: "responses/sensor-042/config",
      correlationData: Buffer.from("req-001"),
    },
  });
 
  // Shared subscription: load-balance across 3 processing workers
  client.subscribe("$share/processors/telemetry/#");
  // Three workers subscribing to the same shared group
  // Each message goes to only ONE worker — automatic load balancing
});

import { createServer, request } from "coap";
 
// ── CoAP Server (runs on the sensor/device) ──
const server = createServer();
 
// Temperature resource
let currentTemp = 23.5;
 
server.on("request", (req, res) => {
  const path = req.url;
 
  if (path === "/temperature" && req.method === "GET") {
    res.setOption("Content-Format", "application/json");
    res.end(JSON.stringify({
      value: currentTemp,
      unit: "celsius",
      ts: Date.now(),
    }));
  }
 
  if (path === "/led" && req.method === "PUT") {
    const payload = JSON.parse(req.payload.toString());
    console.log(`LED set to: ${payload.state}`);
    res.code = "2.04"; // Changed
    res.end();
  }
 
  if (path === "/temperature" && req.method === "GET" && req.headers.Observe === 0) {
    // Observe: push updates when value changes
    res.setOption("Content-Format", "application/json");
 
    const interval = setInterval(() => {
      currentTemp += (Math.random() - 0.5) * 0.5;
      res.write(JSON.stringify({
        value: parseFloat(currentTemp.toFixed(1)),
        ts: Date.now(),
      }));
    }, 5000);
 
    res.on("finish", () => clearInterval(interval));
  }
});
 
server.listen(5683, () => {
  console.log("CoAP server listening on port 5683");
});

// ── CoAP Client (runs on the gateway or cloud) ──
import { request } from "coap";
 
// Simple GET request
const req = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/temperature",
  method: "GET",
  confirmable: true, // CON message — reliable
});
 
req.on("response", (res) => {
  const data = JSON.parse(res.payload.toString());
  console.log(`Temperature: ${data.value}°C`);
});
 
req.end();
 
// PUT command to control an actuator
const cmdReq = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/led",
  method: "PUT",
  confirmable: true,
});
 
cmdReq.write(JSON.stringify({ state: "on", brightness: 80 }));
cmdReq.end();
 
// Observe: subscribe to temperature changes
const obsReq = request({
  hostname: "sensor-042.local",
  port: 5683,
  pathname: "/temperature",
  method: "GET",
  observe: true,
});
 
obsReq.on("response", (res) => {
  res.on("data", (chunk: Buffer) => {
    const data = JSON.parse(chunk.toString());
    console.log(`[Observe] Temperature: ${data.value}°C`);
  });
});
 
obsReq.end();

// Block-wise transfer for large payloads (e.g., firmware metadata)
// The CoAP library handles block negotiation automatically
 
const req = request({
  hostname: "sensor-042.local",
  pathname: "/firmware/info",
  method: "GET",
  // Block2 option: request 256-byte blocks
});
 
req.on("response", (res) => {
  // Library reassembles blocks automatically
  const firmwareInfo = JSON.parse(res.payload.toString());
  console.log(`Firmware: ${firmwareInfo.version}, Size: ${firmwareInfo.size}`);
});
 
req.end();

# docker-compose.yml
services:
  rabbitmq:
    image: rabbitmq:3-management
    container_name: iot-rabbitmq
    ports:
      - "5672:5672"     # AMQP
      - "15672:15672"   # Management UI
      - "1883:1883"     # MQTT plugin (optional)
    environment:
      RABBITMQ_DEFAULT_USER: iot_admin
      RABBITMQ_DEFAULT_PASS: secure_password
    volumes:
      - ./rabbitmq/enabled_plugins:/etc/rabbitmq/enabled_plugins
      - rabbitmq_data:/var/lib/rabbitmq
    restart: unless-stopped
 
volumes:
  rabbitmq_data:

# rabbitmq/enabled_plugins
[rabbitmq_management, rabbitmq_mqtt, rabbitmq_web_stomp].

import amqplib from "amqplib";
 
// ── Producer: IoT Gateway sends telemetry ──
async function publishTelemetry() {
  const connection = await amqplib.connect("amqp://iot_admin:secure_password@localhost");
  const channel = await connection.createChannel();
 
  // Declare a topic exchange for IoT data
  const exchange = "iot.telemetry";
  await channel.assertExchange(exchange, "topic", { durable: true });
 
  // Publish sensor data with routing key
  const routingKey = "factory.zone1.temperature";
  const message = {
    deviceId: "sensor-042",
    value: 28.5,
    unit: "celsius",
    timestamp: new Date().toISOString(),
  };
 
  channel.publish(
    exchange,
    routingKey,
    Buffer.from(JSON.stringify(message)),
    {
      persistent: true,        // Survive broker restart
      contentType: "application/json",
      messageId: crypto.randomUUID(),
      timestamp: Date.now(),
      headers: {
        "x-device-type": "temperature-sensor",
        "x-firmware": "2.1.0",
      },
    }
  );
 
  console.log(`Published: ${routingKey} → ${message.value}°C`);
}

// ── Consumer: Temperature processing service ──
async function consumeTemperature() {
  const connection = await amqplib.connect("amqp://iot_admin:secure_password@localhost");
  const channel = await connection.createChannel();
 
  const exchange = "iot.telemetry";
  const queue = "temperature-processor";
 
  await channel.assertExchange(exchange, "topic", { durable: true });
  await channel.assertQueue(queue, {
    durable: true,
    deadLetterExchange: "iot.dlx", // Failed messages go here
    messageTtl: 3600000,           // 1 hour TTL
  });
 
  // Bind: only receive temperature messages from any zone
  await channel.bindQueue(queue, exchange, "factory.*.temperature");
 
  // Prefetch: process 10 messages at a time (flow control)
  await channel.prefetch(10);
 
  channel.consume(queue, (msg) => {
    if (!msg) return;
 
    try {
      const data = JSON.parse(msg.content.toString());
      console.log(`Processing: ${data.deviceId} → ${data.value}°C`);
 
      // Check for alerts
      if (data.value > 35) {
        // Publish alert to a different routing key
        channel.publish(
          exchange,
          "factory.alerts.temperature",
          Buffer.from(JSON.stringify({
            type: "highTemperature",
            device: data.deviceId,
            value: data.value,
            threshold: 35,
            timestamp: new Date().toISOString(),
          })),
          { persistent: true }
        );
      }
 
      channel.ack(msg); // Acknowledge successful processing
    } catch (error) {
      // Reject and don't requeue — send to dead letter queue
      channel.nack(msg, false, false);
    }
  });
 
  console.log("Temperature processor listening...");
}

async function setupDeadLetterQueue() {
  const connection = await amqplib.connect("amqp://localhost");
  const channel = await connection.createChannel();
 
  // Dead letter exchange
  await channel.assertExchange("iot.dlx", "topic", { durable: true });
  await channel.assertQueue("iot.failed-messages", { durable: true });
  await channel.bindQueue("iot.failed-messages", "iot.dlx", "#");
 
  // Main queue with dead letter routing
  await channel.assertQueue("temperature-processor", {
    durable: true,
    deadLetterExchange: "iot.dlx",
    deadLetterRoutingKey: "failed.temperature",
    messageTtl: 3600000, // Messages older than 1 hour → DLQ
    maxLength: 100000,   // Queue overflow → DLQ
  });
 
  // Monitor the dead letter queue
  channel.consume("iot.failed-messages", (msg) => {
    if (!msg) return;
    const data = JSON.parse(msg.content.toString());
    const reason = msg.properties.headers?.["x-death"]?.[0]?.reason;
    console.error(`Dead letter [${reason}]: ${JSON.stringify(data)}`);
    // Log to monitoring system, trigger investigation
    channel.ack(msg);
  });
}

import { WebSocketServer, WebSocket } from "ws";
import mqtt from "mqtt";
 
// Bridge MQTT telemetry to WebSocket clients (dashboards)
const wss = new WebSocketServer({ port: 8080 });
const mqttClient = mqtt.connect("mqtt://localhost:1883");
 
// Track which topics each WebSocket client is watching
const clientSubscriptions = new Map<WebSocket, Set<string>>();
 
mqttClient.on("connect", () => {
  mqttClient.subscribe("factory/#");
  console.log("MQTT bridge connected");
});
 
mqttClient.on("message", (topic, payload) => {
  const message = JSON.stringify({
    topic,
    data: JSON.parse(payload.toString()),
    receivedAt: new Date().toISOString(),
  });
 
  // Forward to interested WebSocket clients
  wss.clients.forEach((ws) => {
    if (ws.readyState !== WebSocket.OPEN) return;
 
    const subs = clientSubscriptions.get(ws);
    if (!subs) return;
 
    // Check if any subscription pattern matches this topic
    for (const pattern of subs) {
      if (topicMatches(pattern, topic)) {
        ws.send(message);
        break;
      }
    }
  });
});
 
wss.on("connection", (ws) => {
  clientSubscriptions.set(ws, new Set());
 
  ws.on("message", (raw) => {
    const msg = JSON.parse(raw.toString());
 
    if (msg.type === "subscribe") {
      clientSubscriptions.get(ws)?.add(msg.topic);
      console.log(`Client subscribed to: ${msg.topic}`);
    }
 
    if (msg.type === "unsubscribe") {
      clientSubscriptions.get(ws)?.delete(msg.topic);
    }
  });
 
  ws.on("close", () => {
    clientSubscriptions.delete(ws);
  });
});
 
// Simple MQTT-style topic matching
function topicMatches(pattern: string, topic: string): boolean {
  const patternParts = pattern.split("/");
  const topicParts = topic.split("/");
 
  for (let i = 0; i < patternParts.length; i++) {
    if (patternParts[i] === "#") return true;
    if (patternParts[i] === "+") continue;
    if (patternParts[i] !== topicParts[i]) return false;
  }
 
  return patternParts.length === topicParts.length;
}

// Browser-side: connect to MQTT broker via WebSocket
import mqtt from "mqtt";
 
// Connect to Mosquitto's WebSocket port (9001)
const client = mqtt.connect("ws://localhost:9001");
 
client.on("connect", () => {
  console.log("Connected to MQTT broker via WebSocket");
 
  // Subscribe to temperature data for a specific zone
  client.subscribe("factory/zone-1/sensors/+/readings");
});
 
client.on("message", (topic, message) => {
  const data = JSON.parse(message.toString());
  const deviceId = topic.split("/")[3];
 
  // Update dashboard UI
  updateChart(deviceId, data.value, data.timestamp);
  updateGauge(deviceId, data.value);
 
  // Check for alerts
  if (data.value > 35) {
    showAlert(`${deviceId}: ${data.value}°C exceeds threshold!`);
  }
});
 
function updateChart(deviceId: string, value: number, timestamp: number) {
  // Push data point to chart library (Chart.js, D3, etc.)
  console.log(`Chart update: ${deviceId} = ${value}°C at ${new Date(timestamp)}`);
}
 
function updateGauge(deviceId: string, value: number) {
  console.log(`Gauge update: ${deviceId} = ${value}°C`);
}
 
function showAlert(message: string) {
  console.warn(`ALERT: ${message}`);
}

import mqtt from "mqtt";
import { createServer as createCoapServer } from "coap";
import amqplib from "amqplib";
 
// ── Multi-protocol IoT gateway ──
class ProtocolBridge {
  private mqttClient: mqtt.MqttClient;
  private amqpChannel: amqplib.Channel | null = null;
 
  constructor() {
    // Outbound: MQTT for real-time telemetry
    this.mqttClient = mqtt.connect("mqtt://broker.example.com", {
      clientId: "gateway-bridge-01",
    });
  }
 
  async initialize() {
    // Outbound: AMQP for reliable processing
    const amqpConn = await amqplib.connect("amqp://localhost");
    this.amqpChannel = await amqpConn.createChannel();
    await this.amqpChannel.assertExchange("iot.ingestion", "topic", {
      durable: true,
    });
 
    // Inbound: CoAP server for local sensor data
    this.startCoapListener();
 
    console.log("Protocol bridge initialized");
  }
 
  private startCoapListener() {
    const server = createCoapServer();
 
    server.on("request", (req, res) => {
      if (req.method !== "POST") {
        res.code = "4.05"; // Method Not Allowed
        res.end();
        return;
      }
 
      const data = JSON.parse(req.payload.toString());
      const deviceId = req.url.split("/")[1]; // /sensor-042/telemetry
 
      // Bridge CoAP → MQTT (real-time)
      this.mqttClient.publish(
        `devices/${deviceId}/telemetry`,
        JSON.stringify(data),
        { qos: 1 }
      );
 
      // Bridge CoAP → AMQP (reliable processing)
      this.amqpChannel?.publish(
        "iot.ingestion",
        `telemetry.${deviceId}`,
        Buffer.from(JSON.stringify({
          ...data,
          gatewayId: "gateway-01",
          bridgedFrom: "coap",
          receivedAt: new Date().toISOString(),
        })),
        { persistent: true }
      );
 
      res.code = "2.01"; // Created
      res.end();
    });
 
    server.listen(5683, () => {
      console.log("CoAP listener on port 5683");
    });
  }
}
 
const bridge = new ProtocolBridge();
bridge.initialize();

import mqtt from "mqtt";
 
// Bridge MQTT alerts to HTTP webhooks
const client = mqtt.connect("mqtt://localhost:1883");
 
const webhooks: Record<string, string> = {
  "alerts/temperature": "https://hooks.slack.com/services/xxx",
  "alerts/device-offline": "https://api.pagerduty.com/events",
  "alerts/battery-low": "https://your-api.com/webhooks/battery",
};
 
client.on("connect", () => {
  // Subscribe to all alert topics
  client.subscribe("alerts/#");
});
 
client.on("message", async (topic, payload) => {
  const data = JSON.parse(payload.toString());
 
  // Find matching webhook
  for (const [pattern, url] of Object.entries(webhooks)) {
    if (topic.startsWith(pattern.replace("#", ""))) {
      try {
        await fetch(url, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            source: "iot-bridge",
            topic,
            data,
            timestamp: new Date().toISOString(),
          }),
        });
        console.log(`Forwarded ${topic} → ${url}`);
      } catch (error) {
        console.error(`Webhook failed for ${topic}: ${error}`);
      }
    }
  }
});

// ❌ BAD: QoS 2 for high-frequency telemetry
setInterval(() => {
  client.publish("sensors/gps", JSON.stringify(gpsData), {
    qos: 2, // 4 round-trips per message × 1000 devices × 1 msg/sec = overloaded broker
  });
}, 1000);

// ✅ GOOD: Match QoS to the data's importance
// GPS every second: QoS 0 (missing one is fine)
client.publish("sensors/gps", JSON.stringify(gpsData), { qos: 0 });
 
// Temperature every 30s: QoS 1 (we want all readings)
client.publish("sensors/temp", JSON.stringify(tempData), { qos: 1 });
 
// Actuator command: QoS 2 (must execute exactly once)
client.publish("actuators/valve/cmd", JSON.stringify(cmd), { qos: 2 });

// ❌ BAD: Inconsistent, flat topic names
client.publish("temp_sensor_42", data);
client.publish("TemperatureSensor/42", data);
client.publish("sensor-temperature-42", data);
// Can't use wildcards effectively, no hierarchy

// ✅ GOOD: Consistent hierarchical topics
client.publish("acme/factory-1/sensors/temp-042/readings", data);
client.publish("acme/factory-1/sensors/humidity-003/readings", data);
client.publish("acme/factory-1/actuators/fan-01/commands", cmd);
// Easy to subscribe: "acme/factory-1/sensors/+/readings"
// Easy to monitor: "acme/factory-1/#"

// ❌ BAD: MQTT for battery-powered device-to-gateway on local network
// MQTT requires TCP connection (3-way handshake + keepalive packets)
// On a coin battery, TCP keepalives drain power even when idle
const client = mqtt.connect("mqtt://gateway.local");
// Keepalive every 60s = ~1440 TCP packets/day even with no data

// ✅ GOOD: CoAP for local, battery-powered devices
// UDP: send one packet, no connection overhead, no keepalives
const req = request({
  hostname: "gateway.local",
  port: 5683,
  pathname: "/telemetry",
  method: "POST",
  confirmable: false, // NON: fire-and-forget (even less overhead)
});
req.write(JSON.stringify(sensorData));
req.end();
// 0 packets between readings — battery lasts months longer

// ❌ BAD: Dashboard shows "unknown" until next status update
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
}));
// New dashboard subscriber sees nothing until next publish

// ✅ GOOD: Retained message — new subscribers get status immediately
client.publish("devices/sensor-042/status", JSON.stringify({
  state: "online",
  firmware: "2.1.0",
  uptime: 86400,
}), { retain: true });
// Dashboard connecting 5 minutes later immediately shows "online"