JSON Array splitting

README.adoc

@@ -35,8 +35,8 @@ security.authRequired,true,Sets whether Basic HTTP Authorization headers are req
 credentials.file,etc/credentials.json,A json file with array of identity:credential mappings
 lookups.hostname.file,etc/hostname.json,Path to username-to-hostname lookup table
 lookups.appname.file,etc/appname.json,Path to username-to-appname lookup table
-payload.splitRegex, \n (newline), A regex based on which incoming requests will be split into multiple outgoing messages
-payload.splitEnabled, false, Sets whether splitting incoming messages by splitRegex is enabled
+payload.splitType, none, Sets how to split incoming messages. Supported split types are 'regex' and 'json_array'. Use 'none' for no splitting.
+payload.splitType.regex.pattern, \n (newline), A regex based on which incoming requests will be split into multiple outgoing messages
 |===
 
 === Lookup tables

etc/config.properties

@@ -20,5 +20,5 @@ credentials.file=etc/credentials.json
 lookups.hostname.file=etc/hostname.json
 lookups.appname.file=etc/appname.json
 
-payload.splitRegex=\n
-payload.splitEnabled=false
+payload.splitType=none
+payload.splitType.regex.pattern=\n

pom.xml

@@ -101,6 +101,17 @@
       <artifactId>log4j-core</artifactId>
       <version>${log4j.version}</version>
     </dependency>
+    <!-- JSON processing -->
+    <dependency>
+      <groupId>jakarta.json</groupId>
+      <artifactId>jakarta.json-api</artifactId>
+      <version>2.1.3</version>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.parsson</groupId>
+      <artifactId>parsson</artifactId>
+      <version>1.1.7</version>
+    </dependency>
     <!-- junit, testing -->
     <dependency>
       <groupId>org.junit.jupiter</groupId>

src/main/java/com/teragrep/lsh_01/HttpInitializer.java

@@ -20,6 +20,7 @@
 package com.teragrep.lsh_01;
 
 import com.teragrep.lsh_01.config.InternalEndpointUrlConfig;
+import com.teragrep.lsh_01.conversion.IMessageHandler;
 import com.teragrep.lsh_01.util.LoggingHttpObjectAggregator;
 import com.teragrep.lsh_01.util.SslHandlerProvider;
 import io.netty.channel.ChannelInitializer;

src/main/java/com/teragrep/lsh_01/HttpServerHandler.java

@@ -20,6 +20,7 @@
 package com.teragrep.lsh_01;
 
 import com.teragrep.lsh_01.config.InternalEndpointUrlConfig;
+import com.teragrep.lsh_01.conversion.IMessageHandler;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;

src/main/java/com/teragrep/lsh_01/Main.java

@@ -22,6 +22,7 @@
 import com.teragrep.lsh_01.authentication.BasicAuthentication;
 import com.teragrep.lsh_01.authentication.BasicAuthenticationFactory;
 import com.teragrep.lsh_01.config.*;
+import com.teragrep.lsh_01.conversion.*;
 import com.teragrep.lsh_01.pool.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -60,18 +61,22 @@ public static void main(String[] args) {
         RelpConnectionFactory relpConnectionFactory = new RelpConnectionFactory(relpConfig);
         Pool<IManagedRelpConnection> pool = new Pool<>(relpConnectionFactory, new ManagedRelpConnectionStub());
 
-        RelpConversion relpConversion = new RelpConversion(
-                pool,
-                securityConfig,
-                basicAuthentication,
-                lookupConfig,
-                payloadConfig
-        );
+        IMessageHandler conversion = new RelpConversion(pool, securityConfig, basicAuthentication, lookupConfig);
+
+        // apply splitting if configured
+        switch (payloadConfig.splitType) { // payloadConfig is Validateable, no need to check wrong config here
+            case "regex":
+                conversion = new RegexConversion(conversion, payloadConfig.regexPattern);
+                break;
+            case "json_array":
+                conversion = new JsonConversion(conversion);
+                break;
+        }
 
         try (
                 NettyHttpServer server = new NettyHttpServer(
                         nettyConfig,
-                        relpConversion,
+                        conversion,
                         null,
                         200,
                         internalEndpointUrlConfig

src/main/java/com/teragrep/lsh_01/MessageProcessor.java

@@ -21,6 +21,7 @@
 
 import com.teragrep.lsh_01.authentication.*;
 import com.teragrep.lsh_01.config.InternalEndpointUrlConfig;
+import com.teragrep.lsh_01.conversion.IMessageHandler;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;

src/main/java/com/teragrep/lsh_01/NettyHttpServer.java

@@ -21,6 +21,7 @@
 
 import com.teragrep.lsh_01.config.InternalEndpointUrlConfig;
 import com.teragrep.lsh_01.config.NettyConfig;
+import com.teragrep.lsh_01.conversion.IMessageHandler;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelOption;

src/main/java/com/teragrep/lsh_01/config/PayloadConfig.java

@@ -24,33 +24,43 @@
 
 public class PayloadConfig implements Validateable {
 
-    public final String splitRegex;
-    public final boolean splitEnabled;
+    public final String regexPattern;
+    public final String splitType;
 
     public PayloadConfig() {
         PropertiesReaderUtilityClass propertiesReader = new PropertiesReaderUtilityClass(
                 System.getProperty("properties.file", "etc/config.properties")
         );
-        splitRegex = propertiesReader.getStringProperty("payload.splitRegex");
-        splitEnabled = propertiesReader.getBooleanProperty("payload.splitEnabled");
+        this.regexPattern = propertiesReader.getStringProperty("payload.splitType.regex.pattern");
+        this.splitType = propertiesReader.getStringProperty("payload.splitType");
     }
 
     @Override
     public void validate() {
-        if (splitEnabled) {
-            try {
-                Pattern.compile(splitRegex);
-            }
-            catch (PatternSyntaxException e) {
+        switch (splitType) {
+            case "regex":
+                try {
+                    Pattern.compile(regexPattern);
+                }
+                catch (PatternSyntaxException e) {
+                    throw new IllegalArgumentException(
+                            "Configuration has an invalid regex (payload.splitType.regex.pattern): " + regexPattern
+                    );
+                }
+                break;
+            case "json_array":
+            case "none":
+                break;
+            default:
                 throw new IllegalArgumentException(
-                        "Configuration has an invalid regex (payload.splitRegex): " + splitRegex
+                        "Configuration has an invalid splitType: " + splitType
+                                + ". Has to be 'regex', 'json_array' or 'none'."
                 );
-            }
         }
     }
 
     @Override
     public String toString() {
-        return "PayloadConfig{" + "splitRegex=" + splitRegex + ", splitEnabled=" + splitEnabled + '}';
+        return "PayloadConfig{" + "splitType=" + splitType + ", splitType.regex.pattern=" + regexPattern + '}';
     }
 }

src/main/java/com/teragrep/lsh_01/conversion/DefaultPayload.java

@@ -0,0 +1,49 @@
+/*
+  logstash-http-input to syslog bridge
+  Copyright 2024 Suomen Kanuuna Oy
+
+  Derivative Work of Elasticsearch
+  Copyright 2012-2015 Elasticsearch
+
+  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.teragrep.lsh_01.conversion;
+
+import java.util.Collections;
+import java.util.List;
+
+public final class DefaultPayload implements Payload {
+
+    private final String message;
+
+    public DefaultPayload(String message) {
+        this.message = message;
+    }
+
+    @Override
+    public List<String> messages() {
+        return Collections.singletonList(message);
+    }
+
+    @Override
+    public boolean equals(final Object object) {
+        if (this == object)
+            return true;
+        if (object == null)
+            return false;
+        if (object.getClass() != this.getClass())
+            return false;
+        final DefaultPayload cast = (DefaultPayload) object;
+        return message.equals(cast.message);
+    }
+}

src/main/java/com/teragrep/lsh_01/IMessageHandler.java → src/main/java/com/teragrep/lsh_01/conversion/IMessageHandler.java

@@ -17,7 +17,7 @@
   See the License for the specific language governing permissions and
   limitations under the License.
 */
-package com.teragrep.lsh_01;
+package com.teragrep.lsh_01.conversion;
 
 import com.teragrep.lsh_01.authentication.Subject;
 

src/main/java/com/teragrep/lsh_01/conversion/JsonConversion.java

@@ -0,0 +1,70 @@
+/*
+  logstash-http-input to syslog bridge
+  Copyright 2024 Suomen Kanuuna Oy
+
+  Derivative Work of Elasticsearch
+  Copyright 2012-2015 Elasticsearch
+
+  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.teragrep.lsh_01.conversion;
+
+import com.teragrep.lsh_01.authentication.Subject;
+
+import java.util.Map;
+
+/**
+ * Decorator for IMessageHandler that splits messages arriving as an array of JSON objects.
+ */
+public final class JsonConversion implements IMessageHandler {
+
+    private final IMessageHandler conversion;
+
+    public JsonConversion(IMessageHandler conversion) {
+        this.conversion = conversion;
+    }
+
+    @Override
+    public boolean onNewMessage(Subject subject, Map<String, String> headers, String body) {
+        JsonPayload originalPayload = new JsonPayload(new DefaultPayload(body));
+
+        boolean msgSent = true;
+        for (String message : originalPayload.messages()) { // each object individually as a String
+            if (!conversion.onNewMessage(subject, headers, message)) {
+                msgSent = false;
+            }
+        }
+
+        return msgSent;
+    }
+
+    @Override
+    public Subject asSubject(String token) {
+        return conversion.asSubject(token);
+    }
+
+    @Override
+    public boolean requiresToken() {
+        return conversion.requiresToken();
+    }
+
+    @Override
+    public IMessageHandler copy() {
+        return new JsonConversion(conversion.copy());
+    }
+
+    @Override
+    public Map<String, String> responseHeaders() {
+        return conversion.responseHeaders();
+    }
+}

src/main/java/com/teragrep/lsh_01/conversion/JsonPayload.java

@@ -0,0 +1,70 @@
+/*
+  logstash-http-input to syslog bridge
+  Copyright 2024 Suomen Kanuuna Oy
+
+  Derivative Work of Elasticsearch
+  Copyright 2012-2015 Elasticsearch
+
+  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.teragrep.lsh_01.conversion;
+
+import jakarta.json.*;
+
+import java.io.StringReader;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A Json array payload splittable into individual json objects.
+ */
+public final class JsonPayload implements Payload {
+
+    private final Payload payload;
+
+    public JsonPayload(Payload payload) {
+        this.payload = payload;
+    }
+
+    /**
+     * Splits the array of JSON objects into payloads with one object each. Has a side effect of removing whitespace
+     * from the payloads because of jsonObject.toString().
+     *
+     * @return list of messages
+     */
+    @Override
+    public List<String> messages() {
+        List<String> allMessages = new ArrayList<>();
+        for (String message : payload.messages()) {
+            JsonReader reader = Json.createReader(new StringReader(message));
+            JsonArray payloadMessages = reader.readArray();
+
+            // transform all json objects into DefaultPayloads and return the list
+            allMessages.addAll(payloadMessages.getValuesAs(JsonObject::toString));
+        }
+
+        return allMessages;
+    }
+
+    @Override
+    public boolean equals(final Object object) {
+        if (this == object)
+            return true;
+        if (object == null)
+            return false;
+        if (object.getClass() != this.getClass())
+            return false;
+        final JsonPayload cast = (JsonPayload) object;
+        return payload.equals(cast.payload);
+    }
+}