IBM · sreckoj · Jul 1, 2024 · Jul 1, 2024
diff --git a/.gitignore b/.gitignore
@@ -4,3 +4,6 @@
 .vscode
 target/
 .DS_Store
+.gitignore
+lib/example_surveyadminservice.yaml
+lib/example_surveygeocoderservice.yaml
diff --git a/lib/example_surveyadminkafkasource.yaml b/lib/example_surveyadminkafkasource.yaml
@@ -4,7 +4,7 @@ metadata:
   name: geocodetopicsource
 spec:
   bootstrapServers:
-  - my-cluster-kafka-bootstrap.amq-streams-kafka.svc:9092
+  - es1-kafka-bootstrap.cp4i.svc:9092
   sink:
     ref:
       apiVersion: serving.knative.dev/v1

diff --git a/lib/example_surveyadminservice.yaml.template b/lib/example_surveyadminservice.yaml.template
@@ -16,7 +16,7 @@ spec:
         imagePullPolicy: Always
         env:
         - name: mp.messaging.connector.liberty-kafka.bootstrap.servers
-          value: my-cluster-kafka-bootstrap.amq-streams-kafka.svc:9092
+          value: es1-kafka-bootstrap.cp4i.svc:9092
         - name: GOOGLE_API_KEY
           value: INSERT_API_KEY
         - name: QRCODE_URL

diff --git a/lib/example_surveygeocoderkafkasource.yaml b/lib/example_surveygeocoderkafkasource.yaml
@@ -4,7 +4,7 @@ metadata:
   name: locationtopicsource
 spec:
   bootstrapServers:
-  - my-cluster-kafka-bootstrap.amq-streams-kafka.svc:9092
+  - es1-kafka-bootstrap.cp4i.svc:9092
   sink:
     ref:
       apiVersion: serving.knative.dev/v1

diff --git a/lib/example_surveygeocoderservice.yaml.template b/lib/example_surveygeocoderservice.yaml.template
@@ -15,7 +15,7 @@ spec:
         imagePullPolicy: Always
         env:
         - name: mp.messaging.connector.liberty-kafka.bootstrap.servers
-          value: my-cluster-kafka-bootstrap.amq-streams-kafka.svc:9092
+          value: es1-kafka-bootstrap.cp4i.svc:9092
         - name: GOOGLE_API_KEY
           value: INSERT_API_KEY
         securityContext:

diff --git a/lib/example_surveyinputservice.yaml b/lib/example_surveyinputservice.yaml
@@ -15,7 +15,7 @@ spec:
         imagePullPolicy: Always
         env:
         - name: mp.messaging.connector.liberty-kafka.bootstrap.servers
-          value: my-cluster-kafka-bootstrap.amq-streams-kafka.svc:9092
+          value: es1-kafka-bootstrap.cp4i.svc:9092
         securityContext:
           allowPrivilegeEscalation: true
           privileged: false

diff --git a/surveyAdminService/pom.xml b/surveyAdminService/pom.xml
@@ -103,6 +103,16 @@ under the License.
       <artifactId>lucene-core</artifactId>
       <version>${dependency.version.lucene-core}</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+      <version>2.10.1</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>2.8.0</version>
+    </dependency>
   </dependencies>
   <build>
     <finalName>${project.artifactId}</finalName>

diff --git a/surveyAdminService/src/main/java/com/example/demo/reactive/CloudEventsProcessor.java b/surveyAdminService/src/main/java/com/example/demo/reactive/CloudEventsProcessor.java
@@ -24,6 +24,8 @@
 
 import com.example.demo.Configuration;
 import com.example.demo.websockets.GeolocationWebSocket;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.CloudEventData;
@@ -55,26 +57,27 @@ public Response geocodeComplete(CloudEvent incoming) {
 		if (LOG.isLoggable(Level.INFO))
 			LOG.info("CloudEventData: " + data);
 
-		String geocodeResults = null;
+		String jsonString = null;
 		try (StringDeserializer deserializer = new StringDeserializer()) {
-			geocodeResults = deserializer.deserialize(null, data.toBytes());
+			jsonString = deserializer.deserialize(null, data.toBytes());
 		}
 
 		if (LOG.isLoggable(Level.INFO))
-			LOG.info("Geocode results: " + geocodeResults);
+			LOG.info("Geocode results: " + jsonString);
 
-		// latitude ' ' longitude ' ' location
-		String latitudeStr = geocodeResults.substring(0, geocodeResults.indexOf(' '));
-		geocodeResults = geocodeResults.substring(geocodeResults.indexOf(' ') + 1);
-		String longitudeStr = geocodeResults.substring(0, geocodeResults.indexOf(' '));
-		geocodeResults = geocodeResults.substring(geocodeResults.indexOf(' ') + 1);
+		JsonObject jsonObj = (new Gson()).fromJson(jsonString, JsonObject.class);
+		String latitudeStr = jsonObj.get("latitude").getAsString();
+		String longitudeStr = jsonObj.get("longitude").getAsString();
+		String location = jsonObj.get("location").getAsString();
+		String color = jsonObj.get("color").getAsString();
+		String key = jsonObj.get("key").getAsString();
 
 		double latitude = Double.parseDouble(latitudeStr);
 		double longitude = Double.parseDouble(longitudeStr);
 		double approximateDistance = SloppyMath.haversinMeters(latitude, longitude, Configuration.getSurveyLatitude(),
 				Configuration.getSuveyLongitude());
 
-		String finalResults = latitude + " " + longitude + " " + approximateDistance + " " + geocodeResults;
+		String finalResults = latitude + " " + longitude + " " + approximateDistance + " " + color + " " + key + " " + location;
 
 		try {
 			GeolocationWebSocket.sendMessageToAllBrowsers(finalResults);

diff --git a/surveyAdminService/src/main/java/com/example/demo/websockets/GeolocationWebSocket.java b/surveyAdminService/src/main/java/com/example/demo/websockets/GeolocationWebSocket.java
@@ -1,10 +1,17 @@
 package com.example.demo.websockets;
 
 import java.io.IOException;
+import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
 import jakarta.websocket.EncodeException;
 import jakarta.websocket.OnClose;
 import jakarta.websocket.OnError;
@@ -51,7 +58,11 @@ public void onBrowserMessage(Session session, String message) {
 
 			if (LOG.isLoggable(Level.INFO))
 				LOG.info("GeolocationWebSocket received message from " + session.getId() + ": " + message);
+
+			// Send message from websocket to Kafka topic
+			sendToKafka(message, "feedbacktopic");
 		}
+
 	}
 
 	@OnClose
@@ -106,4 +117,18 @@ public static final void sendMessageToAllBrowsers(String text) throws IOExceptio
 			}
 		}
 	}
+
+	private void sendToKafka(String message, String topic) {
+		String bootstrapServer = System.getenv("mp.messaging.connector.liberty-kafka.bootstrap.servers");
+
+		Properties props = new Properties();
+		props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
+		props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+		KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
+		ProducerRecord<String, String> record = new ProducerRecord<String,String>(topic, message);
+		producer.send(record);
+		producer.flush();
+		producer.close();
+	}
 }