Merge pull request #15 from grookage/messageProcessor

MessageProcessor
grookage · Dec 12, 2024 · f6cb05e · f6cb05e
2 parents dcbcefd + 20ff0d0
commit f6cb05e
Show file tree

Hide file tree

Showing 19 changed files with 66 additions and 23 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,11 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.0.1-RC11]
+
+- MessageProcessor
+  - Introduced MessageProcessor in LeiaMessageProduceClient instead of a lambda
+  - GetMessages from LeiaMessageProducerClient is now public
 
 ## [0.0.1-RC10]
 

diff --git a/leia-bom/pom.xml b/leia-bom/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
     </parent>
 
     <artifactId>leia-bom</artifactId>

diff --git a/leia-client-dropwizard/pom.xml b/leia-client-dropwizard/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-client-dropwizard/src/main/java/com/grookage/leia/dw/client/LeiaClientBundle.java b/leia-client-dropwizard/src/main/java/com/grookage/leia/dw/client/LeiaClientBundle.java
@@ -19,6 +19,7 @@
 import com.google.common.base.Preconditions;
 import com.grookage.leia.client.LeiaMessageProduceClient;
 import com.grookage.leia.client.datasource.NamespaceDataSource;
+import com.grookage.leia.client.processor.MessageProcessor;
 import com.grookage.leia.client.refresher.LeiaClientRefresher;
 import com.grookage.leia.client.refresher.LeiaClientSupplier;
 import com.grookage.leia.provider.config.LeiaHttpConfiguration;
@@ -65,6 +66,8 @@ protected LeiaSchemaValidator getSchemaValidator(T configuration,
                 .build();
     }
 
+    protected abstract Supplier<MessageProcessor> getMessageProcessor(T configuration);
+
     @Override
     public void run(T configuration, Environment environment) {
         final var namespaceDataSource = getNamespaceDataSource(configuration);
@@ -94,6 +97,7 @@ public void run(T configuration, Environment environment) {
                     .refresher(clientRefresher)
                     .schemaValidator(validator)
                     .mapper(environment.getObjectMapper())
+                    .messageProcessor(getMessageProcessor(configuration))
                     .build();
             producerClient.start();
         }

diff --git a/leia-client/pom.xml b/leia-client/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-client/src/main/java/com/grookage/leia/client/LeiaMessageProduceClient.java b/leia-client/src/main/java/com/grookage/leia/client/LeiaMessageProduceClient.java
@@ -16,6 +16,7 @@
 
 package com.grookage.leia.client;
 
+import com.grookage.leia.client.processor.MessageProcessor;
 import com.grookage.leia.models.mux.LeiaMessage;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.models.schema.transformer.TransformationTarget;
@@ -32,6 +33,7 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 
 
@@ -43,6 +45,8 @@ public class LeiaMessageProduceClient extends AbstractSchemaClient {
 
     private final Map<SchemaKey, Map<String, JsonPath>> compiledPaths = new HashMap<>();
 
+    private final Supplier<MessageProcessor> messageProcessor;
+
     /*
         Multiplexes from source and generates the list of messages as applicable
         a) Fetches the schemaDetails from schemaKey of source
@@ -80,7 +84,7 @@ private Optional<JsonPath> getJsonPath(SchemaKey schemaKey, String attributeName
                 Optional.ofNullable(compiledPaths.get(schemaKey).get(attributeName)) : Optional.empty();
     }
 
-    private Map<SchemaKey, LeiaMessage> getMessages(SchemaKey schemaKey, byte[] sourceMessage) {
+    public Map<SchemaKey, LeiaMessage> getMessages(SchemaKey schemaKey, byte[] sourceMessage) {
         final var messages = new HashMap<SchemaKey, LeiaMessage>();
         messages.put(schemaKey, LeiaMessage.builder()
                 .schemaKey(schemaKey)
@@ -104,9 +108,8 @@ private Map<SchemaKey, LeiaMessage> getMessages(SchemaKey schemaKey, byte[] sour
     }
 
     public void processMessages(final SchemaKey schemaKey,
-                                final byte[] sourceMessage,
-                                final UnaryOperator<Map<SchemaKey, LeiaMessage>> messageHandler) {
-        messageHandler.apply(getMessages(schemaKey, sourceMessage));
+                                final byte[] sourceMessage) {
+        messageProcessor.get().processMessages(getMessages(schemaKey, sourceMessage));
     }
 
     @Override

diff --git a/leia-client/src/main/java/com/grookage/leia/client/processor/MessageProcessor.java b/leia-client/src/main/java/com/grookage/leia/client/processor/MessageProcessor.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2024. Koushik R <[email protected]>.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.grookage.leia.client.processor;
+
+import com.grookage.leia.models.mux.LeiaMessage;
+import com.grookage.leia.models.schema.SchemaKey;
+
+import java.util.Map;
+
+public interface MessageProcessor {
+
+    void processMessages(Map<SchemaKey, LeiaMessage> messages);
+
+}
diff --git a/leia-client/src/test/java/com/grookage/leia/client/LeiaMessageProduceClientTest.java b/leia-client/src/test/java/com/grookage/leia/client/LeiaMessageProduceClientTest.java
@@ -17,11 +17,13 @@
 package com.grookage.leia.client;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.grookage.leia.client.processor.MessageProcessor;
 import com.grookage.leia.client.refresher.LeiaClientRefresher;
 import com.grookage.leia.client.stubs.TargetSchema;
 import com.grookage.leia.client.stubs.TestSchema;
 import com.grookage.leia.client.stubs.TestSchemaUnit;
 import com.grookage.leia.models.ResourceHelper;
+import com.grookage.leia.models.mux.LeiaMessage;
 import com.grookage.leia.models.schema.SchemaDetails;
 import com.grookage.leia.models.schema.SchemaKey;
 import com.grookage.leia.validator.LeiaSchemaValidator;
@@ -31,6 +33,7 @@
 import org.mockito.Mockito;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 class LeiaMessageProduceClientTest {
@@ -63,17 +66,17 @@ void testLeiaMessageProduceClient() {
                 .mapper(new ObjectMapper())
                 .refresher(clientRefresher)
                 .schemaValidator(schemaValidator)
+                .messageProcessor(() -> messages -> {
+                    Assertions.assertFalse(messages.isEmpty());
+                    Assertions.assertEquals(2, messages.size());
+                })
                 .build();
         schemaClient.start();
         final var testSchema = TestSchema.builder()
                 .userName("testUser")
                 .schemaUnits(List.of(TestSchemaUnit.builder()
                         .registeredName("testRegisteredName").build()))
                 .build();
-        schemaClient.processMessages(sourceSchema, mapper.writeValueAsBytes(testSchema), messages -> {
-            Assertions.assertFalse(messages.isEmpty());
-            Assertions.assertEquals(2, messages.size());
-            return messages;
-        });
+        schemaClient.processMessages(sourceSchema, mapper.writeValueAsBytes(testSchema));
     }
 }
diff --git a/leia-common/pom.xml b/leia-common/pom.xml
@@ -23,7 +23,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-core/pom.xml b/leia-core/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-dropwizard-es/pom.xml b/leia-dropwizard-es/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-dropwizard/pom.xml b/leia-dropwizard/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-elastic/pom.xml b/leia-elastic/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-models/pom.xml b/leia-models/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-parent/pom.xml b/leia-parent/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-bom</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-bom</relativePath>
     </parent>
 

diff --git a/leia-refresher/pom.xml b/leia-refresher/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-repository/pom.xml b/leia-repository/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/leia-schema-validator/pom.xml b/leia-schema-validator/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>com.grookage.leia</groupId>
         <artifactId>leia-parent</artifactId>
-        <version>0.0.1-RC10</version>
+        <version>0.0.1-RC11</version>
         <relativePath>../leia-parent</relativePath>
     </parent>
 

diff --git a/pom.xml b/pom.xml
@@ -18,7 +18,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.grookage.leia</groupId>
     <artifactId>leia</artifactId>
-    <version>0.0.1-RC10</version>
+    <version>0.0.1-RC11</version>
     <packaging>pom</packaging>
 
     <name>Leia</name>