TH2-4916

build.gradle

@@ -1,10 +1,13 @@
 plugins {
     id 'java-library'
+    id 'org.jetbrains.kotlin.jvm' version '1.6.21'
     id 'application'
     id 'com.palantir.docker' version '0.25.0'
     id "org.owasp.dependencycheck" version "8.1.2"
 }
 
+apply plugin: 'kotlin-kapt'
+
 dependencyCheck {
     format='HTML'
 }
@@ -38,7 +41,7 @@ repositories {
 
     mavenLocal()
 
-    configurations.all {
+    configurations.configureEach {
         resolutionStrategy.cacheChangingModulesFor 0, 'seconds'
         resolutionStrategy.cacheDynamicVersionsFor 0, 'seconds'
     }
@@ -60,10 +63,14 @@ jar {
 
 dependencies {
     api platform('com.exactpro.th2:bom:4.2.0')
-    implementation 'com.exactpro.th2:common:5.2.0-dev'
-    implementation 'com.exactpro.th2:codec:5.2.0-dev'
+    implementation 'com.exactpro.th2:common:5.3.0-json-raw-body-5144283961-5ae02cd-SNAPSHOT'
+    implementation 'com.exactpro.th2:codec:5.2.0-new-proto-5111859401-e24c7b8-SNAPSHOT'
+
+    implementation 'io.netty:netty-buffer:4.1.93.Final'
 
     implementation "org.slf4j:slf4j-api"
+    implementation 'org.apache.commons:commons-lang3:3.12.0'
+    implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'
 
     implementation 'net.sourceforge.javacsv:javacsv:2.0'
     implementation 'org.jetbrains:annotations:23.0.0'

src/main/java/com/exactpro/th2/codec/csv/CodecFactory.java

@@ -57,7 +57,7 @@ public Class<? extends IPipelineCodecSettings> getSettingsClass() {
     @Override
     public IPipelineCodec create(@Nullable IPipelineCodecSettings settings) {
         if (settings instanceof CsvCodecConfiguration) {
-            return new CsvCodec((CsvCodecConfiguration)settings);
+            return new TransportCsvCodec((CsvCodecConfiguration)settings);
         }
         throw new IllegalArgumentException("Unexpected setting type: " + (settings == null ? "null" : settings.getClass()));
     }
@@ -73,4 +73,4 @@ public void init(@NotNull InputStream inputStream) {
     @Override
     public void close() {
     }
-}
+}

src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java

@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package com.exactpro.th2.codec.csv;
 
 import static java.util.Objects.requireNonNull;
@@ -48,18 +49,18 @@
 import com.exactpro.th2.common.value.ValueUtils;
 import com.google.protobuf.ByteString;
 
-public class CsvCodec implements IPipelineCodec {
+public abstract class CsvCodec implements IPipelineCodec {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(CsvCodec.class);
-    private static final String HEADER_MSG_TYPE = "Csv_Header";
-    private static final String CSV_MESSAGE_TYPE = "Csv_Message";
-    private static final String HEADER_FIELD_NAME = "Header";
+    protected static final String HEADER_MSG_TYPE = "Csv_Header";
+    protected static final String CSV_MESSAGE_TYPE = "Csv_Message";
+    protected static final String HEADER_FIELD_NAME = "Header";
 
-    private static final String OVERRIDE_MESSAGE_TYPE_PROP_NAME_LOWERCASE = "th2.csv.override_message_type";
+    protected static final String OVERRIDE_MESSAGE_TYPE_PROP_NAME_LOWERCASE = "th2.csv.override_message_type";
 
-    private final CsvCodecConfiguration configuration;
-    private final String[] defaultHeader;
-    private final Charset charset;
+    protected final CsvCodecConfiguration configuration;
+    protected final String[] defaultHeader;
+    protected final Charset charset;
 
     public CsvCodec(CsvCodecConfiguration configuration) {
         this.configuration = requireNonNull(configuration, "'Configuration' parameter");
@@ -79,7 +80,7 @@ public CsvCodec(CsvCodecConfiguration configuration) {
     @Override
     public MessageGroup decode(@NotNull MessageGroup messageGroup) {
         MessageGroup.Builder groupBuilder = MessageGroup.newBuilder();
-        Collection<ErrorHolder> errors = new ArrayList<>();
+        Collection<ErrorHolder<RawMessage>> errors = new ArrayList<>();
         for (AnyMessage anyMessage : messageGroup.getMessagesList()) {
             if (anyMessage.hasMessage()) {
                 groupBuilder.addMessages(anyMessage);
@@ -97,12 +98,12 @@ public MessageGroup decode(@NotNull MessageGroup messageGroup) {
             }
 
             ByteString body = rawMessage.getBody();
-            List<String[]> data = decodeValues(body);
+            List<String[]> data = decodeValues(body.toByteArray());
             if (data.isEmpty()) {
                 if (LOGGER.isErrorEnabled()) {
                     LOGGER.error("The raw message does not contains any data: {}", MessageUtils.toJson(rawMessage));
                 }
-                errors.add(new ErrorHolder("The raw message does not contains any data", rawMessage));
+                errors.add(new ErrorHolder<>("The raw message does not contains any data", rawMessage));
                 continue;
             }
             decodeCsvData(errors, groupBuilder, rawMessage, data);
@@ -119,7 +120,7 @@ public MessageGroup decode(@NotNull MessageGroup messageGroup, @NotNull IReporti
         return decode(messageGroup);
     }
 
-    private RuntimeException createException(Collection<ErrorHolder> errors) {
+    private RuntimeException createException(Collection<ErrorHolder<RawMessage>> errors) {
         return new DecodeException(
                 "Cannot decode some messages: " + System.lineSeparator() + errors.stream()
                         .map(it -> "Message " + MessageUtils.toJson(it.originalMessage.getMetadata().getId()) + " cannot be decoded because " + it.text)
@@ -140,10 +141,9 @@ public MessageGroup encode(@NotNull MessageGroup messageGroup, @NotNull IReporti
     }
 
     @Override
-    public void close() {
-    }
+    public void close() {}
 
-    private void decodeCsvData(Collection<ErrorHolder> errors, MessageGroup.Builder groupBuilder, RawMessage rawMessage, Iterable<String[]> data) {
+    private void decodeCsvData(Collection<ErrorHolder<RawMessage>> errors, MessageGroup.Builder groupBuilder, RawMessage rawMessage, Iterable<String[]> data) {
         RawMessageMetadata originalMetadata = rawMessage.getMetadata();
 
         final String outputMessageType = originalMetadata.getPropertiesOrDefault(OVERRIDE_MESSAGE_TYPE_PROP_NAME_LOWERCASE, CSV_MESSAGE_TYPE);
@@ -157,7 +157,7 @@ private void decodeCsvData(Collection<ErrorHolder> errors, MessageGroup.Builder
                 if (LOGGER.isErrorEnabled()) {
                     LOGGER.error("Empty raw at {} index (starts with 1). Data: {}", currentIndex, data);
                 }
-                errors.add(new ErrorHolder("Empty raw at " + currentIndex + " index (starts with 1)", rawMessage));
+                errors.add(new ErrorHolder<>("Empty raw at " + currentIndex + " index (starts with 1)", rawMessage));
                 continue;
             }
 
@@ -188,7 +188,7 @@ private void decodeCsvData(Collection<ErrorHolder> errors, MessageGroup.Builder
                 if (LOGGER.isDebugEnabled()) {
                     LOGGER.debug(rawMessage.toString());
                 }
-                errors.add(new ErrorHolder(msg, strings, rawMessage));
+                errors.add(new ErrorHolder<>(msg, strings, rawMessage));
             }
 
             AnyMessage.Builder messageBuilder = groupBuilder.addMessagesBuilder();
@@ -223,7 +223,7 @@ private void decodeCsvData(Collection<ErrorHolder> errors, MessageGroup.Builder
         return copyArr;
     }
 
-    private int getHeaderArrayLength(String[] header, int index) {
+    protected int getHeaderArrayLength(String[] header, int index) {
         int length = 1;
         for (int i = index + 1; i < header.length && header[i].isEmpty(); i++) {
             length++;
@@ -244,8 +244,8 @@ private void setMetadata(RawMessageMetadata originalMetadata, Message.Builder me
         );
     }
 
-    private List<String[]> decodeValues(ByteString body) {
-        try (InputStream in = new ByteArrayInputStream(body.toByteArray())) {
+    protected List<String[]> decodeValues(byte[] body) {
+        try (InputStream in = new ByteArrayInputStream(body)) {
             CsvReader reader = new CsvReader(in, configuration.getDelimiter(), charset);
             try {
                 List<String[]> result = new ArrayList<>();
@@ -261,23 +261,23 @@ private List<String[]> decodeValues(ByteString body) {
         }
     }
 
-    private static class ErrorHolder {
+    protected static class ErrorHolder<T> {
         public final String text;
         public final String[] currentRow;
-        public final RawMessage originalMessage;
+        public final T originalMessage;
 
-        private ErrorHolder(String text, String[] currentRow, RawMessage originalMessage) {
+        protected ErrorHolder(String text, String[] currentRow, T originalMessage) {
             this.text = text;
             this.currentRow = currentRow;
             this.originalMessage = originalMessage;
         }
 
-        private ErrorHolder(String text, RawMessage originalMessage) {
+        protected ErrorHolder(String text, T originalMessage) {
             this(text, null, originalMessage);
         }
     }
 
-    private void trimEachElement(String[] elements) {
+    protected void trimEachElement(String[] elements) {
         for (int i = 0; i < elements.length; i++) {
             String element = elements[i];
             elements[i] = element.trim();

src/main/kotlin/com/exactpro/th2/codec/csv/TransportCsvCodec.kt

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2023 Exactpro (Exactpro Systems Limited)
+ *
+ * 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.exactpro.th2.codec.csv
+
+import com.exactpro.th2.codec.DecodeException
+import com.exactpro.th2.codec.csv.cfg.CsvCodecConfiguration
+import com.exactpro.th2.codec.util.toJson
+import com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.*
+import mu.KotlinLogging
+import java.util.*
+
+class TransportCsvCodec(
+    config: CsvCodecConfiguration
+) : CsvCodec(config) {
+    override fun decode(messageGroup: MessageGroup): MessageGroup {
+        val decodedMessages = mutableListOf<Message<*>>()
+        val errors: MutableCollection<ErrorHolder<RawMessage>> = ArrayList()
+
+        for (message in messageGroup.messages) {
+            if (message is ParsedMessage) {
+                decodedMessages.add(message)
+                continue
+            }
+
+            if (message !is RawMessage) {
+                LOGGER.error { "Message should either have a raw or parsed message but has nothing: $message" }
+                continue
+            }
+
+            val protocol = message.protocol
+            if ("" != protocol && !"csv".equals(protocol, ignoreCase = true)) {
+                LOGGER.error { "Wrong protocol: message should have empty or 'csv' protocol but has $protocol" }
+                continue
+            }
+
+            val data = decodeValues(message.body.array())
+
+            if (data.isEmpty()) {
+                LOGGER.error { "The raw message does not contains any data: ${message.toJson()}" }
+                errors.add(ErrorHolder<RawMessage>("The raw message does not contains any data", message))
+                continue
+            }
+
+            decodeCsvData(errors, decodedMessages, message, data)
+        }
+
+        if (!errors.isEmpty()) {
+            throw createException(errors)
+        }
+
+        return MessageGroup(decodedMessages)
+    }
+
+    private fun decodeCsvData(
+        errors: MutableCollection<ErrorHolder<RawMessage>>,
+        groupBuilder: MutableList<Message<*>>,
+        rawMessage: RawMessage,
+        data: Iterable<Array<String>>
+    ) {
+        val originalMetadata = rawMessage.metadata
+        val outputMessageType = originalMetadata[OVERRIDE_MESSAGE_TYPE_PROP_NAME_LOWERCASE] ?: CSV_MESSAGE_TYPE
+        var currentIndex = 0
+        var header = defaultHeader
+        for (strings in data) {
+
+            currentIndex++
+            if (strings.isEmpty()) {
+                LOGGER.error { "Empty raw at $currentIndex index (starts with 1). Data: $data" }
+                errors += ErrorHolder("Empty raw at $currentIndex index (starts with 1)", rawMessage)
+                continue
+            }
+
+            trimEachElement(strings)
+
+            if (header == null) {
+                LOGGER.debug { "Set header to: ${strings.contentToString()}" }
+                header = strings
+                if (configuration.isPublishHeader) {
+                    groupBuilder += ParsedMessage(
+                        id = rawMessage.id.toBuilder().addSubsequence(currentIndex).build(),
+                        eventId = rawMessage.eventId, // not set in proto version?
+                        type = HEADER_MSG_TYPE,
+                        metadata = rawMessage.metadata,
+                        body = mapOf(HEADER_FIELD_NAME to strings)
+                    )
+                }
+                continue
+            }
+
+            if (strings.size != header.size && configuration.validateLength) {
+                val msg = "Wrong fields count in message. Expected count: ${header.size}; actual: ${strings.size}; session alias: ${rawMessage.id.sessionAlias}"
+                LOGGER.error { msg }
+                LOGGER.debug { rawMessage.toString() }
+                errors.add(ErrorHolder(msg, strings, rawMessage))
+            }
+
+            var i = 0
+            val body = mutableMapOf<String, Any>()
+            while (i < header.size && i < strings.size) {
+                val extraLength = getHeaderArrayLength(header, i)
+                if (extraLength == 1) {
+                    body[header[i]] = strings[i]
+                    i++
+                } else {
+                    val values = copyArray(strings, i, i + extraLength)
+                    body[header[i]] = values
+                    i += extraLength
+                }
+            }
+
+            groupBuilder += ParsedMessage(
+                id = rawMessage.id.toBuilder().addSubsequence(currentIndex).build(),
+                eventId = rawMessage.eventId, // not set in proto version?
+                type = outputMessageType,
+                metadata = rawMessage.metadata,
+                body = body
+            )
+        }
+    }
+
+    private fun createException(errors: Collection<ErrorHolder<RawMessage>>): RuntimeException = DecodeException(
+        "Cannot decode some messages:\n$" + errors.joinToString(separator = "\n") {
+            "Message ${it.originalMessage.toJson()} cannot be decoded because ${it.text}"
+        }
+    )
+
+    companion object {
+        private val LOGGER = KotlinLogging.logger {}
+    }
+}