MACRO NEPAL
Java IoT with Eclipse IoT: Complete Guide for Connected Devices
awiskar acharya
November 16, 2025
6:41 am

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.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper