implemented changes to allow MapResource to consume events directly f…

…rom topic and dispatch to websocket endpoint
gunnarmorling · Nov 14, 2020 · f8eb772 · f8eb772
1 parent 2c434f3
commit f8eb772
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 9 deletions.
diff --git a/temperature-map/src/main/java/eu/javaland/knative/MapResource.java b/temperature-map/src/main/java/eu/javaland/knative/MapResource.java
@@ -18,9 +18,15 @@
 
 import io.quarkus.runtime.StartupEvent;
 
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+
 @Path("/")
 public class MapResource {
 
+    @ConfigProperty(name = "map.read.from.topic", defaultValue = "false")
+    boolean readFromTopic;
+
     @Inject
     MapEndpoint websocketEndpoint;
 
@@ -38,6 +44,22 @@ public Response newMeasurement(String request) {
 
         JsonReader reader = Json.createReader(new StringReader(request));
         JsonObject measurement = reader.readObject();
+
+        websocketEndpoint.onMeasurement(initializeMeasurementFromJson(measurement));
+
+        return Response.ok().build();
+    }
+
+    @Incoming("temperature-values")
+    public void onMeasurement(JsonObject measurement) {
+        if (!readFromTopic) {
+            return;
+        }
+
+        websocketEndpoint.onMeasurement(initializeMeasurementFromJson(measurement));
+    }
+
+    private Measurement initializeMeasurementFromJson(JsonObject measurement) {
         Measurement temperatureMeasurement = new Measurement();
 
         temperatureMeasurement.stationId = measurement.getInt("stationId");
@@ -51,10 +73,7 @@ public Response newMeasurement(String request) {
         temperatureMeasurement.ts = measurement.getString("ts");
         temperatureMeasurement.value = measurement.getJsonNumber("value").doubleValue();
         temperatureMeasurement.icon = measurement.getString("icon");
-
-        websocketEndpoint.onMeasurement(temperatureMeasurement);
-
-        return Response.ok().build();
+        return temperatureMeasurement;
     }
 
     public void onStartup(@Observes StartupEvent se) {

diff --git a/temperature-map/src/main/resources/application.properties b/temperature-map/src/main/resources/application.properties
@@ -1,5 +1,5 @@
-# mp.messaging.incoming.temperature-values.connector=smallrye-kafka
-# mp.messaging.incoming.temperature-values.topic=temperature-values-enriched
-# mp.messaging.incoming.temperature-values.key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
-# mp.messaging.incoming.temperature-values.value.deserializer=eu.javaland.knative.JsonpDeserializer
-# mp.messaging.incoming.temperature-values.auto.offset.reset=latest
+mp.messaging.incoming.temperature-values.connector=smallrye-kafka
+mp.messaging.incoming.temperature-values.topic=temperature-values-enriched
+mp.messaging.incoming.temperature-values.key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
+mp.messaging.incoming.temperature-values.value.deserializer=eu.javaland.knative.JsonpDeserializer
+mp.messaging.incoming.temperature-values.auto.offset.reset=latest