edge-iot

Edge Computing & IoT

Overview

Building applications for edge devices, IoT protocols, and embedded systems integration.

MQTT Protocol

Broker Setup (Mosquitto)

# docker-compose.yml
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
      - "9001:9001"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf
      - mosquitto_data:/mosquitto/data
      - mosquitto_log:/mosquitto/log

volumes:
  mosquitto_data:
  mosquitto_log:

# mosquitto.conf
listener 1883
listener 9001
protocol websockets

allow_anonymous false
password_file /mosquitto/config/passwd

persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log

Node.js MQTT Client

import mqtt from 'mqtt';

class MQTTClient {
  private client: mqtt.MqttClient;
  private subscriptions = new Map<string, Set<Function>>();

  constructor(brokerUrl: string, options?: mqtt.IClientOptions) {
    this.client = mqtt.connect(brokerUrl, {
      clientId: `node_${Math.random().toString(16).slice(2, 10)}`,
      clean: true,
      reconnectPeriod: 5000,
      ...options,
    });

    this.client.on('connect', () => {
      console.log('MQTT connected');
      // Resubscribe to all topics
      this.subscriptions.forEach((_, topic) => {
        this.client.subscribe(topic);
      });
    });

    this.client.on('message', (topic, payload) => {
      const handlers = this.getMatchingHandlers(topic);
      const message = this.parsePayload(payload);

      handlers.forEach(handler => handler(topic, message));
    });

    this.client.on('error', (error) => {
      console.error('MQTT error:', error);
    });
  }

  subscribe(topic: string, handler: (topic: string, message: any) => void) {
    if (!this.subscriptions.has(topic)) {
      this.subscriptions.set(topic, new Set());
      this.client.subscribe(topic);
    }

    this.subscriptions.get(topic)!.add(handler);

    return () => {
      this.subscriptions.get(topic)?.delete(handler);
      if (this.subscriptions.get(topic)?.size === 0) {
        this.subscriptions.delete(topic);
        this.client.unsubscribe(topic);
      }
    };
  }

  publish(topic: string, message: any, options?: mqtt.IClientPublishOptions) {
    const payload = typeof message === 'string'
      ? message
      : JSON.stringify(message);

    this.client.publish(topic, payload, {
      qos: 1,
      ...options,
    });
  }

  private getMatchingHandlers(topic: string): Set<Function> {
    const handlers = new Set<Function>();

    this.subscriptions.forEach((topicHandlers, pattern) => {
      if (this.topicMatches(pattern, topic)) {
        topicHandlers.forEach(h => handlers.add(h));
      }
    });

    return handlers;
  }

  private 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;
  }

  private parsePayload(payload: Buffer): any {
    const str = payload.toString();
    try {
      return JSON.parse(str);
    } catch {
      return str;
    }
  }

  disconnect() {
    this.client.end();
  }
}

// Usage
const mqtt = new MQTTClient('mqtt://localhost:1883', {
  username: 'user',
  password: 'pass',
});

// Subscribe to device telemetry
mqtt.subscribe('devices/+/telemetry', (topic, data) => {
  const deviceId = topic.split('/')[1];
  console.log(`Device ${deviceId}:`, data);
});

// Subscribe to all events from a device
mqtt.subscribe('devices/sensor-001/#', (topic, data) => {
  console.log(`${topic}:`, data);
});

// Publish command to device
mqtt.publish('devices/sensor-001/commands', {
  action: 'reboot',
  timestamp: Date.now(),
});