Additional properties related to message corrupetion strategies. Avoi…

…d disconnect loop by skipping corrupted messages during recovery from cradle.
th2-net · Oct 23, 2024 · 730cce2 · 730cce2
1 parent 0d69bcb
commit 730cce2
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 1 deletion.
diff --git a/src/main/java/com/exactpro/th2/FixHandler.java b/src/main/java/com/exactpro/th2/FixHandler.java
@@ -1496,6 +1496,7 @@ private void transformProcessor(
         strategyState.transformIfCondition(
             x -> x <= config.getNumberOfTimesToTransform(),
             () -> {
+                metadata.put("OriginalMessageType", msgTypeField.getValue());
                 messageTransformer.transformWithoutResults(message, transformation.getCombinedActions());
                 if(transformation.getNewPassword() != null) {
                     if(transformation.getEncryptKey() != null) {
@@ -1694,6 +1695,7 @@ private Map<String, String> corruptMessageStructureProcessor(ByteBuf message, Ma
         }
 
         state.getMessageCorrupted().set(true);
+        metadata.put("isCorruptedMessage", "Y");
 
         return null;
     }
@@ -1720,6 +1722,7 @@ private Map<String, String> adjustSendingTimeStrategyProcessor(ByteBuf message,
         AdjustSendingTimeConfiguration config = strategy.getAdjustSendingTimeConfiguration();
         metadata.put("sendingTimeUpdated", "Y");
         metadata.put("sendingTimeUpdateSeconds", Long.toString(config.getAdjustDuration().toSeconds()));
+        metadata.put("sendingTimeUpdateSign", config.getSubstract() ? "-" : "+");
         metadata.put(ENCODE_MODE_PROPERTY_NAME, DIRTY_ENCODE_MODE_NAME);
 
         updateSendingTime(message, config.getAdjustDuration(), config.getSubstract());

diff --git a/src/main/kotlin/com/exactpro/th2/conn/dirty/fix/MessageLoader.kt b/src/main/kotlin/com/exactpro/th2/conn/dirty/fix/MessageLoader.kt
@@ -234,7 +234,11 @@ class MessageLoader(
             )
 
             while (iterator.hasNext()) {
-                val message = Unpooled.buffer().writeBytes(iterator.next().message.bodyRaw.toByteArray())
+                val next = iterator.next().message
+                if(next.messagePropertiesMap.getOrDefault("isCorruptedMessage", "N") == "Y") {
+                    continue
+                }
+                val message = Unpooled.buffer().writeBytes(next.bodyRaw.toByteArray())
                 K_LOGGER.info { "Sending message to recovery processor: ${message.toString(US_ASCII)}" }
                 if (!processMessage(message)) break
             }