5 files changed, 95 insertions, 0 deletions
diff --git a/jws/kafka/src/main/java/fr/epita/assistants/StringInfoProcessor.java b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoProcessor.java
new file mode 100644
index 0000000..b419623
--- /dev/null
+++ b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoProcessor.java
@@ -0,0 +1,41 @@
+package fr.epita.assistants;
+
+import fr.epita.assistants.utils.StringInfo;
+import io.smallrye.reactive.messaging.annotations.Broadcast;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+
+public class StringInfoProcessor {
+    private boolean isPalindrome(String word) {
+        if (word == null) return false;
+        word = word.toLowerCase().strip();
+        if (word.isEmpty()) return true;
+        for (int i = 0, j = word.length() - 1; i < j; i++, j--) {
+            while (word.charAt(i) == ' ') {
+                i++;
+            }
+            while (word.charAt(j) == ' ') {
+                j--;
+            }
+            if (word.charAt(i) != word.charAt(j)) return false;
+        }
+        return true;
+    }
+
+    @Incoming("string-info-command")
+    @Outgoing("string-info-aggregate")
+    @Broadcast
+    public StringInfo process(String s) {
+        String tmp = s.toLowerCase();
+        int v = 0;
+        int c = 0;
+        for (int i = 0; i < tmp.length(); i++) {
+            if (tmp.charAt(i) == 'a' || tmp.charAt(i) == 'e' || tmp.charAt(i) == 'i' || tmp.charAt(i) == 'o' || tmp.charAt(i) == 'u' || tmp.charAt(i) == 'y')
+                v++;
+            else if (tmp.charAt(i) <= 'z' && tmp.charAt(i) >= 'a')
+                c++;
+        }
+        return new StringInfo(s, v, c, isPalindrome(s));
+    }
+
+}
+\ No newline at end of file
diff --git a/jws/kafka/src/main/java/fr/epita/assistants/StringInfoResource.java b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoResource.java
new file mode 100644
index 0000000..62a5d51
--- /dev/null
+++ b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoResource.java
@@ -0,0 +1,25 @@
+package fr.epita.assistants;
+
+import io.smallrye.reactive.messaging.annotations.Broadcast;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import org.eclipse.microprofile.reactive.messaging.Channel;
+import org.eclipse.microprofile.reactive.messaging.Emitter;
+
+import java.util.concurrent.CompletionStage;
+
+@Path("/send")
+@Consumes
+public class StringInfoResource {
+    @Inject
+    @Broadcast
+    @Channel("string-info-command")
+    Emitter<String> commandEmitter;
+
+    @POST
+    public void send(String s) {
+        CompletionStage<Void> v = commandEmitter.send(s);
+    }
+}
+\ No newline at end of file
diff --git a/jws/kafka/src/main/java/fr/epita/assistants/StringInfoSubscriber.java b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoSubscriber.java
new file mode 100644
index 0000000..c955270
--- /dev/null
+++ b/jws/kafka/src/main/java/fr/epita/assistants/StringInfoSubscriber.java
@@ -0,0 +1,13 @@
+package fr.epita.assistants;
+
+import fr.epita.assistants.utils.StringInfo;
+import io.smallrye.reactive.messaging.annotations.Broadcast;
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+
+public class StringInfoSubscriber {
+    @Incoming("string-info-aggregate")
+    @Broadcast
+    public void subscribe(StringInfo si) {
+        System.out.println(si.toString());
+    }
+}
+\ No newline at end of file
diff --git a/jws/kafka/src/main/java/fr/epita/assistants/utils/StringInfo.java b/jws/kafka/src/main/java/fr/epita/assistants/utils/StringInfo.java
new file mode 100644
index 0000000..c328ac3
--- /dev/null
+++ b/jws/kafka/src/main/java/fr/epita/assistants/utils/StringInfo.java
@@ -0,0 +1,13 @@
+package fr.epita.assistants.utils;
+
+import lombok.ToString;
+import lombok.Value;
+
+@Value
+@ToString
+public class StringInfo {
+    String word;
+    Integer nbVowels;
+    Integer nbConsonants;
+    Boolean isPalindrome;
+}
diff --git a/jws/kafka/src/main/resources/application.properties b/jws/kafka/src/main/resources/application.properties
new file mode 100644
index 0000000..9db5e48
--- /dev/null
+++ b/jws/kafka/src/main/resources/application.properties
@@ -0,0 +1,3 @@
+quarkus.devservices.enabled=true
+quarkus.kafka.devservices.image-name=reg.undercloud.cri.epita.fr/docker/redpandadata/redpanda:v24.1.2
+%test.quarkus.devservices.enabled=false