Add new configuration parameter: keepOpenPullingTimeout

th2-net · Nov 22, 2024 · 61b35e0 · 61b35e0
1 parent f46103f
commit 61b35e0
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 1 deletion.
diff --git a/README.md b/README.md
@@ -124,6 +124,7 @@ spec:
 #      * 0: no compression
 #      * 1: best speed
 #      * 9: best compression
+#   keepOpenPullingTimeout: 100 # time to wait in between attempts to get new data from data storage when `keepOpen` request parameter is used.
 
   pins: # pins are used to communicate with codec components to parse message data
     - name: to_codec
@@ -182,6 +183,7 @@ spec:
 #   validateCradleData: false # validate data loaded from cradle. NOTE: Enabled validation affect performance 
 #   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)
+#   keepOpenPullingTimeout: 100 # time to wait in between attempts to get new data from data storage when `keepOpen` request parameter is used.
 
 
   # pins are used to communicate with codec components to parse message data

diff --git a/app/src/main/kotlin/com/exactpro/th2/lwdataprovider/configuration/Configuration.kt b/app/src/main/kotlin/com/exactpro/th2/lwdataprovider/configuration/Configuration.kt
@@ -49,6 +49,7 @@ class CustomConfigurationClass(
     @JsonDeserialize(using = ByteSizeDeserializer::class)
     val batchSizeBytes: Int? = null,
     val downloadTaskTTL: Long? = null,
+    val keepOpenPullingTimeoutMs: Long? = null
 )
 
 class Configuration(customConfiguration: CustomConfigurationClass) {
@@ -79,6 +80,8 @@ class Configuration(customConfiguration: CustomConfigurationClass) {
     val gzipCompressionLevel: Int = VariableBuilder.getVariable(customConfiguration::gzipCompressionLevel, -1)
     val batchSizeBytes: Int = VariableBuilder.getVariable(customConfiguration::batchSizeBytes, 256 * 1024)
     val downloadTaskTTL: Long = VariableBuilder.getVariable(customConfiguration::downloadTaskTTL, TimeUnit.HOURS.toMillis(1))
+    val keepOpenPullingTimeoutMs: Long = VariableBuilder.getVariable(customConfiguration::keepOpenPullingTimeoutMs, 10)
+
     init {
         require(bufferPerQuery <= maxBufferDecodeQueue) {
             "buffer per queue ($bufferPerQuery) must be less or equal to the total buffer size ($maxBufferDecodeQueue)"

diff --git a/app/src/main/kotlin/com/exactpro/th2/lwdataprovider/handlers/SearchMessagesHandler.kt b/app/src/main/kotlin/com/exactpro/th2/lwdataprovider/handlers/SearchMessagesHandler.kt
@@ -140,6 +140,7 @@ class SearchMessagesHandler(
                         val allLoaded = hashSetOf<Stream>()
                         do {
                             val continuePulling = pullUpdates(request, order, sink, allLoaded)
+                            Thread.sleep(configuration.keepOpenPullingTimeoutMs)
                         } while (continuePulling)
                     }
                 }
@@ -293,13 +294,14 @@ class SearchMessagesHandler(
                         }
                         val lastTimestamp: Instant = request.endTimestamp
                         val allGroupLoaded = hashSetOf<String>()
-                        // TODO: Maybe we need to wait between pulling attempts to not overwhelm storage with requests
+
                         do {
                             val keepPulling = pullUpdates(request, lastTimestamp, sink, parameters, allGroupLoaded)
                             sink.canceled?.apply {
                                 logger.info { "request canceled: $message" }
                                 return@use
                             }
+                            Thread.sleep(configuration.keepOpenPullingTimeoutMs)
                         } while (keepPulling)
                     }
                 } catch (ex: Exception) {