diff --git a/README.md b/README.md index 5c83aa9..2a85fc6 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,20 @@ -# Csv codec (3.2.1) +# Csv codec (4.0.0) ## Description Designed for decode csv raw messages from csv reader to the parsed messages. +It is based on [th2-codec](https://github.com/th2-net/th2-codec). +You can find additional information [here](https://github.com/th2-net/th2-codec/blob/master/README.md) ## Decoding The codec decodes each raw message in the received batch. -Each raw message might contains several line in CSV format. +Each raw message might contain several line in CSV format. If the default header parameter is not set the codec trites the first line from the raw message as a header. Otherwise, the default header will be used for decoding the rest of data. If no data was decoded from raw message, the message will be skipped, and an error event will be reported. +**NOTE: the encoding is not supported**. + ## Decode Example Simple example: @@ -53,7 +57,7 @@ into ``` ## Settings -Csv codec has following parameters: +Csv codec has the following parameters: ```yaml default-header: [A, B, C] @@ -73,38 +77,58 @@ The default value for the name is `CodecCsv`. **validate-length** - check if csv have different count of values against header's count -## Pins - -The CSV codec requires at leas one pin with the following attributes: -1. `decode_in` and `subscribe` - to receive raw data. -2. `decode_out` and `publish` - to send decoded data. - -The number of pins with each set of attributes is not limited. **Pins can use filters**. - ## Full configuration example ```yaml apiVersion: th2.exactpro.com/v1 -kind: Th2GenericBox +kind: Th2Box metadata: name: codec-csv spec: + image-name: ghcr.io/th2-net/th2-codec-csv + image-version: 4.0.0 custom-config: - default-header: [A, B, C] - delimiter: ',' - encoding: UTF-8 + codecSettings: + default-header: [A, B, C] + delimiter: ',' + encoding: UTF-8 pins: + # encoder + - name: in_codec_encode + connection-type: mq + attributes: [ 'encoder_in', 'parsed', 'subscribe' ] + - name: out_codec_encode + connection-type: mq + attributes: [ 'encoder_out', 'raw', 'publish' ] # decoder - name: in_codec_decode connection-type: mq - attributes: ['decode_in', 'subscribe', 'group'] + attributes: ['decoder_in', 'raw', 'subscribe'] - name: out_codec_decode connection-type: mq - attributes: ['decode_out', 'publish', 'parsed'] + attributes: ['decoder_out', 'parsed', 'publish'] + # encoder general (technical) + - name: in_codec_general_encode + connection-type: mq + attributes: ['general_encoder_in', 'parsed', 'subscribe'] + - name: out_codec_general_encode + connection-type: mq + attributes: ['general_encoder_out', 'raw', 'publish'] + # decoder general (technical) + - name: in_codec_general_decode + connection-type: mq + attributes: ['general_decoder_in', 'raw', 'subscribe'] + - name: out_codec_general_decode + connection-type: mq + attributes: ['general_decoder_out', 'parsed', 'publish'] ``` ## Release notes +### 4.0.0 + ++ Migrated to `th2-codec` core part. Uses the standard configuration format and pins for th2-codec + ### 3.2.1 + fixed: last array-field as simple value diff --git a/build.gradle b/build.gradle index 32a9903..dca4f2f 100644 --- a/build.gradle +++ b/build.gradle @@ -53,11 +53,16 @@ dependencies { api platform('com.exactpro.th2:bom:3.1.0') implementation 'com.exactpro.th2:common:3.37.1' + implementation 'com.exactpro.th2:codec:4.7.0' implementation "org.slf4j:slf4j-log4j12" implementation "org.slf4j:slf4j-api" implementation 'net.sourceforge.javacsv:javacsv:2.0' + implementation 'org.jetbrains:annotations:23.0.0' + + compileOnly 'com.google.auto.service:auto-service-annotations:1.0.1' + annotationProcessor 'com.google.auto.service:auto-service:1.0.1' testImplementation 'org.mockito:mockito-core:3.5.15' testImplementation 'org.junit.jupiter:junit-jupiter:5.6.2' @@ -68,13 +73,13 @@ test { } application { - mainClassName 'com.exactpro.th2.codec.csv.Main' + mainClassName 'com.exactpro.th2.codec.MainKt' } applicationName = 'service' distTar { - archiveName "${applicationName}.tar" + archiveFileName.set("${applicationName}.tar") } dockerPrepare { diff --git a/gradle.properties b/gradle.properties index f005651..ae63539 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1,3 +1,3 @@ -release_version = 3.2.1 +release_version = 4.0.0 docker_image_name= diff --git a/src/main/java/com/exactpro/th2/codec/csv/CodecFactory.java b/src/main/java/com/exactpro/th2/codec/csv/CodecFactory.java new file mode 100644 index 0000000..7a9f27f --- /dev/null +++ b/src/main/java/com/exactpro/th2/codec/csv/CodecFactory.java @@ -0,0 +1,76 @@ +/* + * Copyright 2022 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 java.io.InputStream; +import java.util.Collections; +import java.util.Set; + +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import com.exactpro.th2.codec.api.IPipelineCodec; +import com.exactpro.th2.codec.api.IPipelineCodecContext; +import com.exactpro.th2.codec.api.IPipelineCodecFactory; +import com.exactpro.th2.codec.api.IPipelineCodecSettings; +import com.exactpro.th2.codec.csv.cfg.CsvCodecConfiguration; +import com.google.auto.service.AutoService; + +@AutoService(IPipelineCodecFactory.class) +public class CodecFactory implements IPipelineCodecFactory { + public static final String PROTOCOL = "csv"; + private static final Set PROTOCOLS = Collections.singleton(PROTOCOL); + + @NotNull + @Override + public String getProtocol() { + return PROTOCOL; + } + + @NotNull + @Override + public Set getProtocols() { + return PROTOCOLS; + } + + @NotNull + @Override + public Class getSettingsClass() { + return CsvCodecConfiguration.class; + } + + @NotNull + @Override + public IPipelineCodec create(@Nullable IPipelineCodecSettings settings) { + if (settings instanceof CsvCodecConfiguration) { + return new CsvCodec((CsvCodecConfiguration)settings); + } + throw new IllegalArgumentException("Unexpected setting type: " + (settings == null ? "null" : settings.getClass())); + } + + @Override + public void init(@NotNull IPipelineCodecContext iPipelineCodecContext) { + } + + @Override + public void init(@NotNull InputStream inputStream) { + } + + @Override + public void close() { + } +} diff --git a/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java b/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java index 370e893..c5b699e 100644 --- a/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java +++ b/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java @@ -25,12 +25,19 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.stream.Collectors; +import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.csvreader.CsvReader; +import com.exactpro.th2.codec.CodecException; +import com.exactpro.th2.codec.DecodeException; +import com.exactpro.th2.codec.api.IPipelineCodec; +import com.exactpro.th2.codec.api.IReportingContext; import com.exactpro.th2.codec.csv.cfg.CsvCodecConfiguration; +import com.exactpro.th2.codec.util.MessageUtilKt; import com.exactpro.th2.common.event.Event; import com.exactpro.th2.common.event.Event.Status; import com.exactpro.th2.common.event.EventUtils; @@ -45,33 +52,26 @@ import com.exactpro.th2.common.grpc.MessageMetadata; import com.exactpro.th2.common.grpc.RawMessage; import com.exactpro.th2.common.grpc.RawMessageMetadata; +import com.exactpro.th2.common.message.MessageUtils; import com.exactpro.th2.common.schema.message.MessageListener; import com.exactpro.th2.common.schema.message.MessageRouter; import com.exactpro.th2.common.value.ValueUtils; import com.google.protobuf.ByteString; import com.google.protobuf.TextFormat; -public class CsvCodec implements MessageListener { +public class CsvCodec implements IPipelineCodec { private static final Logger LOGGER = LoggerFactory.getLogger(CsvCodec.class); - static final String DECODE_IN_ATTRIBUTE = "decode_in"; - static final String DECODE_OUT_ATTRIBUTE = "decode_out"; 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"; - private final MessageRouter router; private final CsvCodecConfiguration configuration; private final String[] defaultHeader; private final Charset charset; - private final MessageRouter eventRouter; - private final EventID rootId; - public CsvCodec(MessageRouter router, MessageRouter eventRouter, EventID rootId, CsvCodecConfiguration configuration) { - this.router = requireNonNull(router, "'Router' parameter"); - this.eventRouter = requireNonNull(eventRouter, "'Event router' parameter"); + public CsvCodec(CsvCodecConfiguration configuration) { this.configuration = requireNonNull(configuration, "'Configuration' parameter"); - this.rootId = requireNonNull(rootId, "'Root id' parameter"); List defaultHeader = configuration.getDefaultHeader(); this.defaultHeader = defaultHeader == null ? null @@ -83,45 +83,66 @@ public CsvCodec(MessageRouter router, MessageRouter errors = new ArrayList<>(); - MessageGroupBatch.Builder batchBuilder = MessageGroupBatch.newBuilder(); - - for (MessageGroup originalGroup : message.getGroupsList()) { - MessageGroup.Builder groupBuilder = batchBuilder.addGroupsBuilder(); - for (AnyMessage anyMessage : originalGroup.getMessagesList()) { - if (anyMessage.hasMessage()) { - groupBuilder.addMessages(anyMessage); - continue; - } - if (!anyMessage.hasRawMessage()) { - LOGGER.error("Message should either have a raw or parsed message but has nothing: {}", anyMessage); - continue; - } - RawMessage rawMessage = anyMessage.getRawMessage(); - ByteString body = rawMessage.getBody(); - List data = decodeValues(body); - if (data.isEmpty()) { - if (LOGGER.isErrorEnabled()) { - LOGGER.error("The raw message does not contains any data: {}", TextFormat.shortDebugString(rawMessage)); - } - errors.add(new ErrorHolder("The raw message does not contains any data", rawMessage)); - continue; - } - decodeCsvData(errors, groupBuilder, rawMessage, data); + public MessageGroup decode(@NotNull MessageGroup messageGroup) { + MessageGroup.Builder groupBuilder = MessageGroup.newBuilder(); + Collection errors = new ArrayList<>(); + for (AnyMessage anyMessage : messageGroup.getMessagesList()) { + if (anyMessage.hasMessage()) { + groupBuilder.addMessages(anyMessage); + continue; + } + if (!anyMessage.hasRawMessage()) { + LOGGER.error("Message should either have a raw or parsed message but has nothing: {}", anyMessage); + continue; + } + RawMessage rawMessage = anyMessage.getRawMessage(); + ByteString body = rawMessage.getBody(); + List data = decodeValues(body); + 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)); + continue; } + decodeCsvData(errors, groupBuilder, rawMessage, data); + } + if (!errors.isEmpty()) { + throw createException(errors); + } + return groupBuilder.build(); + } - send(batchBuilder.build()); - if (!errors.isEmpty()) { - reportErrors(errors); - } + @NotNull + @Override + public MessageGroup decode(@NotNull MessageGroup messageGroup, @NotNull IReportingContext iReportingContext) { + return decode(messageGroup); + } - } catch (Exception ex) { - LOGGER.error("Cannot process batch for {}: {}", consumerTag, message, ex); - } + private RuntimeException createException(Collection 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) + .collect(Collectors.joining(System.lineSeparator())) + ); + } + + @NotNull + @Override + public MessageGroup encode(@NotNull MessageGroup messageGroup) { + throw new UnsupportedOperationException("encode method is not implemented"); + } + + @NotNull + @Override + public MessageGroup encode(@NotNull MessageGroup messageGroup, @NotNull IReportingContext iReportingContext) { + throw new UnsupportedOperationException("encode method is not implemented"); + } + + @Override + public void close() { } private void decodeCsvData(Collection errors, MessageGroup.Builder groupBuilder, RawMessage rawMessage, Iterable data) { @@ -219,7 +240,7 @@ private void setMetadata(RawMessageMetadata originalMetadata, Message.Builder me ); } - private List decodeValues(ByteString body) throws IOException { + private List decodeValues(ByteString body) { try (InputStream in = new ByteArrayInputStream(body.toByteArray())) { CsvReader reader = new CsvReader(in, configuration.getDelimiter(), charset); try { @@ -231,6 +252,8 @@ private List decodeValues(ByteString body) throws IOException { } finally { reader.close(); } + } catch (IOException e) { + throw new RuntimeException("cannot read data from raw bytes", e); } } @@ -250,37 +273,6 @@ private ErrorHolder(String text, RawMessage originalMessage) { } } - private void reportErrors(List errors) { - try { - EventBatch.Builder errorBatch = EventBatch.newBuilder() - .setParentEventId(rootId); - for (ErrorHolder error : errors) { - Event event = Event.start() - .endTimestamp() - .type("DecodingError") - .name("Error during decoding data") - .status(Status.FAILED) - .messageID(error.originalMessage.getMetadata().getId()) - .bodyData(EventUtils.createMessageBean(error.text)); - if (error.currentRow != null) { - event.bodyData(EventUtils.createMessageBean("Current data: " + Arrays.toString(error.currentRow))); - } - errorBatch.addEvents(event.toProtoEvent(rootId.getId())); - } - - eventRouter.send(errorBatch.build()); - } catch (Exception ex) { - LOGGER.error("Cannot send error to the event store", ex); - } - } - - private void send(MessageGroupBatch batch) throws IOException { - if (batch.getGroupsList().isEmpty()) { - return; - } - router.sendAll(batch, DECODE_OUT_ATTRIBUTE); - } - private void trimEachElement(String[] elements) { for (int i = 0; i < elements.length; i++) { String element = elements[i]; diff --git a/src/main/java/com/exactpro/th2/codec/csv/Main.java b/src/main/java/com/exactpro/th2/codec/csv/Main.java deleted file mode 100644 index 38fc69d..0000000 --- a/src/main/java/com/exactpro/th2/codec/csv/Main.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright 2020-2020 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 static com.exactpro.th2.common.metrics.CommonMetrics.setLiveness; -import static com.exactpro.th2.common.metrics.CommonMetrics.setReadiness; - -import java.util.Deque; -import java.util.Objects; -import java.util.concurrent.ConcurrentLinkedDeque; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.exactpro.th2.codec.csv.cfg.CsvCodecConfiguration; -import com.exactpro.th2.common.event.Event; -import com.exactpro.th2.common.event.EventUtils; -import com.exactpro.th2.common.grpc.EventBatch; -import com.exactpro.th2.common.grpc.MessageBatch; -import com.exactpro.th2.common.grpc.MessageGroupBatch; -import com.exactpro.th2.common.grpc.RawMessageBatch; -import com.exactpro.th2.common.schema.factory.CommonFactory; -import com.exactpro.th2.common.schema.message.MessageRouter; -import com.exactpro.th2.common.schema.message.SubscriberMonitor; -import com.fasterxml.jackson.core.JsonProcessingException; - -public class Main { - - private static final Logger LOGGER = LoggerFactory.getLogger(Main.class); - - public static void main(String[] args) { - Deque resources = new ConcurrentLinkedDeque<>(); - Lock lock = new ReentrantLock(); - Condition condition = lock.newCondition(); - configureShutdownHook(resources, lock, condition); - try { - var factory = CommonFactory.createFromArguments(args); - resources.add(factory); - setLiveness(true); - - var configuration = factory.getCustomConfiguration(CsvCodecConfiguration.class); - MessageRouter messageGroupRouter = factory.getMessageRouterMessageGroupBatch(); - MessageRouter eventBatchRouter = factory.getEventBatchRouter(); - var rootEvent = createRootEvent(configuration); - eventBatchRouter.send(EventBatch.newBuilder().addEvents(rootEvent).build()); - - var codec = new CsvCodec(messageGroupRouter, eventBatchRouter, rootEvent.getId(), configuration); - SubscriberMonitor monitor = Objects.requireNonNull(messageGroupRouter.subscribeAll(codec, CsvCodec.DECODE_IN_ATTRIBUTE), "Cannot subscribe to raw queues"); - resources.add(monitor::unsubscribe); - - setReadiness(true); - awaitShutdown(lock, condition); - } catch (Exception e) { - LOGGER.error("Error occurred. Exit the program", e); - System.exit(-1); - } - } - - private static void awaitShutdown(Lock lock, Condition condition) throws InterruptedException { - try { - lock.lock(); - LOGGER.info("Wait shutdown"); - condition.await(); - LOGGER.info("App shutdown"); - } finally { - lock.unlock(); - } - } - - private static void configureShutdownHook(Deque resources, Lock lock, Condition condition) { - Runtime.getRuntime().addShutdownHook(new Thread(() -> shutDown(resources, lock, condition), "Shutdown hook")); - } - - private static void shutDown(Deque resources, Lock lock, Condition condition) { - LOGGER.info("Shutdown start"); - setReadiness(false); - try { - lock.lock(); - condition.signalAll(); - } finally { - lock.unlock(); - } - - resources.descendingIterator().forEachRemaining(resource -> { - try { - resource.close(); - } catch (Exception e) { - LOGGER.error(e.getMessage(), e); - } - }); - setLiveness(false); - LOGGER.info("Shutdown end"); - } - - private static com.exactpro.th2.common.grpc.Event createRootEvent(CsvCodecConfiguration cfg) throws JsonProcessingException { - String displayName = Objects.requireNonNull(cfg.getDisplayName(), "Display name should not be null"); - return Event.start() - .type("CodecCsv") - .name(displayName + "_" + System.currentTimeMillis()) - .bodyData(EventUtils.createMessageBean("Root event for CSV coded. All errors during decoding will be attached here")) - .toProtoEvent(null); - } -} diff --git a/src/main/java/com/exactpro/th2/codec/csv/cfg/CsvCodecConfiguration.java b/src/main/java/com/exactpro/th2/codec/csv/cfg/CsvCodecConfiguration.java index 29cb74c..729223f 100644 --- a/src/main/java/com/exactpro/th2/codec/csv/cfg/CsvCodecConfiguration.java +++ b/src/main/java/com/exactpro/th2/codec/csv/cfg/CsvCodecConfiguration.java @@ -19,11 +19,12 @@ import java.nio.charset.StandardCharsets; import java.util.List; +import com.exactpro.th2.codec.api.IPipelineCodecSettings; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonPropertyDescription; import com.fasterxml.jackson.databind.annotation.JsonDeserialize; -public class CsvCodecConfiguration { +public class CsvCodecConfiguration implements IPipelineCodecSettings { @JsonProperty("default-header") @JsonPropertyDescription("The default header that will be used for parsing received batch if no header found in the batch") private List defaultHeader; diff --git a/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java b/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java index a7faac9..b0611be 100644 --- a/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java +++ b/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java @@ -19,68 +19,45 @@ import static org.junit.jupiter.api.Assertions.assertAll; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.argThat; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; import java.io.IOException; import java.nio.charset.StandardCharsets; -import java.util.ArrayList; import java.util.List; import java.util.function.Supplier; import org.apache.commons.lang3.StringUtils; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.function.Executable; -import org.mockito.ArgumentCaptor; -import org.mockito.Mockito; +import com.exactpro.th2.codec.DecodeException; import com.exactpro.th2.codec.csv.cfg.CsvCodecConfiguration; import com.exactpro.th2.common.grpc.AnyMessage; -import com.exactpro.th2.common.grpc.Event; -import com.exactpro.th2.common.grpc.EventBatch; -import com.exactpro.th2.common.grpc.EventID; import com.exactpro.th2.common.grpc.ListValue; import com.exactpro.th2.common.grpc.Message; import com.exactpro.th2.common.grpc.MessageGroup; -import com.exactpro.th2.common.grpc.MessageGroupBatch; import com.exactpro.th2.common.grpc.MessageID; import com.exactpro.th2.common.grpc.RawMessage; import com.exactpro.th2.common.grpc.RawMessage.Builder; import com.exactpro.th2.common.grpc.RawMessageMetadata; import com.exactpro.th2.common.grpc.Value; -import com.exactpro.th2.common.schema.message.MessageRouter; import com.google.protobuf.ByteString; class TestCsvCodec { - @SuppressWarnings("unchecked") - private final MessageRouter routerMock = Mockito.mock(MessageRouter.class); - @SuppressWarnings("unchecked") - private final MessageRouter eventRouterMock = Mockito.mock(MessageRouter.class); - @Nested class TestPositive { @Test - void decodeArrayWithDifferentLength() throws IOException { - CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("A,B, , ,", "1,2,3,4")) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); +void decodeArrayWithDifferentLength() throws IOException { + CsvCodecConfiguration configuration = new CsvCodecConfiguration(); + configuration.setValidateLength(false); + CsvCodec codec = createCodec(configuration); + MessageGroup group = MessageGroup.newBuilder() + .addMessages(createCsvMessage("A,B, , ,", "1,2,3,4")) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -116,20 +93,13 @@ void decodeArrayWithDifferentLength() throws IOException { @Test void decodeArrayInEnd() throws IOException { - CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("A,B,C ,", "1,2,3")) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + CsvCodecConfiguration configuration = new CsvCodecConfiguration(); + configuration.setValidateLength(false); + CsvCodec codec = createCodec(configuration); + MessageGroup group = MessageGroup.newBuilder() + .addMessages(createCsvMessage("A,B,C ,", "1,2,3")) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -165,19 +135,10 @@ void decodeArrayInEnd() throws IOException { @Test void decodeArrayInMiddle() throws IOException { CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("A,B, ,C", "1,2,3,4")) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages(createCsvMessage("A,B, ,C", "1,2,3,4")) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -212,21 +173,13 @@ void decodeArrayInMiddle() throws IOException { } @Test - void decodesDataAndSkipsHeader() throws IOException { + void decodesDataAndSkipsHeader() { CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() + MessageGroup group = MessageGroup.newBuilder() .addMessages(createCsvMessage("A,B,C", "1,2,3")) - ).build(); - codec.handler("", batch); + .build(); - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -256,22 +209,13 @@ void decodesDataAndSkipsHeader() throws IOException { } @Test - void trimsEndOfTheLine() throws IOException { + void trimsEndOfTheLine() { CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("A,B,C\n\r1,2,3\n")) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages(createCsvMessage("A,B,C\n\r1,2,3\n")) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -300,26 +244,17 @@ void trimsEndOfTheLine() throws IOException { } @Test - void decodesDataUsingDefaultHeader() throws IOException { + void decodesDataUsingDefaultHeader() { CsvCodecConfiguration configuration = new CsvCodecConfiguration(); configuration.setDefaultHeader(List.of("A", "B", "C")); CsvCodec codec = createCodec(configuration); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages( - createCsvMessage("1,2,3") - ) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages( + createCsvMessage("1,2,3") + ) + .build(); + MessageGroup value = codec.decode(group); assertEquals(1, value.getMessagesCount()); Message message = getMessage(value, 0); @@ -336,24 +271,15 @@ void decodesDataUsingDefaultHeader() throws IOException { } @Test - void decodesDataWithEscapedCharacters() throws IOException { + void decodesDataWithEscapedCharacters() { CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages( - createCsvMessage("A,B", "\"1,2\",\"\"\"value\"\"\"") - ) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages( + createCsvMessage("A,B", "\"1,2\",\"\"\"value\"\"\"") + ) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -382,26 +308,17 @@ void decodesDataWithEscapedCharacters() throws IOException { } @Test - void decodesDataCustomDelimiter() throws IOException { + void decodesDataCustomDelimiter() { CsvCodecConfiguration configuration = new CsvCodecConfiguration(); configuration.setDelimiter(';'); CsvCodec codec = createCodec(configuration); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages( - createCsvMessage("A;B", "1,2;3") - ) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages( + createCsvMessage("A;B", "1,2;3") + ) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -430,24 +347,15 @@ void decodesDataCustomDelimiter() throws IOException { } @Test - void trimsWhitespacesDuringDecoding() throws IOException { + void trimsWhitespacesDuringDecoding() { CsvCodec codec = createCodec(); - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages( - createCsvMessage("A, B, C", "1, , 3 3") - ) - ).build(); - codec.handler("", batch); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); - - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + MessageGroup group = MessageGroup.newBuilder() + .addMessages( + createCsvMessage("A, B, C", "1, , 3 3") + ) + .build(); + MessageGroup value = codec.decode(group); assertEquals(2, value.getMessagesCount()); Message header = getMessage(value, 0); @@ -480,63 +388,33 @@ void trimsWhitespacesDuringDecoding() throws IOException { @Nested class TestNegative { @Test - void reportsErrorIfNotDataFound() throws IOException { + void reportsErrorIfNotDataFound() { CsvCodec codec = createCodec(); - codec.handler("", MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("")) - ).build()); - - assertAll( - () -> verify(routerMock).sendAll(argThat(it -> it.getGroupsCount() == 1 && it.getGroups(0).getMessagesCount() == 0 ), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)), - () -> verify(eventRouterMock).send(any()) - ); + Assertions.assertThrows(DecodeException.class, () -> + codec.decode(MessageGroup.newBuilder().addMessages(createCsvMessage("")).build())); } @Test - void reportsErrorIfRawDataIsEmpty() throws IOException { + void reportsErrorIfRawDataIsEmpty() { CsvCodec codec = createCodec(); - codec.handler("", MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() + + Assertions.assertThrows(DecodeException.class, () -> + codec.decode(MessageGroup.newBuilder() .addMessages(createCsvMessage("A,B,C")) .addMessages(createCsvMessage("")) - ).build()); - - assertAll( - () -> verify(routerMock).sendAll( - argThat(batch -> batch.getGroupsCount() == 1 - && "Csv_Header".equals(getMessage(batch.getGroups(0), 0).getMetadata().getMessageType())), - eq(CsvCodec.DECODE_OUT_ATTRIBUTE)), - () -> verify(eventRouterMock).send(any()) + .build()) ); } @Test - void reportsErrorIfDefaultHeaderAndDataHaveDifferentSize() throws IOException { + void reportsErrorIfDefaultHeaderAndDataHaveDifferentSize() { CsvCodecConfiguration configuration = new CsvCodecConfiguration(); configuration.setDefaultHeader(List.of("A", "B")); CsvCodec codec = createCodec(configuration); - codec.handler("", MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("1,2,3")) - ).build()); - - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - assertAll( - () -> verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)), - () -> verify(eventRouterMock).send(any()) - ); - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup messageBatch = actualBatch.getGroups(0); - assertEquals(1, messageBatch.getMessagesCount(), () -> "Batch: " + messageBatch); - Message message = getMessage(messageBatch, 0); - assertFieldCount(2, message); - assertAll( - () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A: " + message)), - () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B: " + message)) + assertThrows(DecodeException.class, () -> + codec.decode(MessageGroup.newBuilder() + .addMessages(createCsvMessage("1,2,3")).build()) ); } } @@ -546,7 +424,7 @@ private CsvCodec createCodec() { } private CsvCodec createCodec(CsvCodecConfiguration configuration) { - return new CsvCodec(routerMock, eventRouterMock, EventID.newBuilder().setId("test").build(), configuration); + return new CsvCodec(configuration); } private AnyMessage createCsvMessage(String... data) {