Admin server feature

build.gradle

@@ -13,11 +13,28 @@ subprojects {
     }
 
     dependencies {
+        if (project.name != 'desm-core') {
+            implementation project(':desm-core')
+        }
+
+        // Testing
         testImplementation platform('org.junit:junit-bom:5.10.0')
         testImplementation 'org.junit.jupiter:junit-jupiter'
         testImplementation 'org.mockito:mockito-core:5.17.0'
         testImplementation 'org.mockito:mockito-junit-jupiter:5.17.0'
         testImplementation 'org.mockito:mockito-inline:5.2.0'
+
+        // Lombok
+        compileOnly 'org.projectlombok:lombok:1.18.38'
+        annotationProcessor 'org.projectlombok:lombok:1.18.38'
+        testCompileOnly 'org.projectlombok:lombok:1.18.38'
+        testAnnotationProcessor 'org.projectlombok:lombok:1.18.38'
+
+        // Gson
+        implementation 'com.google.code.gson:gson:2.13.1'
+
+        // MQTT
+        implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
     }
 
     repositories {

desm-core/src/main/java/org/example/desm/common/Constant.java

@@ -0,0 +1,14 @@
+package org.example.desm.common;
+
+public class Constant {
+    public static final String MQTT_BROKER_ADDRESS = "tcp://localhost:1883";
+
+    // MQTT topics
+    public static final String TOPIC_ENERGY_REQUEST = "energy/request";
+    public static final String TOPIC_CO2_MEASUREMENT = "co2/measurement";
+
+    // API endpoints
+    public static final String ENDPOINT_PLANTS = "/plants";
+    public static final String ENDPOINT_REGISTER_PLANT = "/register";
+    public static final String ENDPOINT_POLLUTION = "/pollution";
+}

desm-core/src/main/java/org/example/desm/common/MyReadWriteLock.java

@@ -0,0 +1,100 @@
+package org.example.desm.common;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+/**
+ * Personal implementation of a Read-Write Lock using a FIFO queue.
+ * <p>
+ * This lock ensures that:
+ * <ul>
+ *   <li>Multiple readers can hold the lock simultaneously if no writer is active or waiting before them.</li>
+ *   <li>Writers have exclusive access, blocking both readers and other writers.</li>
+ *   <li>Requests are processed in FIFO order, avoiding readers or writers starvation exploiting fairness.</li>
+ * </ul>
+ * <p>
+ * Each thread adds a request to a queue and waits until it reaches the head and the lock is available.
+ */
+public class MyReadWriteLock {
+    /**
+     * A support class that represents a request to acquire a lock, either read or write.
+     */
+    private static final class Request {
+        // Support class
+    }
+
+    private int activeReaders;
+    private boolean isWriting;
+    private final Queue<Request> queue;
+
+    public MyReadWriteLock() {
+        this.activeReaders = 0;
+        this.isWriting = false;
+        this.queue = new LinkedList<>();
+    }
+
+    /**
+     * Acquires the read lock. Multiple threads can acquire the read lock as long as:
+     * <ul>
+     *   <li>There is no writer currently holding the lock.</li>
+     *   <li>The current read request is at the head of the queue.</li>
+     * </ul>
+     *
+     * @throws InterruptedException if the thread is interrupted while waiting.
+     */
+    public synchronized void lockRead() throws InterruptedException {
+        Request request = new Request();
+        queue.add(request);
+
+        while (queue.peek() != request || isWriting) {
+            wait();
+        }
+
+        queue.poll();
+        activeReaders++;
+        notifyAll();
+    }
+
+    /**
+     * Releases the read lock. If no more readers are present, waiting writers
+     * may be notified.
+     */
+    public synchronized void unlockRead() {
+        activeReaders--;
+        if (activeReaders == 0) {
+            notifyAll();
+        }
+    }
+
+    /**
+     * Acquires the write lock. This method blocks until:
+     * <ul>
+     *   <li>The current write request is at the head of the queue.</li>
+     *   <li>There are no active readers.</li>
+     *   <li>There is no other writer holding the lock.</li>
+     * </ul>
+     *
+     * @throws InterruptedException if the thread is interrupted while waiting.
+     */
+    public synchronized void lockWrite() throws InterruptedException {
+        Request request = new Request();
+        queue.add(request);
+
+        while (queue.peek() != request || isWriting || activeReaders > 0) {
+            wait();
+        }
+
+        queue.poll();
+        isWriting = true;
+        notifyAll();
+    }
+
+    /**
+     * Releases the write lock. Notifies waiting readers or writers so they
+     * can attempt to acquire the lock.
+     */
+    public synchronized void unlockWrite() {
+        isWriting = false;
+        notifyAll();
+    }
+}

desm-core/src/main/java/org/example/desm/common/SlidingWindowBuffer.java

@@ -0,0 +1,48 @@
+package org.example.desm.common;
+
+import org.example.desm.common.simulator.Buffer;
+import org.example.desm.common.simulator.Measurement;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SlidingWindowBuffer implements Buffer {
+    private static final int WINDOW_SIZE = 8;
+    private static final int SLIDE_SIZE = WINDOW_SIZE / 2;
+
+    private final String plantId;
+    private final List<Measurement> buffer;
+    private final List<Measurement> computedAverages;
+
+    public SlidingWindowBuffer(String plantId) {
+        this.plantId = plantId;
+        this.buffer = new ArrayList<>();
+        this.computedAverages = new ArrayList<>();
+    }
+
+    @Override
+    public synchronized void addMeasurement(Measurement m) {
+        buffer.add(m);
+        if (buffer.size() == WINDOW_SIZE) {
+            Measurement computedAverage = computeAverage();
+            computedAverages.add(computedAverage);
+            buffer.subList(0, SLIDE_SIZE).clear();
+        }
+    }
+
+    @Override
+    public synchronized List<Measurement> readAllAndClean() {
+        List<Measurement> result = List.copyOf(computedAverages);
+        computedAverages.clear();
+        return result;
+    }
+
+    private Measurement computeAverage() {
+        double sum = 0;
+        for (Measurement measurement : buffer) {
+            sum += measurement.getValue();
+        }
+        double avg = sum / buffer.size();
+        return new Measurement(plantId, "CO2_AVG", avg, System.currentTimeMillis());
+    }
+}

desm-core/src/main/java/org/example/desm/common/model/AverageEmission.java

@@ -0,0 +1,14 @@
+package org.example.desm.common.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class AverageEmission {
+    private long startTimestamp;
+    private long endTimestamp;
+    private double average_co2;
+}

desm-core/src/main/java/org/example/desm/common/model/ErrorResponse.java

@@ -0,0 +1,14 @@
+package org.example.desm.common.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class ErrorResponse {
+    private int status;
+    private String error;
+    private String message;
+}

desm-core/src/main/java/org/example/desm/common/model/MeasurementsWrapper.java

@@ -0,0 +1,16 @@
+package org.example.desm.common.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import org.example.desm.common.simulator.Measurement;
+
+import java.util.List;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class MeasurementsWrapper {
+    private List<Measurement> measurements;
+}

desm-core/src/main/java/org/example/desm/common/model/PowerPlant.java

@@ -0,0 +1,14 @@
+package org.example.desm.common.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class PowerPlant {
+    private String id;
+    private String address;
+    private int port;
+}

desm-core/src/main/java/org/example/desm/common/model/PowerRequest.java

@@ -0,0 +1,13 @@
+package org.example.desm.common.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+public class PowerRequest {
+    private int power;
+    private long timestamp;
+}

desm-core/src/main/java/org/example/desm/common/simulator/Buffer.java

@@ -0,0 +1,8 @@
+package org.example.desm.common.simulator;
+
+import java.util.List;
+
+public interface Buffer {
+    void addMeasurement(Measurement m);
+    List<Measurement> readAllAndClean();
+}

desm-core/src/main/java/org/example/desm/common/simulator/Measurement.java

@@ -0,0 +1,25 @@
+package org.example.desm.common.simulator;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@AllArgsConstructor
+@Data
+public class Measurement implements Comparable<Measurement> {
+    private String id;
+    private String type;
+    private double value;
+    private long timestamp;
+
+    @Override
+    public int compareTo(Measurement m) {
+        Long thisTimestamp = timestamp;
+        Long otherTimestamp = m.getTimestamp();
+        return thisTimestamp.compareTo(otherTimestamp);
+    }
+
+    @Override
+    public String toString(){
+        return value + " " + timestamp;
+    }
+}

desm-core/src/main/java/org/example/desm/common/simulator/PollutionSensor.java

@@ -0,0 +1,37 @@
+package org.example.desm.common.simulator;
+
+public class PollutionSensor extends Simulator {
+    private static final int MEAN = 125000;
+    private static final int VARIANCE = 5000;
+
+    private static int ID = 1;
+
+    public PollutionSensor(String id, Buffer buffer) {
+        super(id, "CO2", buffer);
+    }
+
+    // Use this constructor to initialize the Pollution Sensor simulator in your project
+    public PollutionSensor(Buffer buffer) {
+        this("CO2-" + ID++, buffer);
+    }
+
+    @Override
+    public void run() {
+        double i = rnd.nextInt();
+        long waitingTime;
+        while(!stopCondition) {
+            double co2 = getCO2Value(i);
+            addMeasurement(co2);
+
+            waitingTime = 2000;
+            sensorSleep(waitingTime);
+
+            i += 0.2;
+        }
+    }
+
+    private double getCO2Value(double t) {
+        double gaussian = rnd.nextGaussian();
+        return MEAN + Math.sqrt(VARIANCE) * gaussian;
+    }
+}