[Ts-1083] raw-parsed-toggle

Dockerfile

@@ -1,4 +1,4 @@
-FROM gradle:6.6-jdk11 AS build
+FROM gradle:7.1-jdk11 AS build
 ARG Prelease_version=0.0.0
 COPY ./ .
 RUN gradle clean build dockerPrepare -Prelease_version=${Prelease_version}

README.md

@@ -1,4 +1,4 @@
-# Lightweight data provider (1.1.0)
+# Lightweight data provider (1.1.1)
 
 # Overview
 This component serves as a data provider for [th2-data-services](https://github.com/th2-net/th2-data-services). It will connect to the cassandra database via [cradle api](https://github.com/th2-net/cradleapi) and expose the data stored in there as REST resources.
@@ -29,6 +29,7 @@ This component is similar to [rpt-data-provider](https://github.com/th2-net/th2-
 - `resultCountLimit` - number - Sets the maximum amount of messages to return. Defaults to `null (unlimited)`.
 - `endTimestamp` - number, unix timestamp in milliseconds - Sets the timestamp to which the search will be performed, starting with `startTimestamp`. When `searchDirection` is `previous`, `endTimestamp` must be less then `startTimestamp`. Defaults to `null` (search can be stopped after reaching `resultCountLimit`).
 - `onlyRaw` - boolean - Disabling decoding messages. If it is true, message body will be empty in all messages. Default `false` 
+- `responseFormats` - text, accepts multiple values - sets response formats. Possible values: BASE_64, PARSED. default value - BASE_64 & PARSED.
 
 
 Elements in channel match the format sse:
@@ -58,6 +59,8 @@ spec:
 #   maxBufferDecodeQueue: 10000 # buffer size for messages that sent to decode but anwers hasn't been received 
 #   decodingTimeout: 60000 # timeout expecting answers from codec. 
 #   batchSize: 100 # batch size from codecs
+#   codecUsePinAttributes: true # send raw message to specified codec (true) or send to all codecs (false) 
+#   responseFormats: string list # resolve data for selected formats only. (allowed values: BASE_64, PARSED)
 
 
   pins: # pins are used to communicate with codec components to parse message data
@@ -92,4 +95,4 @@ spec:
       requests:
         memory: 300Mi
         cpu: 50m
-```
+```

build.gradle

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ * Copyright 2009-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.
@@ -51,7 +51,7 @@ repositories {
 }
 
 dependencies {
-    api platform('com.exactpro.th2:bom:3.0.0')
+    api platform('com.exactpro.th2:bom:3.2.0')
 
     implementation 'org.slf4j:slf4j-api'
     implementation 'org.slf4j:slf4j-log4j12'
@@ -66,9 +66,9 @@ dependencies {
     implementation('net.jpountz.lz4:lz4:1.3.0') {
         because('cassandra driver requires lz4 impl in classpath for compression')
     }
-    implementation 'com.exactpro.th2:grpc-data-provider:1.1.0'
+    implementation 'com.exactpro.th2:grpc-data-provider:1.1.0-TS-1083-response-formats-2759013731-SNAPSHOT'
 
-    implementation 'io.github.microutils:kotlin-logging:2.1.14'
+    implementation 'io.github.microutils:kotlin-logging:2.1.23'
 
     implementation('io.prometheus:simpleclient') {
         because('need add custom metrics to provider')
@@ -83,7 +83,7 @@ dependencies {
     }
 
     testImplementation 'org.junit.jupiter:junit-jupiter:5.8.2'
-    testImplementation 'org.assertj:assertj-core:3.12.2'
+    testImplementation 'org.assertj:assertj-core:3.23.1'
 }
 
 test {

gradle.properties

@@ -1,5 +1,5 @@
 ################################################################################
-# Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+# Copyright 2009-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.
@@ -16,6 +16,6 @@
 
 kotlin.code.style=official
 
-release_version=1.1.0
+release_version=1.1.1
 
 docker_image_name=

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 ################################################################################
-# Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+# Copyright 2009-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.
@@ -15,7 +15,7 @@
 ################################################################################
 
 #Tue Apr 14 11:21:33 MSK 2020
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.6-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.1-bin.zip
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStorePath=wrapper/dists

settings.gradle

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright 2009-2020 Exactpro (Exactpro Systems Limited)
+ * Copyright 2009-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.

src/main/kotlin/com/exactpro/th2/lwdataprovider/RabbitMqDecoder.kt

@@ -16,9 +16,7 @@
 
 package com.exactpro.th2.lwdataprovider
 
-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.message.MessageRouter
 import com.exactpro.th2.common.schema.message.QueueAttribute
 import com.exactpro.th2.lwdataprovider.configuration.Configuration
@@ -37,9 +35,13 @@ class RabbitMqDecoder(private val configuration: Configuration,
     companion object {
         private val logger = KotlinLogging.logger { }
     }
-
-    fun sendBatchMessage(batch: MessageGroupBatch, session: String) {
-        this.messageRouterRawBatch.send(batch, session, QueueAttribute.RAW.value)
+
+    fun sendBatchMessage(batch: MessageGroupBatch, session: String, codecUsePinAttributes: Boolean) {
+        if (codecUsePinAttributes) {
+            this.messageRouterRawBatch.send(batch, session, QueueAttribute.RAW.value)
+        } else {
+            this.messageRouterRawBatch.sendAll(batch, QueueAttribute.RAW.value)
+        }
     }
 
     fun registerMessage(message: RequestedMessageDetails) {

src/main/kotlin/com/exactpro/th2/lwdataprovider/RequestContext.kt

@@ -19,6 +19,7 @@ package com.exactpro.th2.lwdataprovider
 import com.exactpro.cradle.messages.StoredMessage
 import com.exactpro.th2.common.grpc.Message
 import com.exactpro.th2.common.grpc.RawMessage
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import com.exactpro.th2.lwdataprovider.entities.responses.LastScannedObjectInfo
 import io.prometheus.client.Histogram
 import mu.KotlinLogging
@@ -85,7 +86,7 @@ abstract class MessageRequestContext (
 
    fun allDataLoadedFromCradle() = allMessagesRequested.set(true)
 
-   abstract fun createMessageDetails(id: String, time: Long, storedMessage: StoredMessage, onResponse: () -> Unit = {}): RequestedMessageDetails;
+    abstract fun createMessageDetails(id: String, time: Long, storedMessage: StoredMessage, responseFormats: List<ResponseFormat>, onResponse: () -> Unit = {}): RequestedMessageDetails;
    abstract fun addStreamInfo();
 
    override fun onMessageSent() {
@@ -137,6 +138,7 @@ abstract class RequestedMessageDetails (
    @Volatile var time: Long,
    val storedMessage: StoredMessage,
    protected open val context: MessageRequestContext,
+   val responseFormats: List<ResponseFormat>,
    var parsedMessage: List<Message>? = null,
    var rawMessage: RawMessage? = null,
    private val onResponse: () -> Unit = {}

src/main/kotlin/com/exactpro/th2/lwdataprovider/configuration/Configuration.kt

@@ -16,6 +16,7 @@
 
 package com.exactpro.th2.lwdataprovider.configuration
 
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import java.util.*
 
 
@@ -31,6 +32,8 @@ class CustomConfigurationClass {
     val mode: String? = null
     val grpcBackPressure : Boolean? = null
     val bufferPerQuery: Int? = null
+    val codecUsePinAttributes: Boolean = true
+    val defaultResponseFormats: List<ResponseFormat> = listOf(ResponseFormat.ALL)
 }
 
 class Configuration(customConfiguration: CustomConfigurationClass) {
@@ -47,6 +50,9 @@ class Configuration(customConfiguration: CustomConfigurationClass) {
         customConfiguration.mode?.let { Mode.valueOf(it.uppercase(Locale.getDefault())) }, Mode.HTTP)
     val grpcBackPressure: Boolean = VariableBuilder.getVariable("grpcBackPressure", customConfiguration.grpcBackPressure, false)
     val bufferPerQuery: Int = VariableBuilder.getVariable("bufferPerQuery", customConfiguration.bufferPerQuery, 0)
+    val codecUsePinAttributes: Boolean = VariableBuilder.getVariable("codecUsePinAttributes", customConfiguration.codecUsePinAttributes, true)
+    val defaultResponseFormats: List<ResponseFormat> = VariableBuilder.getVariable("defaultResponseFormats",
+        customConfiguration.defaultResponseFormats, listOf(ResponseFormat.ALL))
 }
 
 enum class Mode {

src/main/kotlin/com/exactpro/th2/lwdataprovider/db/CradleMessageExtractor.kt

@@ -23,6 +23,7 @@ import com.exactpro.cradle.messages.StoredMessageId
 import com.exactpro.th2.common.grpc.MessageGroupBatch
 import com.exactpro.th2.common.grpc.RawMessage
 import com.exactpro.th2.common.message.plusAssign
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import com.exactpro.th2.lwdataprovider.MessageRequestContext
 import com.exactpro.th2.lwdataprovider.RabbitMqDecoder
 import com.exactpro.th2.lwdataprovider.RequestedMessageDetails
@@ -36,14 +37,15 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
 
     private val storage = cradleManager.storage
     private val batchSize = configuration.batchSize
+    private val codecUsePinAttributes = configuration.codecUsePinAttributes
 
     companion object {
         private val logger = KotlinLogging.logger { }
     }
 
     fun getStreams(): Collection<String> = storage.streams
 
-    fun getMessages(filter: StoredMessageFilter, requestContext: MessageRequestContext) {
+    fun getMessages(filter: StoredMessageFilter, requestContext: MessageRequestContext, responseFormats: List<ResponseFormat>) {
 
         var msgCount = 0
         val time = measureTimeMillis { 
@@ -65,7 +67,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
                 msgId = storedMessage.id
                 val id = storedMessage.id.toString()
                 val decodingStep = requestContext.startStep("decoding")
-                val tmp = requestContext.createMessageDetails(id, 0, storedMessage) { decodingStep.finish() }
+                val tmp = requestContext.createMessageDetails(id, 0, storedMessage, responseFormats) { decodingStep.finish() }
                 messageBuffer.add(tmp)
                 ++msgBufferCount
                 tmp.rawMessage = requestContext.startStep("raw_message_parsing").use { RawMessage.parseFrom(storedMessage.content) }.also {
@@ -79,8 +81,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
                         decoder.registerMessage(it)
                         requestContext.registerMessage(it)
                     }
-                    decoder.sendBatchMessage(builder.build(), sessionName)
-
+                    decoder.sendBatchMessage(builder.build(), sessionName, codecUsePinAttributes)
                     messageBuffer.clear()
                     builder.clear()
                     msgCount += msgBufferCount
@@ -101,7 +102,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
             }
 
             if (msgBufferCount > 0) {
-                decoder.sendBatchMessage(builder.build(), sessionName)
+                decoder.sendBatchMessage(builder.build(), sessionName, codecUsePinAttributes)
 
                 val sendingTime = System.currentTimeMillis()
                 messageBuffer.forEach { 
@@ -121,7 +122,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
     }
 
 
-    fun getRawMessages(filter: StoredMessageFilter, requestContext: MessageRequestContext) {
+    fun getRawMessages(filter: StoredMessageFilter, requestContext: MessageRequestContext, responseFormats: List<ResponseFormat>) {
 
         var msgCount = 0
         val time = measureTimeMillis {
@@ -136,7 +137,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
                 }
                 msgId = storedMessageBatch.id
                 val id = storedMessageBatch.id.toString()
-                val tmp = requestContext.createMessageDetails(id, time, storedMessageBatch)
+                val tmp = requestContext.createMessageDetails(id, time, storedMessageBatch, responseFormats)
                 tmp.rawMessage = RawMessage.parseFrom(storedMessageBatch.content)
                 tmp.responseMessage()
                 msgCount++
@@ -166,7 +167,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
 
             val time = System.currentTimeMillis()
             val decodingStep = if (onlyRaw) null else requestContext.startStep("decoding")
-            val tmp = requestContext.createMessageDetails(message.id.toString(), time, message) { decodingStep?.finish() }
+            val tmp = requestContext.createMessageDetails(message.id.toString(), time, message, emptyList()) { decodingStep?.finish() }
             tmp.rawMessage = RawMessage.parseFrom(message.content)
             requestContext.loadedMessages += 1
 
@@ -178,7 +179,7 @@ class CradleMessageExtractor(configuration: Configuration, private val cradleMan
                     .build()
                 decoder.registerMessage(tmp)
                 requestContext.registerMessage(tmp)
-                decoder.sendBatchMessage(msgBatch, message.streamName)
+                decoder.sendBatchMessage(msgBatch, message.streamName, codecUsePinAttributes)
             }
 
         }

src/main/kotlin/com/exactpro/th2/lwdataprovider/entities/requests/GetMessageRequest.kt

@@ -17,17 +17,20 @@
 package com.exactpro.th2.lwdataprovider.entities.requests
 
 import com.exactpro.th2.common.grpc.MessageID
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import com.exactpro.th2.lwdataprovider.grpc.toStoredMessageId
 
 
 data class GetMessageRequest(
     val msgId: String,
-    val onlyRaw: Boolean
+    val onlyRaw: Boolean,
+    val responseFormats: List<ResponseFormat>? = listOf(ResponseFormat.ALL)
 ) {
 
     constructor(msgId: String, parameters: Map<String, List<String>>) : this(
         msgId = msgId,
-        onlyRaw = parameters["onlyRaw"]?.firstOrNull()?.toBoolean() ?: false
+        onlyRaw = parameters["onlyRaw"]?.firstOrNull()?.toBoolean() ?: false,
+        responseFormats = parameters["responseFormats"]?.map { x -> ResponseFormat.valueOf(x) }
     )
 
     constructor(msgId: MessageID) : this(

src/main/kotlin/com/exactpro/th2/lwdataprovider/entities/requests/SseMessageSearchRequest.kt

@@ -20,14 +20,14 @@ import com.exactpro.cradle.Direction
 import com.exactpro.cradle.TimeRelation
 import com.exactpro.cradle.messages.StoredMessageId
 import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import com.exactpro.th2.dataprovider.grpc.MessageStreamPointer
 import com.exactpro.th2.lwdataprovider.entities.exceptions.InvalidRequestException
 import com.exactpro.th2.lwdataprovider.grpc.toInstant
 import com.exactpro.th2.lwdataprovider.grpc.toProviderMessageStreams
 import com.exactpro.th2.lwdataprovider.grpc.toProviderRelation
 import com.exactpro.th2.lwdataprovider.grpc.toStoredMessageId
 import java.time.Instant
-import kotlin.streams.toList
 
 data class SseMessageSearchRequest(
     val startTimestamp: Instant?,
@@ -39,7 +39,10 @@ data class SseMessageSearchRequest(
     val attachedEvents: Boolean,
     val lookupLimitDays: Int?,
     val resumeFromIdsList: List<StoredMessageId>?,
-    val onlyRaw: Boolean
+    @Deprecated("Use responseFormats instead", ReplaceWith("responseFormats",
+        "com.exactpro.th2.lwdataprovider.entities.requests"), DeprecationLevel.WARNING)
+    val onlyRaw: Boolean,
+    val responseFormats: List<ResponseFormat>? = listOf(ResponseFormat.ALL)
 ) {
 
     companion object {
@@ -86,7 +89,8 @@ data class SseMessageSearchRequest(
         keepOpen = parameters["keepOpen"]?.firstOrNull()?.toBoolean() ?: false,
         attachedEvents = parameters["attachedEvents"]?.firstOrNull()?.toBoolean() ?: false,
         lookupLimitDays = parameters["lookupLimitDays"]?.firstOrNull()?.toInt(),
-        onlyRaw = parameters["onlyRaw"]?.firstOrNull()?.toBoolean() ?: false
+        onlyRaw = parameters["onlyRaw"]?.firstOrNull()?.toBoolean() ?: false,
+        responseFormats = parameters["responseFormats"]?.map { x -> ResponseFormat.valueOf(x) }
     )
 
     constructor(grpcRequest: MessageSearchRequest) : this(
@@ -99,7 +103,8 @@ data class SseMessageSearchRequest(
         keepOpen = if (grpcRequest.hasKeepOpen()) grpcRequest.keepOpen.value else false,
         attachedEvents = false, // disabled
         lookupLimitDays = null,
-        onlyRaw = false // NOT SUPPORTED in GRPC
+        onlyRaw = false, // NOT SUPPORTED in GRPC
+        responseFormats = grpcRequest.responseFormatsList
     )
 
     private fun checkEndTimestamp() {

src/main/kotlin/com/exactpro/th2/lwdataprovider/grpc/GrpcDataProviderImpl.kt

@@ -146,7 +146,7 @@ open class GrpcDataProviderImpl(
         val grpcResponseHandler = GrpcResponseHandler(queue)
         val context = GrpcMessageRequestContext(grpcResponseHandler, maxMessagesPerRequest = configuration.bufferPerQuery)
         val loadingStep = context.startStep("messages_loading")
-        searchMessagesHandler.loadMessages(requestParams, context)
+        searchMessagesHandler.loadMessages(requestParams, context, configuration)
         try {
             processResponse(responseObserver, grpcResponseHandler, context, loadingStep::finish) { it.message }
         } catch (ex: Exception) {

src/main/kotlin/com/exactpro/th2/lwdataprovider/grpc/GrpcMessageRequestContext.kt

@@ -19,6 +19,7 @@ package com.exactpro.th2.lwdataprovider.grpc
 import com.exactpro.cradle.messages.StoredMessage
 import com.exactpro.th2.common.grpc.Message
 import com.exactpro.th2.common.grpc.RawMessage
+import com.exactpro.th2.dataprovider.grpc.MessageSearchRequest.ResponseFormat
 import com.exactpro.th2.dataprovider.grpc.MessageSearchResponse
 import com.exactpro.th2.dataprovider.grpc.MessageStreamPointers
 import com.exactpro.th2.lwdataprovider.GrpcResponseHandler
@@ -42,8 +43,8 @@ class GrpcMessageRequestContext (
     maxMessagesPerRequest = maxMessagesPerRequest) {
 
 
-    override fun createMessageDetails(id: String, time: Long, storedMessage: StoredMessage, onResponse: () -> Unit): GrpcRequestedMessageDetails {
-        return GrpcRequestedMessageDetails(id, time, storedMessage, this, onResponse)
+    override fun createMessageDetails(id: String, time: Long, storedMessage: StoredMessage, responseFormats: List<ResponseFormat>, onResponse: () -> Unit): GrpcRequestedMessageDetails {
+        return GrpcRequestedMessageDetails(id, time, storedMessage, this, responseFormats, onResponse)
     }
 
     override fun addStreamInfo() {
@@ -58,10 +59,11 @@ class GrpcRequestedMessageDetails(
     time: Long,
     storedMessage: StoredMessage,
     override val context: GrpcMessageRequestContext,
+    responseFormats: List<ResponseFormat>,
     onResponse: () -> Unit,
     parsedMessage: List<Message>? = null,
     rawMessage: RawMessage? = null
-) : RequestedMessageDetails(id, time, storedMessage, context, parsedMessage, rawMessage, onResponse) {
+) : RequestedMessageDetails(id, time, storedMessage, context, responseFormats, parsedMessage, rawMessage, onResponse) {
 
     override fun responseMessageInternal() {
         val msg = GrpcMessageProducer.createMessage(this)