Eclipse IoT provides a comprehensive set of open-source technologies for building Internet of Things solutions. This guide covers Eclipse Paho (MQTT), Eclipse Mosquitto, Eclipse Californium (CoAP), and Eclipse Ditto for digital twins.

Project Setup and Dependencies

Maven Configuration

<!-- pom.xml --> <properties> <paho.version>1.2.5</paho.version> <californium.version>3.7.0</californium.version> <jetty.version>9.4.48.v20220622</jetty.version> </properties> <dependencies> <!-- Eclipse Paho MQTT Client --> <dependency> <groupId>org.eclipse.paho</groupId> <artifactId>org.eclipse.paho.client.mqttv3</artifactId> <version>${paho.version}</version> </dependency> <!-- Eclipse Californium CoAP --> <dependency> <groupId>org.eclipse.californium</groupId> <artifactId>californium-core</artifactId> <version>${californium.version}</version> </dependency> <dependency> <groupId>org.eclipse.californium</groupId> <artifactId>californium-proxy2</artifactId> <version>${californium.version}</version> </dependency> <!-- HTTP Client for REST APIs --> <dependency> <groupId>org.eclipse.jetty</groupId> <artifactId>jetty-client</artifactId> <version>${jetty.version}</version> </dependency> <!-- JSON Processing --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.15.2</version> </dependency> <!-- Logging --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>2.0.7</version> </dependency> </dependencies>

MQTT with Eclipse Paho

MQTT Client Manager

package com.iot.mqtt; import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class MQTTClientManager { private static final Logger logger = LoggerFactory.getLogger(MQTTClientManager.class); private MqttClient client; private final String brokerUrl; private final String clientId; private final ConcurrentHashMap<String, MessageHandler> topicHandlers; private final ScheduledExecutorService reconnectScheduler; private static final int QOS = 1; private static final int RECONNECT_DELAY = 5; // seconds public MQTTClientManager(String brokerUrl, String clientId) { this.brokerUrl = brokerUrl; this.clientId = clientId; this.topicHandlers = new ConcurrentHashMap<>(); this.reconnectScheduler = Executors.newSingleThreadScheduledExecutor(); } public void connect() throws MQTTException { try { MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(true); options.setAutomaticReconnect(true); options.setConnectionTimeout(30); options.setKeepAliveInterval(60); // Optional authentication // options.setUserName("username"); // options.setPassword("password".toCharArray()); client = new MqttClient(brokerUrl, clientId, new MemoryPersistence()); client.setCallback(new MQTTMessageCallback()); logger.info("Connecting to MQTT broker: {}", brokerUrl); client.connect(options); logger.info("Connected to MQTT broker successfully"); } catch (MqttException e) { logger.error("Failed to connect to MQTT broker", e); scheduleReconnect(); throw new MQTTException("Connection failed", e); } } public void subscribe(String topic, MessageHandler handler) throws MQTTException { try { if (client != null && client.isConnected()) { client.subscribe(topic, QOS); topicHandlers.put(topic, handler); logger.info("Subscribed to topic: {}", topic); } else { throw new MQTTException("Client not connected"); } } catch (MqttException e) { logger.error("Failed to subscribe to topic: {}", topic, e); throw new MQTTException("Subscription failed", e); } } public void publish(String topic, String message) throws MQTTException { publish(topic, message, QOS, false); } public void publish(String topic, String message, int qos, boolean retained) throws MQTTException { try { if (client != null && client.isConnected()) { MqttMessage mqttMessage = new MqttMessage(message.getBytes()); mqttMessage.setQos(qos); mqttMessage.setRetained(retained); client.publish(topic, mqttMessage); logger.debug("Published message to topic {}: {}", topic, message); } else { throw new MQTTException("Client not connected"); } } catch (MqttException e) { logger.error("Failed to publish message to topic: {}", topic, e); throw new MQTTException("Publish failed", e); } } public void disconnect() { try { if (client != null && client.isConnected()) { client.disconnect(); logger.info("Disconnected from MQTT broker"); } reconnectScheduler.shutdown(); } catch (MqttException e) { logger.error("Error during disconnect", e); } } private void scheduleReconnect() { logger.info("Scheduling reconnect in {} seconds", RECONNECT_DELAY); reconnectScheduler.schedule(() -> { try { connect(); } catch (MQTTException e) { logger.error("Reconnect failed, scheduling again"); scheduleReconnect(); } }, RECONNECT_DELAY, TimeUnit.SECONDS); } private class MQTTMessageCallback implements MqttCallback { @Override public void connectionLost(Throwable cause) { logger.warn("Connection to MQTT broker lost", cause); scheduleReconnect(); } @Override public void messageArrived(String topic, MqttMessage message) throws Exception { String payload = new String(message.getPayload()); logger.debug("Message received on topic {}: {}", topic, payload); MessageHandler handler = topicHandlers.get(topic); if (handler != null) { handler.handleMessage(topic, payload); } else { logger.warn("No handler registered for topic: {}", topic); } } @Override public void deliveryComplete(IMqttDeliveryToken token) { // Message delivery confirmed logger.debug("Message delivery complete"); } } @FunctionalInterface public interface MessageHandler { void handleMessage(String topic, String message); } public static class MQTTException extends Exception { public MQTTException(String message) { super(message); } public MQTTException(String message, Throwable cause) { super(message, cause); } } }

Smart Device Simulator using MQTT

package com.iot.devices; import com.iot.mqtt.MQTTClientManager; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Random; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class SmartThermostat { private static final Logger logger = LoggerFactory.getLogger(SmartThermostat.class); private static final ObjectMapper mapper = new ObjectMapper(); private final String deviceId; private final MQTTClientManager mqttClient; private final ScheduledExecutorService scheduler; private double currentTemperature; private double targetTemperature; private boolean heatingOn; private boolean coolingOn; private String mode; // "HEAT", "COOL", "OFF" private static final String TELEMETRY_TOPIC = "devices/%s/telemetry"; private static final String COMMAND_TOPIC = "devices/%s/commands"; private static final String STATE_TOPIC = "devices/%s/state"; public SmartThermostat(String deviceId, String brokerUrl) { this.deviceId = deviceId; this.mqttClient = new MQTTClientManager(brokerUrl, "thermostat-" + deviceId); this.scheduler = Executors.newSingleThreadScheduledExecutor(); this.currentTemperature = 20.0; // Starting temperature this.targetTemperature = 22.0; this.mode = "HEAT"; this.heatingOn = false; this.coolingOn = false; } public void start() { try { // Connect to MQTT broker mqttClient.connect(); // Subscribe to command topic String commandTopic = String.format(COMMAND_TOPIC, deviceId); mqttClient.subscribe(commandTopic, this::handleCommand); // Start telemetry publishing scheduler.scheduleAtFixedRate(this::publishTelemetry, 0, 10, TimeUnit.SECONDS); // Start state simulation scheduler.scheduleAtFixedRate(this::simulateEnvironment, 0, 2, TimeUnit.SECONDS); // Publish initial state publishState(); logger.info("Smart thermostat {} started successfully", deviceId); } catch (Exception e) { logger.error("Failed to start smart thermostat", e); } } private void handleCommand(String topic, String message) { try { ObjectNode command = (ObjectNode) mapper.readTree(message); String commandType = command.get("command").asText(); logger.info("Received command: {}", commandType); switch (commandType) { case "SET_TEMPERATURE": double newTemp = command.get("temperature").asDouble(); setTargetTemperature(newTemp); break; case "SET_MODE": String newMode = command.get("mode").asText(); setMode(newMode); break; case "GET_STATE": publishState(); break; default: logger.warn("Unknown command: {}", commandType); } // Send command acknowledgment publishCommandAck(commandType); } catch (Exception e) { logger.error("Error processing command", e); } } private void setTargetTemperature(double temperature) { this.targetTemperature = temperature; logger.info("Target temperature set to: {}°C", temperature); updateHVACState(); publishState(); } private void setMode(String mode) { this.mode = mode; logger.info("Mode set to: {}", mode); updateHVACState(); publishState(); } private void updateHVACState() { double diff = currentTemperature - targetTemperature; double tolerance = 0.5; switch (mode) { case "HEAT": heatingOn = diff < -tolerance; coolingOn = false; break; case "COOL": coolingOn = diff > tolerance; heatingOn = false; break; case "OFF": default: heatingOn = false; coolingOn = false; break; } } private void simulateEnvironment() { Random random = new Random(); // Simulate temperature changes double externalEffect = (random.nextDouble() - 0.5) * 0.2; // Random environmental effect if (heatingOn) { currentTemperature += 0.1 + externalEffect; } else if (coolingOn) { currentTemperature -= 0.1 + externalEffect; } else { currentTemperature += externalEffect; // Drift towards ambient } // Add some random fluctuation currentTemperature += (random.nextDouble() - 0.5) * 0.05; // Keep temperature within reasonable bounds currentTemperature = Math.max(10.0, Math.min(35.0, currentTemperature)); updateHVACState(); } private void publishTelemetry() { try { ObjectNode telemetry = mapper.createObjectNode(); telemetry.put("deviceId", deviceId); telemetry.put("timestamp", System.currentTimeMillis()); telemetry.put("currentTemperature", Math.round(currentTemperature * 10.0) / 10.0); telemetry.put("targetTemperature", targetTemperature); telemetry.put("heatingOn", heatingOn); telemetry.put("coolingOn", coolingOn); telemetry.put("mode", mode); telemetry.put("powerConsumption", calculatePowerConsumption()); String topic = String.format(TELEMETRY_TOPIC, deviceId); mqttClient.publish(topic, telemetry.toString()); } catch (Exception e) { logger.error("Failed to publish telemetry", e); } } private void publishState() { try { ObjectNode state = mapper.createObjectNode(); state.put("deviceId", deviceId); state.put("online", true); state.put("currentTemperature", Math.round(currentTemperature * 10.0) / 10.0); state.put("targetTemperature", targetTemperature); state.put("mode", mode); state.put("heatingOn", heatingOn); state.put("coolingOn", coolingOn); state.put("lastUpdate", System.currentTimeMillis()); String topic = String.format(STATE_TOPIC, deviceId); mqttClient.publish(topic, state.toString(), 1, true); // Retained message } catch (Exception e) { logger.error("Failed to publish state", e); } } private void publishCommandAck(String command) { try { ObjectNode ack = mapper.createObjectNode(); ack.put("deviceId", deviceId); ack.put("command", command); ack.put("status", "SUCCESS"); ack.put("timestamp", System.currentTimeMillis()); String topic = "devices/" + deviceId + "/ack"; mqttClient.publish(topic, ack.toString()); } catch (Exception e) { logger.error("Failed to publish command ACK", e); } } private double calculatePowerConsumption() { double consumption = 0.0; if (heatingOn) consumption += 1.5; // kW if (coolingOn) consumption += 1.2; // kW return consumption; } public void stop() { scheduler.shutdown(); mqttClient.disconnect(); logger.info("Smart thermostat {} stopped", deviceId); } // Getters for testing public double getCurrentTemperature() { return currentTemperature; } public double getTargetTemperature() { return targetTemperature; } public String getMode() { return mode; } }

CoAP with Eclipse Californium

CoAP Server for Constrained Devices

package com.iot.coap; import org.eclipse.californium.core.CoapResource; import org.eclipse.californium.core.CoapServer; import org.eclipse.californium.core.coap.CoAP; import org.eclipse.californium.core.network.config.NetworkConfig; import org.eclipse.californium.core.server.resources.CoapExchange; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.concurrent.ConcurrentHashMap; public class CoAPDeviceServer extends CoapServer { private static final Logger logger = LoggerFactory.getLogger(CoAPDeviceServer.class); private static final ObjectMapper mapper = new ObjectMapper(); private final ConcurrentHashMap<String, DeviceResource> devices; public CoAPDeviceServer(int port) { super(port); this.devices = new ConcurrentHashMap<>(); // Add default resources add(new DeviceDiscoveryResource()); add(new HealthResource()); logger.info("CoAP Server started on port {}", port); } public void registerDevice(String deviceId, DeviceInfo deviceInfo) { DeviceResource deviceResource = new DeviceResource(deviceId, deviceInfo); devices.put(deviceId, deviceResource); add(deviceResource); logger.info("Registered device: {}", deviceId); } public void unregisterDevice(String deviceId) { DeviceResource device = devices.remove(deviceId); if (device != null) { remove(device); logger.info("Unregistered device: {}", deviceId); } } // Device discovery resource private class DeviceDiscoveryResource extends CoapResource { public DeviceDiscoveryResource() { super("devices"); setObservable(true); } @Override public void handleGET(CoapExchange exchange) { try { ObjectNode response = mapper.createObjectNode(); devices.forEach((id, device) -> { response.putObject(id).put("name", device.getDeviceInfo().getName()); }); exchange.respond(CoAP.ResponseCode.CONTENT, response.toString()); } catch (Exception e) { logger.error("Error handling device discovery", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); } } } // Health check resource private class HealthResource extends CoapResource { public HealthResource() { super("health"); } @Override public void handleGET(CoapExchange exchange) { ObjectNode health = mapper.createObjectNode(); health.put("status", "healthy"); health.put("timestamp", System.currentTimeMillis()); health.put("deviceCount", devices.size()); exchange.respond(CoAP.ResponseCode.CONTENT, health.toString()); } } // Individual device resource public static class DeviceResource extends CoapResource { private final String deviceId; private final DeviceInfo deviceInfo; private String status = "online"; private long lastSeen; public DeviceResource(String deviceId, DeviceInfo deviceInfo) { super(deviceId); this.deviceId = deviceId; this.deviceInfo = deviceInfo; this.lastSeen = System.currentTimeMillis(); // Add sub-resources add(new TelemetryResource()); add(new CommandResource()); add(new ConfigurationResource()); } @Override public void handleGET(CoapExchange exchange) { try { ObjectNode info = mapper.createObjectNode(); info.put("deviceId", deviceId); info.put("name", deviceInfo.getName()); info.put("type", deviceInfo.getType()); info.put("status", status); info.put("lastSeen", lastSeen); info.put("manufacturer", deviceInfo.getManufacturer()); exchange.respond(CoAP.ResponseCode.CONTENT, info.toString()); } catch (Exception e) { logger.error("Error handling device info request", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); } } public DeviceInfo getDeviceInfo() { return deviceInfo; } public void updateLastSeen() { this.lastSeen = System.currentTimeMillis(); } // Telemetry sub-resource private class TelemetryResource extends CoapResource { public TelemetryResource() { super("telemetry"); setObservable(true); getAttributes().setObservable(); } @Override public void handleGET(CoapExchange exchange) { updateLastSeen(); try { // Simulate telemetry data ObjectNode telemetry = mapper.createObjectNode(); telemetry.put("deviceId", deviceId); telemetry.put("timestamp", System.currentTimeMillis()); telemetry.put("temperature", 23.5 + Math.random() * 5); telemetry.put("humidity", 45 + Math.random() * 20); telemetry.put("battery", 85 + Math.random() * 15); exchange.respond(CoAP.ResponseCode.CONTENT, telemetry.toString()); } catch (Exception e) { logger.error("Error handling telemetry request", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); } } } // Command sub-resource private class CommandResource extends CoapResource { public CommandResource() { super("command"); } @Override public void handlePOST(CoapExchange exchange) { updateLastSeen(); try { String request = exchange.getRequestText(); ObjectNode command = (ObjectNode) mapper.readTree(request); String action = command.get("action").asText(); logger.info("Received command for device {}: {}", deviceId, action); // Process command ObjectNode response = mapper.createObjectNode(); response.put("deviceId", deviceId); response.put("action", action); response.put("status", "executed"); response.put("timestamp", System.currentTimeMillis()); exchange.respond(CoAP.ResponseCode.CHANGED, response.toString()); } catch (Exception e) { logger.error("Error handling command", e); exchange.respond(CoAP.ResponseCode.BAD_REQUEST); } } } // Configuration sub-resource private class ConfigurationResource extends CoapResource { public ConfigurationResource() { super("config"); } @Override public void handleGET(CoapExchange exchange) { updateLastSeen(); try { ObjectNode config = mapper.createObjectNode(); config.put("deviceId", deviceId); config.put("pollingInterval", 30); config.put("reportingEnabled", true); config.put("lowBatteryThreshold", 20); exchange.respond(CoAP.ResponseCode.CONTENT, config.toString()); } catch (Exception e) { logger.error("Error handling config request", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); } } @Override public void handlePUT(CoapExchange exchange) { updateLastSeen(); try { String configJson = exchange.getRequestText(); logger.info("Updated configuration for device {}: {}", deviceId, configJson); exchange.respond(CoAP.ResponseCode.CHANGED); } catch (Exception e) { logger.error("Error updating configuration", e); exchange.respond(CoAP.ResponseCode.BAD_REQUEST); } } } } public static class DeviceInfo { private final String name; private final String type; private final String manufacturer; public DeviceInfo(String name, String type, String manufacturer) { this.name = name; this.type = type; this.manufacturer = manufacturer; } public String getName() { return name; } public String getType() { return type; } public String getManufacturer() { return manufacturer; } } }

CoAP Client for Device Communication

package com.iot.coap; import org.eclipse.californium.core.CoapClient; import org.eclipse.californium.core.CoapResponse; import org.eclipse.californium.core.coap.MediaTypeRegistry; import org.eclipse.californium.core.coap.OptionSet; import org.eclipse.californium.core.coap.Request; import org.eclipse.californium.core.coap.CoAP; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class CoAPDeviceClient { private static final Logger logger = LoggerFactory.getLogger(CoAPDeviceClient.class); private static final ObjectMapper mapper = new ObjectMapper(); private final String serverBaseUrl; private final ScheduledExecutorService scheduler; public CoAPDeviceClient(String serverHost, int serverPort) { this.serverBaseUrl = "coap://" + serverHost + ":" + serverPort; this.scheduler = Executors.newScheduledThreadPool(2); } public CompletableFuture<String> discoverDevices() { return CompletableFuture.supplyAsync(() -> { try { CoapClient client = new CoapClient(serverBaseUrl + "/devices"); CoapResponse response = client.get(); if (response != null && response.isSuccess()) { return response.getResponseText(); } else { throw new RuntimeException("Failed to discover devices: " + (response != null ? response.getCode() : "No response")); } } catch (Exception e) { throw new RuntimeException("Device discovery failed", e); } }); } public CompletableFuture<String> getDeviceInfo(String deviceId) { return CompletableFuture.supplyAsync(() -> { try { CoapClient client = new CoapClient(serverBaseUrl + "/" + deviceId); CoapResponse response = client.get(); if (response != null && response.isSuccess()) { return response.getResponseText(); } else { throw new RuntimeException("Failed to get device info: " + (response != null ? response.getCode() : "No response")); } } catch (Exception e) { throw new RuntimeException("Get device info failed", e); } }); } public CompletableFuture<String> getTelemetry(String deviceId) { return CompletableFuture.supplyAsync(() -> { try { CoapClient client = new CoapClient(serverBaseUrl + "/" + deviceId + "/telemetry"); CoapResponse response = client.get(); if (response != null && response.isSuccess()) { return response.getResponseText(); } else { throw new RuntimeException("Failed to get telemetry: " + (response != null ? response.getCode() : "No response")); } } catch (Exception e) { throw new RuntimeException("Get telemetry failed", e); } }); } public CompletableFuture<Boolean> sendCommand(String deviceId, String action) { return CompletableFuture.supplyAsync(() -> { try { CoapClient client = new CoapClient(serverBaseUrl + "/" + deviceId + "/command"); ObjectNode command = mapper.createObjectNode(); command.put("action", action); command.put("timestamp", System.currentTimeMillis()); CoapResponse response = client.post(command.toString(), MediaTypeRegistry.APPLICATION_JSON); if (response != null && response.isSuccess()) { logger.info("Command executed successfully: {}", action); return true; } else { logger.error("Command failed: {}", response != null ? response.getCode() : "No response"); return false; } } catch (Exception e) { logger.error("Send command failed", e); return false; } }); } public void startTelemetryMonitoring(String deviceId, int intervalSeconds, TelemetryCallback callback) { scheduler.scheduleAtFixedRate(() -> { try { String telemetry = getTelemetry(deviceId).get(5, TimeUnit.SECONDS); callback.onTelemetryReceived(deviceId, telemetry); } catch (Exception e) { logger.error("Telemetry monitoring failed for device: {}", deviceId, e); callback.onError(deviceId, e); } }, 0, intervalSeconds, TimeUnit.SECONDS); } public void observeDevice(String deviceId, ObservationCallback callback) { try { CoapClient client = new CoapClient(serverBaseUrl + "/" + deviceId); Request request = Request.newGet(); request.setObserve(); request.setURI(serverBaseUrl + "/" + deviceId); client.observe(request, new org.eclipse.californium.core.coap.ResponseHandler() { @Override public void onLoad(CoapResponse response) { callback.onObservation(deviceId, response.getResponseText()); } @Override public void onError() { callback.onObservationError(deviceId, new RuntimeException("Observation error")); } }); } catch (Exception e) { logger.error("Failed to observe device: {}", deviceId, e); } } @FunctionalInterface public interface TelemetryCallback { void onTelemetryReceived(String deviceId, String telemetryData); default void onError(String deviceId, Exception e) { logger.error("Telemetry error for device {}", deviceId, e); } } @FunctionalInterface public interface ObservationCallback { void onObservation(String deviceId, String data); default void onObservationError(String deviceId, Exception e) { logger.error("Observation error for device {}", deviceId, e); } } public void shutdown() { scheduler.shutdown(); } }

IoT Gateway Implementation

MQTT to CoAP Bridge

package com.iot.gateway; import com.iot.mqtt.MQTTClientManager; import com.iot.coap.CoAPDeviceClient; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentHashMap; public class MQTTToCoAPBridge { private static final Logger logger = LoggerFactory.getLogger(MQTTToCoAPBridge.class); private static final ObjectMapper mapper = new ObjectMapper(); private final MQTTClientManager mqttClient; private final CoAPDeviceClient coapClient; private final ConcurrentHashMap<String, String> deviceMappings; public MQTTToCoAPBridge(String mqttBrokerUrl, String coapServerHost, int coapServerPort) { this.mqttClient = new MQTTClientManager(mqttBrokerUrl, "mqtt-coap-bridge"); this.coapClient = new CoAPDeviceClient(coapServerHost, coapServerPort); this.deviceMappings = new ConcurrentHashMap<>(); } public void start() { try { // Connect to MQTT broker mqttClient.connect(); // Subscribe to MQTT topics for CoAP device commands mqttClient.subscribe("gateway/coap/+/command", this::handleCoAPCommand); mqttClient.subscribe("gateway/coap/+/discover", this::handleDiscoveryRequest); // Discover CoAP devices and publish to MQTT discoverAndPublishDevices(); // Start periodic device discovery startPeriodicDiscovery(); logger.info("MQTT to CoAP bridge started successfully"); } catch (Exception e) { logger.error("Failed to start MQTT to CoAP bridge", e); } } private void handleCoAPCommand(String topic, String message) { try { // Extract device ID from topic: gateway/coap/{deviceId}/command String[] topicParts = topic.split("/"); if (topicParts.length >= 4) { String deviceId = topicParts[2]; String command = topicParts[3]; logger.info("Forwarding command to CoAP device {}: {}", deviceId, command); // Forward command to CoAP device coapClient.sendCommand(deviceId, command) .thenAccept(success -> { if (success) { publishCommandResult(deviceId, command, "SUCCESS"); } else { publishCommandResult(deviceId, command, "FAILED"); } }) .exceptionally(e -> { logger.error("Command execution failed for device: {}", deviceId, e); publishCommandResult(deviceId, command, "ERROR"); return null; }); } } catch (Exception e) { logger.error("Error handling CoAP command", e); } } private void handleDiscoveryRequest(String topic, String message) { try { logger.info("Handling device discovery request"); discoverAndPublishDevices(); } catch (Exception e) { logger.error("Error handling discovery request", e); } } private void discoverAndPublishDevices() { coapClient.discoverDevices() .thenAccept(devicesJson -> { try { // Publish discovered devices to MQTT mqttClient.publish("gateway/coap/devices/discovered", devicesJson); logger.info("Published discovered CoAP devices to MQTT"); // Update device mappings and start monitoring updateDeviceMappings(devicesJson); } catch (Exception e) { logger.error("Error publishing discovered devices", e); } }) .exceptionally(e -> { logger.error("Device discovery failed", e); return null; }); } private void updateDeviceMappings(String devicesJson) { try { ObjectNode devices = (ObjectNode) mapper.readTree(devicesJson); devices.fields().forEachRemaining(entry -> { String deviceId = entry.getKey(); ObjectNode deviceInfo = (ObjectNode) entry.getValue(); deviceMappings.put(deviceId, deviceInfo.get("name").asText()); // Start telemetry monitoring for this device startDeviceMonitoring(deviceId); }); } catch (Exception e) { logger.error("Error updating device mappings", e); } } private void startDeviceMonitoring(String deviceId) { // Monitor telemetry every 30 seconds coapClient.startTelemetryMonitoring(deviceId, 30, (monitoredDeviceId, telemetry) -> { try { // Forward telemetry to MQTT String topic = "gateway/coap/" + monitoredDeviceId + "/telemetry"; mqttClient.publish(topic, telemetry); logger.debug("Forwarded telemetry from CoAP device {} to MQTT", monitoredDeviceId); } catch (Exception e) { logger.error("Error forwarding telemetry for device: {}", monitoredDeviceId, e); } }); } private void startPeriodicDiscovery() { // Discover devices every 5 minutes new Thread(() -> { while (true) { try { Thread.sleep(5 * 60 * 1000); // 5 minutes discoverAndPublishDevices(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { logger.error("Periodic discovery failed", e); } } }).start(); } private void publishCommandResult(String deviceId, String command, String status) { try { ObjectNode result = mapper.createObjectNode(); result.put("deviceId", deviceId); result.put("command", command); result.put("status", status); result.put("timestamp", System.currentTimeMillis()); String topic = "gateway/coap/" + deviceId + "/command/result"; mqttClient.publish(topic, result.toString()); } catch (Exception e) { logger.error("Error publishing command result", e); } } public void stop() { mqttClient.disconnect(); coapClient.shutdown(); logger.info("MQTT to CoAP bridge stopped"); } }

IoT Data Processing and Analytics

Real-time Data Processor

package com.iot.analytics; import com.iot.mqtt.MQTTClientManager; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class IoTDataProcessor { private static final Logger logger = LoggerFactory.getLogger(IoTDataProcessor.class); private static final ObjectMapper mapper = new ObjectMapper(); private final MQTTClientManager mqttClient; private final ScheduledExecutorService analyticsScheduler; private final ConcurrentHashMap<String, DeviceStatistics> deviceStats; public IoTDataProcessor(String mqttBrokerUrl) { this.mqttClient = new MQTTClientManager(mqttBrokerUrl, "data-processor"); this.analyticsScheduler = Executors.newScheduledThreadPool(2); this.deviceStats = new ConcurrentHashMap<>(); } public void start() { try { mqttClient.connect(); // Subscribe to telemetry from all devices mqttClient.subscribe("devices/+/telemetry", this::processTelemetry); mqttClient.subscribe("devices/+/state", this::processDeviceState); // Start periodic analytics startAnalyticsProcessing(); logger.info("IoT Data Processor started successfully"); } catch (Exception e) { logger.error("Failed to start IoT Data Processor", e); } } private void processTelemetry(String topic, String message) { try { ObjectNode telemetry = (ObjectNode) mapper.readTree(message); String deviceId = telemetry.get("deviceId").asText(); // Update device statistics DeviceStatistics stats = deviceStats.computeIfAbsent(deviceId, id -> new DeviceStatistics(id)); stats.update(telemetry); // Detect anomalies if (stats.detectAnomalies(telemetry)) { publishAnomalyAlert(deviceId, telemetry, "ANOMALY_DETECTED"); } // Check thresholds checkThresholds(deviceId, telemetry); } catch (Exception e) { logger.error("Error processing telemetry", e); } } private void processDeviceState(String topic, String message) { try { ObjectNode state = (ObjectNode) mapper.readTree(message); String deviceId = state.get("deviceId").asText(); boolean online = state.get("online").asBoolean(); if (!online) { publishDeviceAlert(deviceId, "DEVICE_OFFLINE", "Device is offline"); } } catch (Exception e) { logger.error("Error processing device state", e); } } private void checkThresholds(String deviceId, ObjectNode telemetry) { try { // Example threshold checks if (telemetry.has("currentTemperature")) { double temperature = telemetry.get("currentTemperature").asDouble(); if (temperature > 30.0) { publishDeviceAlert(deviceId, "HIGH_TEMPERATURE", String.format("Temperature too high: %.1f°C", temperature)); } else if (temperature < 10.0) { publishDeviceAlert(deviceId, "LOW_TEMPERATURE", String.format("Temperature too low: %.1f°C", temperature)); } } if (telemetry.has("battery")) { double battery = telemetry.get("battery").asDouble(); if (battery < 20.0) { publishDeviceAlert(deviceId, "LOW_BATTERY", String.format("Battery low: %.1f%%", battery)); } } } catch (Exception e) { logger.error("Error checking thresholds", e); } } private void publishAnomalyAlert(String deviceId, ObjectNode telemetry, String alertType) { try { ObjectNode alert = mapper.createObjectNode(); alert.put("deviceId", deviceId); alert.put("alertType", alertType); alert.put("severity", "HIGH"); alert.put("timestamp", System.currentTimeMillis()); alert.set("telemetry", telemetry); alert.put("description", "Anomaly detected in device telemetry"); mqttClient.publish("alerts/anomaly", alert.toString()); logger.warn("Anomaly alert published for device: {}", deviceId); } catch (Exception e) { logger.error("Error publishing anomaly alert", e); } } private void publishDeviceAlert(String deviceId, String alertType, String description) { try { ObjectNode alert = mapper.createObjectNode(); alert.put("deviceId", deviceId); alert.put("alertType", alertType); alert.put("severity", "MEDIUM"); alert.put("timestamp", System.currentTimeMillis()); alert.put("description", description); mqttClient.publish("alerts/device", alert.toString()); logger.warn("Device alert published: {} - {}", alertType, description); } catch (Exception e) { logger.error("Error publishing device alert", e); } } private void startAnalyticsProcessing() { // Generate analytics every minute analyticsScheduler.scheduleAtFixedRate(() -> { try { generateAggregatedMetrics(); generateDeviceHealthReport(); } catch (Exception e) { logger.error("Analytics processing failed", e); } }, 1, 1, TimeUnit.MINUTES); } private void generateAggregatedMetrics() { try { ObjectNode metrics = mapper.createObjectNode(); metrics.put("timestamp", System.currentTimeMillis()); metrics.put("totalDevices", deviceStats.size()); // Calculate aggregated metrics long onlineDevices = deviceStats.values().stream() .filter(DeviceStatistics::isOnline) .count(); metrics.put("onlineDevices", onlineDevices); double avgTemperature = deviceStats.values().stream() .filter(DeviceStatistics::hasTemperature) .mapToDouble(DeviceStatistics::getLastTemperature) .average() .orElse(0.0); metrics.put("averageTemperature", Math.round(avgTemperature * 10.0) / 10.0); mqttClient.publish("analytics/metrics", metrics.toString()); logger.debug("Published aggregated metrics"); } catch (Exception e) { logger.error("Error generating aggregated metrics", e); } } private void generateDeviceHealthReport() { try { ObjectNode healthReport = mapper.createObjectNode(); healthReport.put("timestamp", System.currentTimeMillis()); deviceStats.forEach((deviceId, stats) -> { ObjectNode deviceHealth = mapper.createObjectNode(); deviceHealth.put("online", stats.isOnline()); deviceHealth.put("lastSeen", stats.getLastUpdate()); deviceHealth.put("messageCount", stats.getMessageCount()); deviceHealth.put("alertCount", stats.getAlertCount()); healthReport.set(deviceId, deviceHealth); }); mqttClient.publish("analytics/health", healthReport.toString()); logger.debug("Published device health report"); } catch (Exception e) { logger.error("Error generating health report", e); } } public void stop() { analyticsScheduler.shutdown(); mqttClient.disconnect(); logger.info("IoT Data Processor stopped"); } // Device statistics class private static class DeviceStatistics { private final String deviceId; private volatile boolean online; private volatile long lastUpdate; private volatile long messageCount; private volatile long alertCount; private volatile double lastTemperature; private volatile boolean hasTemperature; public DeviceStatistics(String deviceId) { this.deviceId = deviceId; this.online = true; this.lastUpdate = System.currentTimeMillis(); } public void update(ObjectNode telemetry) { this.lastUpdate = System.currentTimeMillis(); this.messageCount++; this.online = true; if (telemetry.has("currentTemperature")) { this.lastTemperature = telemetry.get("currentTemperature").asDouble(); this.hasTemperature = true; } } public boolean detectAnomalies(ObjectNode telemetry) { // Simple anomaly detection based on rate of change if (telemetry.has("currentTemperature") && hasTemperature) { double currentTemp = telemetry.get("currentTemperature").asDouble(); double change = Math.abs(currentTemp - lastTemperature); // If temperature changes more than 5°C in one reading, consider it anomalous return change > 5.0; } return false; } // Getters public boolean isOnline() { // Consider device offline if no update in last 5 minutes return online && (System.currentTimeMillis() - lastUpdate) < (5 * 60 * 1000); } public long getLastUpdate() { return lastUpdate; } public long getMessageCount() { return messageCount; } public long getAlertCount() { return alertCount; } public double getLastTemperature() { return lastTemperature; } public boolean hasTemperature() { return hasTemperature; } } }

Complete IoT Application

Main IoT Application Class

package com.iot; import com.iot.devices.SmartThermostat; import com.iot.coap.CoAPDeviceServer; import com.iot.gateway.MQTTToCoAPBridge; import com.iot.analytics.IoTDataProcessor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class IoTApplication { private static final Logger logger = LoggerFactory.getLogger(IoTApplication.class); // Configuration private static final String MQTT_BROKER = "tcp://localhost:1883"; private static final String COAP_SERVER_HOST = "localhost"; private static final int COAP_SERVER_PORT = 5683; private final ScheduledExecutorService appScheduler; private SmartThermostat thermostat; private CoAPDeviceServer coapServer; private MQTTToCoAPBridge bridge; private IoTDataProcessor dataProcessor; public IoTApplication() { this.appScheduler = Executors.newScheduledThreadPool(3); } public void start() { logger.info("Starting IoT Application..."); try { // 1. Start CoAP Server for constrained devices startCoAPServer(); // 2. Start MQTT to CoAP bridge startBridge(); // 3. Start data processor startDataProcessor(); // 4. Start simulated smart devices startSmartDevices(); // 5. Schedule periodic tasks schedulePeriodicTasks(); logger.info("IoT Application started successfully"); } catch (Exception e) { logger.error("Failed to start IoT Application", e); stop(); } } private void startCoAPServer() { coapServer = new CoAPDeviceServer(COAP_SERVER_PORT); // Register some simulated CoAP devices coapServer.registerDevice("sensor-001", new CoAPDeviceServer.DeviceInfo("Temperature Sensor", "SENSOR", "IoT Corp")); coapServer.registerDevice("sensor-002", new CoAPDeviceServer.DeviceInfo("Humidity Sensor", "SENSOR", "IoT Corp")); coapServer.registerDevice("actuator-001", new CoAPDeviceServer.DeviceInfo("Smart Relay", "ACTUATOR", "Smart Devices Inc")); coapServer.start(); logger.info("CoAP Server started with {} devices", 3); } private void startBridge() { bridge = new MQTTToCoAPBridge(MQTT_BROKER, COAP_SERVER_HOST, COAP_SERVER_PORT); bridge.start(); logger.info("MQTT to CoAP Bridge started"); } private void startDataProcessor() { dataProcessor = new IoTDataProcessor(MQTT_BROKER); dataProcessor.start(); logger.info("IoT Data Processor started"); } private void startSmartDevices() { // Create and start a smart thermostat thermostat = new SmartThermostat("thermostat-livingroom", MQTT_BROKER); thermostat.start(); logger.info("Smart Thermostat started"); } private void schedulePeriodicTasks() { // Health check every 30 seconds appScheduler.scheduleAtFixedRate(this::healthCheck, 30, 30, TimeUnit.SECONDS); // System status report every 5 minutes appScheduler.scheduleAtFixedRate(this::systemStatusReport, 1, 5, TimeUnit.MINUTES); } private void healthCheck() { logger.debug("System health check - All components running"); // In a real application, this would check component health } private void systemStatusReport() { logger.info("=== IoT System Status Report ==="); logger.info("CoAP Server: Running on port {}", COAP_SERVER_PORT); logger.info("MQTT Bridge: Active"); logger.info("Data Processor: Active"); logger.info("Smart Devices: 1 thermostat active"); logger.info("================================="); } public void stop() { logger.info("Stopping IoT Application..."); if (thermostat != null) { thermostat.stop(); } if (bridge != null) { bridge.stop(); } if (dataProcessor != null) { dataProcessor.stop(); } if (coapServer != null) { coapServer.destroy(); } appScheduler.shutdown(); logger.info("IoT Application stopped"); } public static void main(String[] args) { IoTApplication app = new IoTApplication(); // Add shutdown hook for graceful shutdown Runtime.getRuntime().addShutdownHook(new Thread(app::stop)); // Start the application app.start(); // Keep the application running try { Thread.sleep(Long.MAX_VALUE); } catch (InterruptedException e) { Thread.currentThread().interrupt(); app.stop(); } } }

Docker Deployment

Docker Compose for IoT Stack

# docker-compose.yml version: '3.8' services: # MQTT Broker (Eclipse Mosquitto) mosquitto: image: eclipse-mosquitto:2.0.15 ports: - "1883:1883" # MQTT - "9001:9001" # Websockets volumes: - ./mosquitto.conf:/mosquitto/config/mosquitto.conf - mosquitto_data:/mosquitto/data - mosquitto_log:/mosquitto/log networks: - iot-network # CoAP Server (Custom Java Application) iot-application: build: . ports: - "5683:5683" # CoAP - "8080:8080" # HTTP (if needed) environment: - MQTT_BROKER_URL=tcp://mosquitto:1883 - COAP_SERVER_PORT=5683 depends_on: - mosquitto networks: - iot-network restart: unless-stopped volumes: mosquitto_data: mosquitto_log: networks: iot-network: driver: bridge

Dockerfile

# Dockerfile FROM openjdk:17-jdk-slim WORKDIR /app # Copy Maven executable and project files COPY target/iot-application.jar app.jar COPY config/application.properties config/ # Install any required native dependencies RUN apt-get update && apt-get install -y \ curl \ && rm -rf /var/lib/apt/lists/* # Expose ports EXPOSE 5683 8080 # Health check HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8080/health || exit 1 # Run the application ENTRYPOINT ["java", "-jar", "app.jar"]

Conclusion

This comprehensive Eclipse IoT implementation provides:

Key Components:

• MQTT Communication with Eclipse Paho for device telemetry and commands
• CoAP Support with Eclipse Californium for constrained devices
• Protocol Bridge for MQTT-CoAP interoperability
• Real-time Analytics for data processing and anomaly detection
• Device Management for smart device simulation and control

Eclipse IoT Technologies Used:

1. Eclipse Paho - MQTT client implementation
2. Eclipse Mosquitto - MQTT broker (deployment)
3. Eclipse Californium - CoAP framework
4. Eclipse Ditto - Digital twins (concept)

Best Practices:

1. Error Handling - Comprehensive exception handling and reconnection logic
2. Resource Management - Proper cleanup of connections and threads
3. Observability - Extensive logging and health monitoring
4. Protocol Compliance - Proper implementation of MQTT and CoAP standards
5. Security - Authentication and TLS support (extensible)

Use Cases:

• Smart home automation
• Industrial IoT monitoring
• Environmental sensing
• Energy management systems
• Asset tracking solutions

This implementation provides a solid foundation for building enterprise-grade IoT solutions using Java and Eclipse IoT technologies.