JSON Array splitting (#77)

* Allow splitting incoming messages with JSON array format

Delete PayloadConfig and include configuration in ConversionFactory that provides the correct object instead
Add secondary ctor to RegexConversion
Use Collections.singletonList in DefaultPayload, RegexConversion as final class
Add object equality test for RegexPayload
Move tests to the correct package
Move config tests from RegexPayloadTest to PayloadConfigTest and add tests for splitType
Refactor split() function in Payload
Refactor JsonPayload and RegexPayload to encapsulate Payload instead of String
Rename Payload's take() function to message()
Implement json_array splitting
Refactor Payload objects and regex splitting
Remove duplicate tests from RegexSplittingTest
Apply spotless
Move regex splitting tests to their own file, refactor the tests
Change config to allow splitting with json_array
Refactor regex splitting to a decorator

* Fix rebase

* Use try-with-resources when loading configuration file

* Add object equality tests for ConversionFactory

* Split object equality test into two tests in RegexPayloadTest

* Added missing equals and hashCode functions, added equality tests

* Add missing hashCodes to ConversionFactory and LookupConfig

* Remove checking System Properties in ConversionFactory, make end-to-end tests use properties from a resource file

* Use rlo_06 for asserting payload correctness, add multithreading test

* Use expected message amounts instead of result list size as an expected value in assertions

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>
@@ -159,6 +170,13 @@
       <artifactId>rlp_03</artifactId>
       <version>${rlp_03.version}</version>
     </dependency>
+    <!-- Syslog parsing, testing -->
+    <dependency>
+      <groupId>com.teragrep</groupId>
+      <artifactId>rlo_06</artifactId>
+      <version>9.0.1</version>
+      <scope>test</scope>
+    </dependency>
     <!-- securemock, testing -->
     <dependency>
       <groupId>org.elasticsearch</groupId>
@@ -231,6 +249,7 @@
             <exclude>src/main/assembly/jar-with-dependencies.xml</exclude>
             <exclude>src/main/resources/*</exclude>
             <exclude>src/test/resources/certificates/*</exclude>
+            <exclude>src/test/resources/properties/*</exclude>
             <!-- readme -->
             <exclude>README.adoc</exclude>
           </excludes>

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

@@ -24,6 +24,7 @@
 import org.apache.logging.log4j.Logger;
 
 import java.util.Base64;
+import java.util.Objects;
 
 public class BasicAuthentication {
 
@@ -67,4 +68,19 @@ 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;
+        return decoder.equals(cast.decoder) && credentialLookup.equals(cast.credentialLookup);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(decoder, credentialLookup);
+    }
 }

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;
+}