Real-time IoT Sensor Data Visualization with MQTT and HiveMQ

Used MQTT to display Temperature sensor data

Internet of things (IoT) has revolutionized how we collect data and process data from the physical devices. Transmitting data and displaying in real-time is one of the critical aspects in the IoT systems. From this blog post let us see How can we use MQTT with HIVEMQ as the MQTT broker to make a robust real-time pipeline for IoT sensor data.

Why choose HIVEMQ?

HIVEMQ

HIVEMQ is a leading MQTT broker that offers both cloud and self-hosted solutions. Given below are some reasons why it is an excellent choice for IoT projects:

1. Enterprise-grade reliability: 99.99% uptime SLA
2. Horizontal scalability: Handle millions of concurrent connections
3. Horizontal scalability: Handle millions of concurrent connections
4. Cloud-native: Available as a managed service (HIVEMQ Cloud)
5. Extensions: Rich ecosystem of plugins and integrations

Architecture Overview:

Our real-time sensor data system consists of four main components:

1. IoT sensors: These are the physical devices that collect data from the environment
2. MQTT Clients: These are the Software that publishes sensor data to MQTT topics
3. HIVEMQ Broker: This is the central message broker that routes data between publishers and subscribes
4. Data visualization: Web dashboard that subscribes to data and displays real-time data updates

Setting Up HIVEMQ

HIVEMQ Cloud is what we are going to use here. It is really a beginner friendly platform. Now, Let us see the steps to do this accurately.

1. Sign Up for a free account at HIVEMQ Cloud

2.Create a new serverless cluster

This is the cluster I created during my project

3.Configure your connection settings: Create the credentials for your devices and applications

4.Note Connection details: Note down details like Broker URL,Port,Username and Password.

Implementing the IoT Sensor Publisher

This can be be done by using python or Node.js. Let us see steps for Node.js

1. Initialize Node.js Project
2. Install MQTT library: For this you can run the command npm install mqtt
3. Create sensor-publisher.js file
4. Import MQTT library and set up connection to HiveMQ Cloud
5. Create functions to simulate sensor data (temperature, humidity)
6. Implement MQTT publishing with proper error handling
7. Set up periodic data publishing
8. Add connection status monitoring and reconnection logic

Sample sensor-publisher.js code

This is just a sample code. You can make your own one with your credentials in it.

const mqtt = require('mqtt');

class IoTSensorPublisher {
  constructor(brokerHost, brokerPort, username, password) {
    this.brokerHost = brokerHost;
    this.brokerPort = brokerPort;
    this.isConnected = false;

    // Connect to HiveMQ Cloud with TLS
    this.client = mqtt.connect(`mqtts://${brokerHost}:${brokerPort}`, {
      username: username,
      password: password,
      protocol: 'mqtts',
      rejectUnauthorized: true
    });

    this.setupEventHandlers();
  }

  setupEventHandlers() {
    this.client.on('connect', () => {
      console.log('✅ Connected to HiveMQ Cloud successfully');
      this.isConnected = true;
      this.startPublishing();
    });

    this.client.on('error', (error) => {
      console.error('❌ Connection error:', error);
      this.isConnected = false;
    });

    this.client.on('close', () => {
      console.log('🔌 Disconnected from HiveMQ Cloud');
      this.isConnected = false;
    });

    this.client.on('offline', () => {
      console.log('📴 Client is offline');
      this.isConnected = false;
    });
  }

  simulateSensorData() {
    // Simulate realistic sensor readings
    const temperature = Math.round((20 + Math.random() * 15) * 100) / 100; // 20-35°C
    const humidity = Math.round((40 + Math.random() * 40) * 100) / 100;    // 40-80%

    return {
      device_id: 'sensor_001',
      timestamp: new Date().toISOString(),
      temperature: temperature,
      humidity: humidity,
      location: 'Office Building A',
      battery_level: Math.round(Math.random() * 40 + 60), // 60-100%
      signal_strength: Math.round(Math.random() * 30 + 70) // 70-100%
    };
  }

  async publishSensorData(data) {
    if (!this.isConnected) {
      console.log('⏳ Waiting for connection...');
      return;
    }

    const topic = `sensors/${data.device_id}/data`;
    const payload = JSON.stringify(data);

    try {
      await new Promise((resolve, reject) => {
        this.client.publish(topic, payload, { qos: 1 }, (error) => {
          if (error) {
            reject(error);
          } else {
            resolve();
          }
        });
      });

      console.log(`📡 Published: ${data.temperature}°C, ${data.humidity}% at ${new Date(data.timestamp).toLocaleTimeString()}`);

    } catch (error) {
      console.error('❌ Publishing error:', error);
    }
  }

  startPublishing() {
    console.log('🚀 Starting sensor data publishing every 5 seconds...');

    const publishInterval = setInterval(() => {
      if (this.isConnected) {
        const sensorData = this.simulateSensorData();
        this.publishSensorData(sensorData);
      }
    }, 5000);

    // Graceful shutdown
    process.on('SIGINT', () => {
      console.log('\n🛑 Shutting down sensor publisher...');
      clearInterval(publishInterval);
      this.client.end();
      process.exit(0);
    });
  }
}

// Usage - Replace with your HiveMQ Cloud credentials
const publisher = new IoTSensorPublisher(
  'your-cluster.s1.eu.hivemq.cloud',  // Your HiveMQ Cloud URL
  8883,                               // TLS port
  'your-username',                    // Your HiveMQ username
  'your-password'                     // Your HiveMQ password
);

// The publisher will automatically start when connection is established

Design MQTT Topic Structure

Use a hierarchical topic structure:

• sensors/{device_id}/data - Raw sensor readings
• sensors/{device_id}/status - Device health status
• sensors/{device_id}/config - Configuration updates

The next step is testing the sensor-publisher.js file. For this run the command the node sensor-publisher.js.

Building the Backend API (Node.js + Express)

Create server.js file in a new directory and then install the dependencies to that. For this you can run the command npm install express mqtt socket.io cors.

Then inside the server.js files these things should be done.

MQTT Subscriber Implementation

1. Connect Express server to HiveMQ Cloud as MQTT subscriber
2. Subscribe to sensor data topics (sensors/+/data)
3. Parse incoming JSON sensor data
4. Store latest readings in memory or database
5. Implement error handling for malformed messages

WebSocket Integration

1. Set up Socket.IO server for real-time communication
2. Configure CORS for React frontend connection
3. Emit sensor data to connected web clients when MQTT messages arrive
4. Handle client connections and disconnections
5. Send initial data to newly connected clients

REST API Endpoints

Create these API endpoints:

1. GET /api/sensors - Get all current sensor data
2. GET /api/sensors/:deviceId - Get specific sensor data
3. GET /api/sensors/:deviceId/history - Get historical data (if storing)

Sample server.js code:

// server.js - Backend API Server for IoT Dashboard
const express = require('express');
const mqtt = require('mqtt');
const http = require('http');
const socketIo = require('socket.io');
const cors = require('cors');

// Initialize Express app
const app = express();
const server = http.createServer(app);

// Configure Socket.IO with CORS for React frontend
const io = socketIo(server, {
  cors: {
    origin: "http://localhost:3000", // React development server URL
    methods: ["GET", "POST"],
    credentials: true
  }
});

// Middleware setup
app.use(cors()); // Enable CORS for all routes
app.use(express.json()); // Parse JSON request bodies

// ==========================================
// MQTT CONNECTION TO HIVEMQ CLOUD
// ==========================================

// IMPORTANT: Replace these with your actual HiveMQ Cloud credentials
const HIVEMQ_CONFIG = {
  host: 'your-cluster.s1.eu.hivemq.cloud', // Your HiveMQ Cloud broker URL
  port: 8883,                              // TLS port for secure connection
  username: 'your-username',               // Your HiveMQ username
  password: 'your-password'                // Your HiveMQ password
};

// Connect to HiveMQ Cloud with TLS encryption
const mqttClient = mqtt.connect(`mqtts://${HIVEMQ_CONFIG.host}:${HIVEMQ_CONFIG.port}`, {
  username: HIVEMQ_CONFIG.username,
  password: HIVEMQ_CONFIG.password,
  protocol: 'mqtts',
  rejectUnauthorized: true // Verify TLS certificates
});

// ==========================================
// DATA STORAGE (In-Memory)
// ==========================================

// Store latest sensor readings in memory
// IMPORTANT: In production, use a database like MongoDB or PostgreSQL
let sensorData = {};

// Store historical data for charts (last 50 readings per sensor)
let sensorHistory = {};

// Helper function to add data to history
function addToHistory(deviceId, data) {
  if (!sensorHistory[deviceId]) {
    sensorHistory[deviceId] = [];
  }

  sensorHistory[deviceId].push(data);

  // Keep only last 50 readings to prevent memory issues
  if (sensorHistory[deviceId].length > 50) {
    sensorHistory[deviceId] = sensorHistory[deviceId].slice(-50);
  }
}

// ==========================================
// MQTT EVENT HANDLERS
// ==========================================

// IMPORTANT: This runs when successfully connected to HiveMQ Cloud
mqttClient.on('connect', () => {
  console.log('🌐 Connected to HiveMQ Cloud broker');

  // Subscribe to all sensor data topics
  // The '+' is a wildcard that matches any device_id
  mqttClient.subscribe('sensors/+/data', { qos: 1 }, (error) => {
    if (error) {
      console.error('❌ Subscription error:', error);
    } else {
      console.log('📡 Subscribed to sensors/+/data');
    }
  });

  // Optionally subscribe to device status topics
  mqttClient.subscribe('sensors/+/status', { qos: 1 });
});

// IMPORTANT: This processes every incoming MQTT message
mqttClient.on('message', (topic, message) => {
  try {
    // Parse the JSON payload from MQTT message
    const data = JSON.parse(message.toString());
    const deviceId = data.device_id;

    console.log(`📊 Received data from ${deviceId}:`, {
      temperature: data.temperature,
      humidity: data.humidity,
      timestamp: data.timestamp
    });

    // Store the latest reading for this device
    // IMPORTANT: This overwrites previous data - latest reading only
    sensorData[deviceId] = {
      ...data,
      lastUpdated: new Date().toISOString() // Add server timestamp
    };

    // Add to historical data for charts
    addToHistory(deviceId, data);

    // IMPORTANT: Broadcast to all connected React clients via WebSocket
    // This is how real-time updates reach the dashboard
    io.emit('sensorData', {
      deviceId: deviceId,
      data: data,
      history: sensorHistory[deviceId] || []
    });

    // Check for alert conditions
    checkAlerts(data);

  } catch (error) {
    console.error('❌ Error processing MQTT message:', error);
  }
});

// Handle MQTT connection errors
mqttClient.on('error', (error) => {
  console.error('❌ MQTT connection error:', error);
});

// Handle MQTT disconnections
mqttClient.on('close', () => {
  console.log('🔌 MQTT connection closed');
});

// ==========================================
// ALERT SYSTEM
// ==========================================

function checkAlerts(data) {
  const alerts = [];

  // Temperature alerts
  if (data.temperature > 30) {
    alerts.push({
      type: 'high_temperature',
      device_id: data.device_id,
      value: data.temperature,
      threshold: 30,
      message: `High temperature detected: ${data.temperature}°C`
    });
  }

  // Humidity alerts
  if (data.humidity > 75) {
    alerts.push({
      type: 'high_humidity',
      device_id: data.device_id,
      value: data.humidity,
      threshold: 75,
      message: `High humidity detected: ${data.humidity}%`
    });
  }

  // Low battery alerts
  if (data.battery_level && data.battery_level < 20) {
    alerts.push({
      type: 'low_battery',
      device_id: data.device_id,
      value: data.battery_level,
      threshold: 20,
      message: `Low battery: ${data.battery_level}%`
    });
  }

  // IMPORTANT: Send alerts to dashboard if any triggered
  if (alerts.length > 0) {
    console.log('🚨 Alerts triggered:', alerts);
    io.emit('alerts', alerts);
  }
}

// ==========================================
// REST API ENDPOINTS
// ==========================================

// Get all current sensor data
// IMPORTANT: This provides the current state of all sensors
app.get('/api/sensors', (req, res) => {
  try {
    res.json({
      success: true,
      data: sensorData,
      count: Object.keys(sensorData).length,
      timestamp: new Date().toISOString()
    });
  } catch (error) {
    res.status(500).json({ 
      success: false, 
      error: 'Failed to fetch sensor data' 
    });
  }
});

// Get specific sensor data by device ID
app.get('/api/sensors/:deviceId', (req, res) => {
  try {
    const deviceId = req.params.deviceId;
    const data = sensorData[deviceId];

    if (data) {
      res.json({
        success: true,
        data: data,
        history: sensorHistory[deviceId] || []
      });
    } else {
      res.status(404).json({ 
        success: false, 
        error: 'Sensor not found' 
      });
    }
  } catch (error) {
    res.status(500).json({ 
      success: false, 
      error: 'Failed to fetch sensor data' 
    });
  }
});

// Get historical data for charts
app.get('/api/sensors/:deviceId/history', (req, res) => {
  try {
    const deviceId = req.params.deviceId;
    const history = sensorHistory[deviceId] || [];

    res.json({
      success: true,
      deviceId: deviceId,
      history: history,
      count: history.length
    });
  } catch (error) {
    res.status(500).json({ 
      success: false, 
      error: 'Failed to fetch historical data' 
    });
  }
});

// Health check endpoint
app.get('/api/health', (req, res) => {
  res.json({
    success: true,
    mqtt_connected: mqttClient.connected,
    active_sensors: Object.keys(sensorData).length,
    uptime: process.uptime(),
    timestamp: new Date().toISOString()
  });
});

// ==========================================
// WEBSOCKET CONNECTIONS (Socket.IO)
// ==========================================

// IMPORTANT: Handle WebSocket connections from React dashboard
io.on('connection', (socket) => {
  console.log('🔗 Dashboard client connected:', socket.id);

  // Send current sensor data to newly connected client
  // IMPORTANT: This ensures new clients get immediate data
  socket.emit('initialData', {
    sensors: sensorData,
    history: sensorHistory
  });

  // Handle client-specific subscriptions (optional)
  socket.on('subscribe', (deviceId) => {
    socket.join(`device_${deviceId}`);
    console.log(`📱 Client subscribed to device: ${deviceId}`);
  });

  // Handle client disconnections
  socket.on('disconnect', () => {
    console.log('📴 Dashboard client disconnected:', socket.id);
  });

  // Optional: Handle commands from dashboard to devices
  socket.on('deviceCommand', (command) => {
    console.log('📤 Received device command:', command);

    // Publish command to MQTT (device control)
    const commandTopic = `sensors/${command.deviceId}/commands`;
    mqttClient.publish(commandTopic, JSON.stringify(command), { qos: 1 });
  });
});

// ==========================================
// SERVER STARTUP
// ==========================================

const PORT = process.env.PORT || 5000;

// Start the server
server.listen(PORT, () => {
  console.log(`🚀 Server running on port ${PORT}`);
  console.log(`📊 Dashboard API: http://localhost:${PORT}/api/sensors`);
  console.log(`🌐 WebSocket endpoint: http://localhost:${PORT}`);
});

// ==========================================
// GRACEFUL SHUTDOWN
// ==========================================

// IMPORTANT: Handle graceful shutdown
process.on('SIGINT', () => {
  console.log('\n🛑 Shutting down server...');

  // Close MQTT connection
  mqttClient.end();

  // Close HTTP server
  server.close(() => {
    console.log('✅ Server shut down gracefully');
    process.exit(0);
  });
});

// Handle uncaught exceptions
process.on('uncaughtException', (error) => {
  console.error('💥 Uncaught Exception:', error);
  process.exit(1);
});

// Handle unhandled promise rejections
process.on('unhandledRejection', (error) => {
  console.error('💥 Unhandled Rejection:', error);
  process.exit(1);
});

Then the back-end should be tested. For this run the command node server.js.

After this the dashboard can be created using react and then it is just a matter of fetching data from back-end APIs.

Humidity data displayed using MQTT in Front-end

Conclusion

Building a real-time IoT sensor data system with MQTT and HiveMQ Cloud provides a robust, scalable foundation for modern IoT applications. The combination of Node.js for both publishers and backend services, Express for API management, and React for visualization creates a powerful, maintainable system.

Key benefits of this approach:

• Unified language: JavaScript across the entire stack
• Real-time capabilities: Instant data updates in the dashboard
• Scalable architecture: Easy to add more sensors and features
• Cloud-managed broker: No infrastructure maintenance required
• Professional visualization: Interactive charts and metrics

This architecture can handle everything from simple prototypes to enterprise-scale IoT deployments, making it an excellent choice for your real-time sensor data visualization needs.