diff --git a/.github/workflows/dev-docker-publish.yml b/.github/workflows/dev-docker-publish.yml index 6433f4f..6f77755 100644 --- a/.github/workflows/dev-docker-publish.yml +++ b/.github/workflows/dev-docker-publish.yml @@ -3,43 +3,17 @@ name: Dev build and publish Docker distributions to Github Container Registry gh on: push: branches-ignore: - - master - - version-* + - master + - version-* + - dependabot** + paths-ignore: + - README.md jobs: - build: - runs-on: ubuntu-20.04 - steps: - - uses: actions/checkout@v2 -# Prepare custom build version - - name: Get branch name - id: branch - run: echo ::set-output name=branch_name::${GITHUB_REF#refs/*/} - - name: Get release_version - id: ver - uses: christian-draeger/read-properties@1.0.1 - with: - path: gradle.properties - property: release_version - - name: Build custom release version - id: release_ver - run: echo ::set-output name=value::"${{ steps.ver.outputs.value }}-${{ steps.branch.outputs.branch_name }}-${{ github.run_id }}" - - name: Show custom release version - run: echo ${{ steps.release_ver.outputs.value }} -# Build and publish image - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v1 - - uses: docker/login-action@v1 - with: - registry: ghcr.io - username: ${{ github.repository_owner }} - password: ${{ secrets.CR_PAT }} - - run: echo "::set-output name=REPOSITORY_NAME::$(echo '${{ github.repository }}' | awk -F '/' '{print $2}')" - id: meta - - name: Build and push - id: docker_build - uses: docker/build-push-action@v2 - with: - push: true - tags: ghcr.io/${{ github.repository }}:${{ steps.release_ver.outputs.value }} - labels: com.exactpro.th2.${{ steps.meta.outputs.REPOSITORY_NAME }}=${{ steps.ver.outputs.value }} + build-job: + uses: th2-net/.github/.github/workflows/compound-java-dev.yml@main + with: + build-target: 'Docker' + docker-username: ${{ github.actor }} + secrets: + docker-password: ${{ secrets.GITHUB_TOKEN }} diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index 9274060..cec50a6 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -3,33 +3,18 @@ name: Build and publish Docker distributions to Github Container Registry ghcr.i on: push: branches: - - master - - version-* + - master + - version-* + paths: + - gradle.properties +# - package_info.json jobs: - build: - runs-on: ubuntu-20.04 - steps: - - uses: actions/checkout@v2 - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v1 - - uses: docker/login-action@v1 - with: - registry: ghcr.io - username: ${{ github.repository_owner }} - password: ${{ secrets.CR_PAT }} - - run: echo "::set-output name=REPOSITORY_NAME::$(echo '${{ github.repository }}' | awk -F '/' '{print $2}')" - id: meta - - name: Read version from gradle.properties - id: read_property - uses: christian-draeger/read-properties@1.0.1 - with: - path: ./gradle.properties - property: release_version - - name: Build and push - id: docker_build - uses: docker/build-push-action@v2 - with: - push: true - tags: ghcr.io/${{ github.repository }}:${{ steps.read_property.outputs.value }} - labels: com.exactpro.th2.${{ steps.meta.outputs.REPOSITORY_NAME }}=${{ steps.read_property.outputs.value }} + build-job: + uses: th2-net/.github/.github/workflows/compound-java.yml@main + with: + build-target: 'Docker' + docker-username: ${{ github.actor }} + secrets: + docker-password: ${{ secrets.GITHUB_TOKEN }} + \ No newline at end of file diff --git a/README.md b/README.md index 5c83aa9..f6a7a92 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,22 @@ -# 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. +Output message type is taken from `th2.csv.override_message_type` property in input message. +If the property missing, the default value (`Csv_Message`) for output message type is used. 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 +59,7 @@ into ``` ## Settings -Csv codec has following parameters: +Csv codec has the following parameters: ```yaml default-header: [A, B, C] @@ -61,6 +67,7 @@ delimiter: ',' encoding: UTF-8 display-name: CodecCsv validate-length: true +publish-header: false ``` **default-header** - the default header for this codec. It will be used if no header found in the received batch. @@ -71,40 +78,64 @@ validate-length: true **display-name** - the name to set in the root event sent to the event store. All errors during decoding will be attached to that root event. 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. +**validate-length** - check if csv have different count of values against header's count. -The number of pins with each set of attributes is not limited. **Pins can use filters**. +**publish-header** - set to publish decoded header. The default value is `false`. ## 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 ++ Updated `bom`: `3.1.0` -> `4.1.0` ++ Updated `common`: `3.37.1` -> `3.44.0` + ### 3.2.1 + fixed: last array-field as simple value diff --git a/build.gradle b/build.gradle index 32a9903..c5a922e 100644 --- a/build.gradle +++ b/build.gradle @@ -2,6 +2,11 @@ plugins { id 'java-library' id 'application' id 'com.palantir.docker' version '0.25.0' + id "org.owasp.dependencycheck" version "7.4.4" +} + +dependencyCheck { + format='HTML' } ext { @@ -50,14 +55,16 @@ jar { } dependencies { - api platform('com.exactpro.th2:bom:3.1.0') + api platform('com.exactpro.th2:bom:4.1.0') - implementation 'com.exactpro.th2:common:3.37.1' - - implementation "org.slf4j:slf4j-log4j12" - implementation "org.slf4j:slf4j-api" + implementation 'com.exactpro.th2:common:3.44.0' + implementation 'com.exactpro.th2:codec:4.7.4' 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 +75,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..e0c860f --- /dev/null +++ b/src/main/java/com/exactpro/th2/codec/csv/CodecFactory.java @@ -0,0 +1,76 @@ +/* + * 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 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..a047bec 100644 --- a/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java +++ b/src/main/java/com/exactpro/th2/codec/csv/CsvCodec.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 Exactpro (Exactpro Systems Limited) + * Copyright 2020-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. @@ -25,53 +25,44 @@ 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.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.common.event.Event; -import com.exactpro.th2.common.event.Event.Status; -import com.exactpro.th2.common.event.EventUtils; import com.exactpro.th2.common.grpc.AnyMessage; -import com.exactpro.th2.common.grpc.EventBatch; -import com.exactpro.th2.common.grpc.EventID; import com.exactpro.th2.common.grpc.Message; import com.exactpro.th2.common.grpc.Message.Builder; 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.MessageMetadata; import com.exactpro.th2.common.grpc.RawMessage; import com.exactpro.th2.common.grpc.RawMessageMetadata; -import com.exactpro.th2.common.schema.message.MessageListener; -import com.exactpro.th2.common.schema.message.MessageRouter; +import com.exactpro.th2.common.message.MessageUtils; 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 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; - 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,50 +74,79 @@ 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; } - - send(batchBuilder.build()); - if (!errors.isEmpty()) { - reportErrors(errors); + if (!anyMessage.hasRawMessage()) { + LOGGER.error("Message should either have a raw or parsed message but has nothing: {}", anyMessage); + continue; + } + RawMessage rawMessage = anyMessage.getRawMessage(); + String protocol = rawMessage.getMetadata().getProtocol(); + if (!"".equals(protocol) && !"csv".equalsIgnoreCase(protocol)) { + LOGGER.error("Wrong protocol: message should have empty or 'csv' protocol but has {}", protocol); + continue; } - } catch (Exception ex) { - LOGGER.error("Cannot process batch for {}: {}", consumerTag, message, ex); + 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(); + } + + @NotNull + @Override + public MessageGroup decode(@NotNull MessageGroup messageGroup, @NotNull IReportingContext iReportingContext) { + return decode(messageGroup); + } + + 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) { RawMessageMetadata originalMetadata = rawMessage.getMetadata(); + final String outputMessageType = originalMetadata.getPropertiesOrDefault(OVERRIDE_MESSAGE_TYPE_PROP_NAME_LOWERCASE, CSV_MESSAGE_TYPE); + int currentIndex = 0; String[] header = defaultHeader; for (String[] strings : data) { @@ -147,12 +167,16 @@ private void decodeCsvData(Collection errors, MessageGroup.Builder LOGGER.debug("Set header to: " + Arrays.toString(strings)); } header = strings; - AnyMessage.Builder messageBuilder = groupBuilder.addMessagesBuilder(); - Builder headerMsg = Message.newBuilder(); - // Not set message type - setMetadata(originalMetadata, headerMsg, HEADER_MSG_TYPE, currentIndex); - headerMsg.putFields(HEADER_FIELD_NAME, ValueUtils.toValue(strings)); - messageBuilder.setMessage(headerMsg); + + if (configuration.isPublishHeader()) { + AnyMessage.Builder messageBuilder = groupBuilder.addMessagesBuilder(); + Builder headerMsg = Message.newBuilder(); + // Not set message type + setMetadata(originalMetadata, headerMsg, HEADER_MSG_TYPE, currentIndex); + headerMsg.putFields(HEADER_FIELD_NAME, ValueUtils.toValue(strings)); + messageBuilder.setMessage(headerMsg); + } + continue; } @@ -170,7 +194,7 @@ private void decodeCsvData(Collection errors, MessageGroup.Builder Builder builder = Message.newBuilder(); // Not set message type - setMetadata(originalMetadata, builder, CSV_MESSAGE_TYPE, currentIndex); + setMetadata(originalMetadata, builder, outputMessageType, currentIndex); int headerLength = header.length; int rowLength = strings.length; @@ -186,7 +210,6 @@ private void decodeCsvData(Collection errors, MessageGroup.Builder } } - messageBuilder.setMessage(builder); } } @@ -219,7 +242,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 +254,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,41 +275,10 @@ 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]; elements[i] = element.trim(); } } -} +} \ No newline at end of file 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..acfdcf4 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 @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 Exactpro (Exactpro Systems Limited) + * Copyright 2020-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. @@ -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; @@ -43,6 +44,10 @@ public class CsvCodecConfiguration { @JsonPropertyDescription("Set to validate length of columns or not") private boolean validateLength = true; + @JsonProperty("publish-header") + @JsonPropertyDescription("Set to enable header publication") + private boolean publishHeader = false; + public List getDefaultHeader() { return defaultHeader; } @@ -82,4 +87,12 @@ public boolean getValidateLength() { public void setValidateLength(boolean validateLength) { this.validateLength = validateLength; } + + public boolean isPublishHeader() { + return publishHeader; + } + + public void setPublishHeader(boolean publishHeader) { + this.publishHeader = publishHeader; + } } 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..317292c 100644 --- a/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java +++ b/src/test/java/com/exactpro/th2/codec/csv/TestCsvCodec.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 Exactpro (Exactpro Systems Limited) + * Copyright 2020-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. @@ -19,68 +19,47 @@ 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.Map; 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); + configuration.setPublishHeader(true); + 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); @@ -101,6 +80,7 @@ void decodeArrayWithDifferentLength() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> { @@ -116,20 +96,14 @@ 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); + configuration.setPublishHeader(true); + 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); @@ -150,6 +124,7 @@ void decodeArrayInEnd() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), @@ -165,19 +140,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); @@ -198,6 +164,7 @@ void decodeArrayInMiddle() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> { @@ -212,21 +179,13 @@ void decodeArrayInMiddle() throws IOException { } @Test - void decodesDataAndSkipsHeader() throws IOException { + void decodesDataAndSkipsHeader() { 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); + 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); @@ -247,6 +206,7 @@ void decodesDataAndSkipsHeader() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), @@ -256,22 +216,75 @@ void decodesDataAndSkipsHeader() throws IOException { } @Test - void trimsEndOfTheLine() throws IOException { - CsvCodec codec = createCodec(); + void skipsHeaderPublishing() { + final var config = new CsvCodecConfiguration(); + config.setPublishHeader(false); + CsvCodec codec = createCodec(config); + MessageGroup group = MessageGroup.newBuilder() + .addMessages(createCsvMessage("A,B,C", "1,2,3")) + .build(); + + MessageGroup value = codec.decode(group); + assertEquals(1, value.getMessagesCount()); + + Message message = getMessage(value, 0); + assertFieldCount(3, message); + + assertAll( + () -> assertAll("Current message: " + message, + () -> { + assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); + assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); + }, + () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), + () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), + () -> assertEquals("3", getFieldValue(message, "C", () -> "No field C. " + message)) + ) + ); + } - MessageGroupBatch batch = MessageGroupBatch.newBuilder() - .addGroups(MessageGroup.newBuilder() - .addMessages(createCsvMessage("A,B,C\n\r1,2,3\n")) - ).build(); - codec.handler("", batch); + @Test + void settingMessageTypeFromIncomingMessage() { + final var customType = "csv_test_type"; + + final var config = new CsvCodecConfiguration(); + config.setPublishHeader(false); + CsvCodec codec = createCodec(config); + final var csvMessage = createCsvMessage(Map.of("th2.csv.override_message_type", customType), "A,B,C", "1,2,3"); + + MessageGroup group = MessageGroup.newBuilder() + .addMessages(csvMessage) + .build(); + + MessageGroup value = codec.decode(group); + assertEquals(1, value.getMessagesCount()); - var captor = ArgumentCaptor.forClass(MessageGroupBatch.class); - verify(routerMock).sendAll(captor.capture(), eq(CsvCodec.DECODE_OUT_ATTRIBUTE)); + Message message = getMessage(value, 0); + assertFieldCount(3, message); - MessageGroupBatch actualBatch = captor.getValue(); - assertNotNull(actualBatch, "Did not capture any publication"); - assertEquals(1, actualBatch.getGroupsCount()); - MessageGroup value = actualBatch.getGroups(0); + assertAll( + () -> assertAll("Current message: " + message, + () -> { + assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); + assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals(customType, message.getMetadata().getMessageType()); + }, + () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), + () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), + () -> assertEquals("3", getFieldValue(message, "C", () -> "No field C. " + message)) + ) + ); + } + + @Test + void trimsEndOfTheLine() { + CsvCodec codec = createCodec(); + + 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); @@ -291,6 +304,7 @@ void trimsEndOfTheLine() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), @@ -300,26 +314,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); @@ -328,6 +333,7 @@ void decodesDataUsingDefaultHeader() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(1, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("2", getFieldValue(message, "B", () -> "No field B. " + message)), @@ -336,24 +342,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); @@ -374,6 +371,7 @@ void decodesDataWithEscapedCharacters() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1,2", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("\"value\"", getFieldValue(message, "B", () -> "No field B. " + message)) @@ -382,26 +380,18 @@ void decodesDataWithEscapedCharacters() throws IOException { } @Test - void decodesDataCustomDelimiter() throws IOException { + void decodesDataCustomDelimiter() { CsvCodecConfiguration configuration = new CsvCodecConfiguration(); configuration.setDelimiter(';'); + configuration.setPublishHeader(true); 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); @@ -422,6 +412,7 @@ void decodesDataCustomDelimiter() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1,2", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("3", getFieldValue(message, "B", () -> "No field B. " + message)) @@ -430,24 +421,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); @@ -468,6 +450,7 @@ void trimsWhitespacesDuringDecoding() throws IOException { () -> { assertEquals(1, message.getMetadata().getId().getSubsequenceCount()); assertEquals(2, message.getMetadata().getId().getSubsequence(0)); + assertEquals("Csv_Message", message.getMetadata().getMessageType()); }, () -> assertEquals("1", getFieldValue(message, "A", () -> "No field A. " + message)), () -> assertEquals("", getFieldValue(message, "B", () -> "No field B. " + message)), @@ -480,80 +463,57 @@ 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()) ); } } private CsvCodec createCodec() { - return createCodec(new CsvCodecConfiguration()); + final var configuration = new CsvCodecConfiguration(); + configuration.setPublishHeader(true); + return createCodec(configuration); } 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) { + return createCsvMessage(Map.of(), data); + } + + private AnyMessage createCsvMessage(Map metadataProps, String... data) { Builder builder = RawMessage.newBuilder() .setBody(ByteString.copyFrom(String.join(StringUtils.LF, data).getBytes(StandardCharsets.UTF_8))); RawMessageMetadata.Builder metadataBuilder = RawMessageMetadata.newBuilder() - .setId(MessageID.newBuilder().setSequence(System.nanoTime()).build()); + .setId(MessageID.newBuilder().setSequence(System.nanoTime()).build()) + .putAllProperties(metadataProps); builder.setMetadata(metadataBuilder.build()); return AnyMessage.newBuilder().setRawMessage(builder).build(); }