[WIP] Series, Entry, Message

pom.xml

@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>de.ph87.data</groupId>
+  <artifactId>Data</artifactId>
+  <version>0.1.0-SNAPSHOT</version>
+
+  <properties>
+    <maven.compiler.source>21</maven.compiler.source>
+    <maven.compiler.target>21</maven.compiler.target>
+    <maven.compiler.release>21</maven.compiler.release>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <parent>
+    <groupId>org.springframework.boot</groupId>
+    <artifactId>spring-boot-starter-parent</artifactId>
+    <version>3.4.2</version>
+  </parent>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-web</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-data-jpa</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.postgresql</groupId>
+      <artifactId>postgresql</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.h2database</groupId>
+      <artifactId>h2</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.eclipse.paho</groupId>
+      <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+      <version>1.2.5</version>
+    </dependency>
+
+  </dependencies>
+
+</project>

src/main/java/de/ph87/data/Backend.java

@@ -0,0 +1,13 @@
+package de.ph87.data;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class Backend {
+
+  public static void main(String[] args) {
+    SpringApplication.run(Backend.class, args);
+  }
+
+}

src/main/java/de/ph87/data/message/IMessageHandler.java

@@ -0,0 +1,9 @@
+package de.ph87.data.message;
+
+import lombok.NonNull;
+
+public interface IMessageHandler {
+
+  void handle(@NonNull final Message message);
+
+}

src/main/java/de/ph87/data/message/Message.java

@@ -0,0 +1,26 @@
+package de.ph87.data.message;
+
+import lombok.Getter;
+import lombok.ToString;
+
+import java.time.ZonedDateTime;
+
+@Getter
+@ToString
+public class Message {
+
+  public final ZonedDateTime date = ZonedDateTime.now();
+
+  public final String topic;
+
+  public final String payload;
+
+  public final String payloadLoggable;
+
+  public Message(final String topic, final String payload) {
+    this.topic = topic;
+    this.payload = payload;
+    this.payloadLoggable = payload.replace("\n", "\\n").replace("\r", "\\r");
+  }
+
+}

src/main/java/de/ph87/data/message/MessageService.java

@@ -0,0 +1,27 @@
+package de.ph87.data.message;
+
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class MessageService {
+
+  private final List<IMessageHandler> messageHandlers;
+
+  public void onMessage(@NonNull final Message message) {
+    messageHandlers.forEach(handler -> {
+      try {
+        handler.handle(message);
+      } catch (Exception e) {
+        log.warn("Message handler error: message={}", e.getMessage(), e);
+      }
+    });
+  }
+
+}

src/main/java/de/ph87/data/message/handler/SimpleJsonHandler.java

@@ -0,0 +1,58 @@
+package de.ph87.data.message.handler;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import de.ph87.data.message.IMessageHandler;
+import de.ph87.data.message.Message;
+import de.ph87.data.series.SeriesInbound;
+import de.ph87.data.series.SeriesService;
+import de.ph87.data.unit.Unit;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class SimpleJsonHandler implements IMessageHandler {
+
+  private final ObjectMapper objectMapper;
+
+  private final SeriesService seriesService;
+
+  @Override
+  public void handle(@NonNull final Message message) {
+    try {
+      final Inbound inbound = objectMapper.readValue(message.payload, Inbound.class);
+      seriesService.receive(new SeriesInbound(message.topic, inbound.date, inbound.value, inbound.unit));
+    } catch (JsonProcessingException e) {
+      log.debug("Failed to parse inbound message: topic={}, message={}, error={}", message.topic, message, e.toString());
+    }
+  }
+
+  @Getter
+  @ToString
+  private static class Inbound {
+
+    public final ZonedDateTime date;
+
+    public final double value;
+
+    public final Unit unit;
+
+    public Inbound(final long timestamp, final double value, final Unit unit) {
+      this.date = ZonedDateTime.ofInstant(Instant.ofEpochSecond(timestamp), ZoneId.systemDefault());
+      this.value = value;
+      this.unit = unit;
+    }
+
+  }
+
+}

src/main/java/de/ph87/data/series/Action.java

@@ -0,0 +1,5 @@
+package de.ph87.data.series;
+
+public enum Action {
+  CREATED, CHANGED, DELETED
+}

src/main/java/de/ph87/data/series/Series.java

@@ -0,0 +1,34 @@
+package de.ph87.data.series;
+
+import de.ph87.data.unit.Unit;
+import jakarta.persistence.*;
+import lombok.*;
+
+import java.util.UUID;
+
+@Entity
+@Getter
+@ToString
+@NoArgsConstructor
+public class Series {
+
+  @Id
+  private String uuid = UUID.randomUUID().toString();
+
+  @Setter
+  @NonNull
+  @Column(nullable = false, unique = true)
+  private String title;
+
+  @Setter
+  @NonNull
+  @Column(nullable = false)
+  @Enumerated(EnumType.STRING)
+  private Unit unit;
+
+  @Setter
+  @NonNull
+  @Column(nullable = false, unique = true)
+  private String source;
+
+}

src/main/java/de/ph87/data/series/SeriesDto.java

@@ -0,0 +1,27 @@
+package de.ph87.data.series;
+
+import de.ph87.data.unit.Unit;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.ToString;
+
+@Getter
+@ToString
+public class SeriesDto {
+
+  public final String uuid;
+
+  public final String title;
+
+  public final Unit unit;
+
+  public final String source;
+
+  public SeriesDto(@NonNull final Series series) {
+    this.uuid = series.getUuid();
+    this.title = series.getTitle();
+    this.unit = series.getUnit();
+    this.source = series.getSource();
+  }
+
+}

src/main/java/de/ph87/data/series/SeriesInbound.java

@@ -0,0 +1,32 @@
+package de.ph87.data.series;
+
+import de.ph87.data.unit.Unit;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.ToString;
+
+import java.time.ZonedDateTime;
+
+@Getter
+@ToString
+public class SeriesInbound {
+
+  @NonNull
+  public final String name;
+
+  @NonNull
+  public final ZonedDateTime date;
+
+  public final double value;
+
+  @NonNull
+  public final Unit unit;
+
+  public SeriesInbound(@NonNull final String name, @NonNull final ZonedDateTime date, final double value, @NonNull final Unit unit) {
+    this.name = name;
+    this.date = date;
+    this.value = value;
+    this.unit = unit;
+  }
+
+}

src/main/java/de/ph87/data/series/SeriesRepository.java

@@ -0,0 +1,7 @@
+package de.ph87.data.series;
+
+import org.springframework.data.repository.ListCrudRepository;
+
+public interface SeriesRepository extends ListCrudRepository<Series, String> {
+
+}

src/main/java/de/ph87/data/series/SeriesService.java

@@ -0,0 +1,81 @@
+package de.ph87.data.series;
+
+import de.ph87.data.series.entry.EntryService;
+import jakarta.annotation.PostConstruct;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+@Slf4j
+@Service
+@Transactional
+@RequiredArgsConstructor
+public class SeriesService {
+
+  private final Map<String, String> sources = new HashMap<>();
+
+  private final SeriesRepository seriesRepository;
+
+  private final EntryService entryService;
+
+  @PostConstruct
+  public void cacheInit() {
+    synchronized (sources) {
+      seriesRepository.findAll().forEach(this::cachePut);
+    }
+  }
+
+  private void cachePut(@NonNull final Series series) {
+    synchronized (sources) {
+      sources.put(series.getUuid(), series.getSource());
+    }
+  }
+
+  private void cacheRemove(final Series series) {
+    synchronized (sources) {
+      sources.remove(series.getUuid());
+    }
+  }
+
+  public SeriesDto modify(@NonNull final String uuid, @NonNull final Consumer<Series> modifier) {
+    final Series series = getByUuid(uuid);
+    modifier.accept(series);
+    cachePut(series);
+    return publish(series, Action.CHANGED);
+  }
+
+  public void delete(@NonNull final String uuid) {
+    final Series series = getByUuid(uuid);
+    cacheRemove(series);
+  }
+
+  private Series getByUuid(@NonNull final String uuid) {
+    return seriesRepository.findById(uuid).orElseThrow();
+  }
+
+  private SeriesDto publish(@NonNull final Series series, @NonNull final Action action) {
+    final SeriesDto dto = toDto(series);
+    log.info("Series {}: {}", action, series);
+    return dto;
+  }
+
+  private SeriesDto toDto(@NonNull final Series series) {
+    return new SeriesDto(series);
+  }
+
+  public void receive(@NonNull final SeriesInbound measure) {
+    final String uuid = sources.get(measure.name);
+    if (uuid == null) {
+      return;
+    }
+    final Series series = getByUuid(uuid);
+    entryService.write(series, measure);
+  }
+
+}

src/main/java/de/ph87/data/series/entry/Entry.java

@@ -0,0 +1,40 @@
+package de.ph87.data.series.entry;
+
+import de.ph87.data.series.Series;
+import jakarta.persistence.*;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+import lombok.ToString;
+
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
+
+@Entity
+@Getter
+@ToString
+@NoArgsConstructor
+public class Entry {
+
+  @Id
+  @GeneratedValue(strategy = GenerationType.IDENTITY)
+  private long id;
+
+  @NonNull
+  @ManyToOne
+  private Series series;
+
+  @NonNull
+  @Column(nullable = false)
+  private ZonedDateTime date;
+
+  @Column(nullable = false)
+  private double value;
+
+  public Entry(@NonNull final Series series, @NonNull final ZonedDateTime date, final double value) {
+    this.series = series;
+    this.date = date.truncatedTo(ChronoUnit.MINUTES);
+    this.value = value;
+  }
+
+}

src/main/java/de/ph87/data/series/entry/EntryRepository.java

@@ -0,0 +1,7 @@
+package de.ph87.data.series.entry;
+
+import org.springframework.data.repository.ListCrudRepository;
+
+public interface EntryRepository extends ListCrudRepository<Entry, Long> {
+
+}

src/main/java/de/ph87/data/series/entry/EntryService.java

@@ -0,0 +1,23 @@
+package de.ph87.data.series.entry;
+
+import de.ph87.data.series.Series;
+import de.ph87.data.series.SeriesInbound;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Slf4j
+@Service
+@Transactional
+@RequiredArgsConstructor
+public class EntryService {
+
+  private final EntryRepository entryRepository;
+
+  public void write(@NonNull final Series series, @NonNull final SeriesInbound measure) {
+    // TODO
+  }
+
+}

src/main/java/de/ph87/data/unit/Unit.java

@@ -0,0 +1,38 @@
+package de.ph87.data.unit;
+
+import lombok.NonNull;
+
+public enum Unit {
+  TEMPERATURE_C("°C"),
+  PRESSURE_HPA("hPa"),
+  HUMIDITY_RELATIVE_PERCENT("%"),
+  HUMIDITY_ABSOLUTE_MGL("mg/L"),
+  HUMIDITY_ABSOLUTE_GM3("g/m³", 1, HUMIDITY_ABSOLUTE_MGL),
+  ILLUMINANCE_LUX("lux"),
+  RESISTANCE_OHMS("Ω"),
+  ALTITUDE_M("m"),
+  POWER_W("W"),
+  POWER_KW("kW", 1000, POWER_W),
+  ENERGY_WH("W"),
+  ENERGY_KWH("kWh", 1000, ENERGY_WH),
+  ;
+
+  public final String unit;
+
+  public final double factor;
+
+  public final Unit base;
+
+  Unit(@NonNull final String unit) {
+    this.unit = unit;
+    this.factor = 1.0;
+    this.base = this;
+  }
+
+  Unit(@NonNull final String unit, double factor, @NonNull Unit base) {
+    this.unit = unit;
+    this.factor = factor;
+    this.base = base;
+  }
+
+}