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
 prometheus.port, 1234, Port used by the server that provides DropWizard metrics
 |===
 

etc/config.properties

@@ -20,7 +20,7 @@ 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
 
 prometheus.port=1234

pom.xml

@@ -140,6 +140,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

@@ -27,32 +27,41 @@
 import com.teragrep.lsh_01.metrics.JmxReport;
 import com.teragrep.lsh_01.metrics.Report;
 import com.teragrep.lsh_01.metrics.Slf4jReport;
+import com.teragrep.lsh_01.conversion.*;
 import com.teragrep.lsh_01.pool.*;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
+import java.util.Map;
 
 public class Main {
 
     private final static Logger LOGGER = LogManager.getLogger(Main.class);
 
     public static void main(String[] args) {
+        Map<String, String> propsMap;
+        try {
+            propsMap = new PathProperties(System.getProperty("properties.file", "etc/config.properties"))
+                    .deepCopyAsUnmodifiableMap();
+        }
+        catch (IOException e) {
+            throw new IllegalArgumentException("Can't find properties file: ", e);
+        }
+
         NettyConfig nettyConfig = new NettyConfig();
         RelpConfig relpConfig = new RelpConfig();
         SecurityConfig securityConfig = new SecurityConfig();
         BasicAuthentication basicAuthentication = new BasicAuthenticationFactory().create();
         InternalEndpointUrlConfig internalEndpointUrlConfig = new InternalEndpointUrlConfig();
         LookupConfig lookupConfig = new LookupConfig();
-        PayloadConfig payloadConfig = new PayloadConfig();
         MetricsConfig metricsConfig = new MetricsConfig();
         try {
             nettyConfig.validate();
             relpConfig.validate();
             securityConfig.validate();
             internalEndpointUrlConfig.validate();
             lookupConfig.validate();
-            payloadConfig.validate();
             metricsConfig.validate();
         }
         catch (IllegalArgumentException e) {
@@ -63,7 +72,6 @@ public static void main(String[] args) {
         LOGGER.info("Got relp config: <[{}]>", relpConfig);
         LOGGER.info("Got internal endpoint config: <[{}]>", internalEndpointUrlConfig);
         LOGGER.info("Got lookup table config: <[{}]>", lookupConfig);
-        LOGGER.info("Got payload config: <[{}]>", payloadConfig);
         LOGGER.info("Authentication required: <[{}]>", securityConfig.authRequired);
 
         // metrics
@@ -76,14 +84,14 @@ public static void main(String[] args) {
         RelpConnectionFactory relpConnectionFactory = new RelpConnectionFactory(relpConfig, metricRegistry);
         Pool<IManagedRelpConnection> pool = new Pool<>(relpConnectionFactory, new ManagedRelpConnectionStub());
 
-        IMessageHandler relpConversion = new MetricRelpConversion(
-                new RelpConversion(pool, securityConfig, basicAuthentication, lookupConfig, payloadConfig),
+        IMessageHandler conversion = new MetricRelpConversion(
+                new ConversionFactory(propsMap, pool, securityConfig, basicAuthentication, lookupConfig).conversion(),
                 metricRegistry
         );
 
         try (
                 HttpServer server = new MetricHttpServer(
-                        new NettyHttpServer(nettyConfig, relpConversion, null, 200, internalEndpointUrlConfig),
+                        new NettyHttpServer(nettyConfig, conversion, null, 200, internalEndpointUrlConfig),
                         report
                 )
         ) {

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/authentication/BasicAuthentication.java

@@ -67,4 +67,15 @@ public Subject asSubject(String token) {
             return subjectStub;
         }
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
+        final BasicAuthentication cast = (BasicAuthentication) o;
+        // equality of CredentialLookup can't be checked. It has no equals() function and the Map is only traversable by key.
+        return decoder.equals(cast.decoder);
+    }
 }

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

@@ -0,0 +1,34 @@
+/*
+  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.config;
+
+import java.io.IOException;
+import java.util.Map;
+
+public interface Configuration {
+
+    /**
+     * Get configuration as an unmodifiable map so that it can't be altered anymore.
+     * 
+     * @return configuration as an unmodifiable map
+     * @throws IOException if configuration file is not found
+     */
+    Map<String, String> deepCopyAsUnmodifiableMap() throws IOException;
+}

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

@@ -0,0 +1,142 @@
+/*
+  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.config;
+
+import com.teragrep.lsh_01.authentication.BasicAuthentication;
+import com.teragrep.lsh_01.conversion.IMessageHandler;
+import com.teragrep.lsh_01.conversion.JsonConversion;
+import com.teragrep.lsh_01.conversion.RegexConversion;
+import com.teragrep.lsh_01.conversion.RelpConversion;
+import com.teragrep.lsh_01.pool.IManagedRelpConnection;
+import com.teragrep.lsh_01.pool.Pool;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+public final class ConversionFactory {
+
+    private final static Logger LOGGER = LogManager.getLogger(ConversionFactory.class);
+
+    private final String splitType;
+    private final String regexPattern;
+    private final Pool<IManagedRelpConnection> pool;
+    private final SecurityConfig securityConfig;
+    private final BasicAuthentication basicAuthentication;
+    private final LookupConfig lookupConfig;
+
+    public ConversionFactory(
+            Map<String, String> configuration,
+            Pool<IManagedRelpConnection> pool,
+            SecurityConfig securityConfig,
+            BasicAuthentication basicAuthentication,
+            LookupConfig lookupConfig
+    ) {
+        // if system property is not specified, defaults to config file (the Map)
+        this(
+                System.getProperty("payload.splitType", configuration.get("payload.splitType")),
+                System
+                        .getProperty(
+                                "payload.splitType.regex.pattern", configuration.get("payload.splitType.regex.pattern")
+                        ),
+                pool,
+                securityConfig,
+                basicAuthentication,
+                lookupConfig
+        );
+    }
+
+    public ConversionFactory(
+            String splitType,
+            String regexPattern,
+            Pool<IManagedRelpConnection> pool,
+            SecurityConfig securityConfig,
+            BasicAuthentication basicAuthentication,
+            LookupConfig lookupConfig
+    ) {
+        this.splitType = splitType;
+        this.regexPattern = regexPattern;
+        this.pool = pool;
+        this.securityConfig = securityConfig;
+        this.basicAuthentication = basicAuthentication;
+        this.lookupConfig = lookupConfig;
+    }
+
+    public IMessageHandler conversion() {
+        LOGGER
+                .info(
+                        "Creating IMessageHandler with given configuration: payload.splitType=<[{}]>, payload.splitType.regex.pattern=<[{}]>",
+                        splitType, regexPattern
+                );
+
+        validateConfiguration();
+
+        IMessageHandler conversion = new RelpConversion(pool, securityConfig, basicAuthentication, lookupConfig);
+
+        // apply splitting if configured. "none" value is skipped
+        switch (splitType) {
+            case "regex":
+                conversion = new RegexConversion(conversion, regexPattern);
+                break;
+            case "json_array":
+                conversion = new JsonConversion(conversion);
+                break;
+        }
+
+        return conversion;
+    }
+
+    private void validateConfiguration() {
+        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 splitType: " + splitType
+                                + ". Has to be 'regex', 'json_array' or 'none'."
+                );
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
+        final ConversionFactory cast = (ConversionFactory) o;
+        return splitType.equals(cast.splitType) && regexPattern.equals(cast.regexPattern) && pool
+                .equals(cast.pool) && securityConfig.equals(cast.securityConfig)
+                && basicAuthentication.equals(cast.basicAuthentication) && lookupConfig.equals(cast.lookupConfig);
+    }
+}

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

@@ -40,4 +40,14 @@ public void validate() {
     public String toString() {
         return "LookupConfig{" + "hostnamePath='" + hostnamePath + '\'' + ", appNamePath='" + appNamePath + '\'' + '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (o == null || getClass() != o.getClass())
+            return false;
+        final LookupConfig cast = (LookupConfig) o;
+        return hostnamePath.equals(cast.hostnamePath) && appNamePath.equals(cast.appNamePath);
+    }
 }