[GH-50] Attached message ids to send event (#51)

Provided `publishSentEvents` option.

Author: Nikita-Smirnov-Exactpro

README.md

@@ -31,6 +31,7 @@ When defined in main, they act as default values for all sessions, or as the par
 + **defaultHeaders** - map of default headers, and their values which will be applied to each request (existing headers
   are not affected, empty by default)
 + **auth** - basic authentication settings (`null` by default)
++ **publishSentEvents** - enables/disables publish of "message sent" events (`true` by default)
 
 ### Authentication configuration
 
@@ -152,6 +153,7 @@ spec:
     useTransport: false
     maxBatchSize: 1000
     maxFlushTime: 1000
+    publishSentEvents: true
     defaultHeaders:
       x-api-key: [ 'apikeywashere' ]
     auth:
@@ -185,7 +187,10 @@ spec:
 
 ### v2.4.0
 
++ [[GH-50] Attached message ids to send event](https://github.com/th2-net/th2-conn-http-client/issues/50)
+  + provided `publishSentEvents` option.
 + [[GH-44] Implemented multi-session feature](https://github.com/th2-net/th2-conn-http-client/issues/44)
+  + provided `sessions` option.
 + Updated:
   + th2 gradle plugin: `0.3.14` (bom: `4.14.3`)
   + kotlin: `2.3.0`

src/main/kotlin/com/exactpro/th2/http/client/Application.kt

@@ -26,7 +26,6 @@ import com.exactpro.th2.common.schema.message.MessageRouter
 import com.exactpro.th2.common.schema.message.QueueAttribute.RAW
 import com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.GroupBatch
 import com.exactpro.th2.common.utils.event.EventBatcher
-import com.exactpro.th2.common.utils.event.storeEvent
 import com.exactpro.th2.common.utils.event.transport.toProto
 import com.exactpro.th2.common.utils.message.RAW_GROUP_SELECTOR
 import com.exactpro.th2.common.utils.message.RawMessageBatcher
@@ -42,6 +41,8 @@ import com.exactpro.th2.http.client.api.IStateManager
 import com.exactpro.th2.http.client.api.IStateManager.StateManagerContext
 import com.exactpro.th2.http.client.api.impl.BasicRequestHandler
 import com.exactpro.th2.http.client.api.impl.BasicStateManager
+import com.exactpro.th2.http.client.util.publishSentEvents
+import com.exactpro.th2.http.client.util.storeEvent
 import com.exactpro.th2.http.client.util.toPrettyString
 import com.exactpro.th2.http.client.util.toProtoMessage
 import com.exactpro.th2.http.client.util.toTransportMessage
@@ -91,29 +92,6 @@ class Application(
                 onBatch = eventRouter::send
             ).also { registerResource("event batcher", it::close) }
 
-            val onError: (Throwable) -> Unit = {
-                eventBatcher.storeEvent(rootEventId, "Batching problem: ${it.message}", "Message batching problem", it)
-            }
-
-            lateinit var transportMB: MessageBatcher
-            lateinit var protoMB: RawMessageBatcher
-            if (useTransport) {
-                transportMB =
-                    MessageBatcher(
-                        maxBatchSize,
-                        maxFlushTime,
-                        book,
-                        GROUP_SELECTOR,
-                        executor,
-                        onError,
-                        transportMR::send
-                    ).also { registerResource("transport message batcher", it::close) }
-            } else {
-                protoMB = RawMessageBatcher(maxBatchSize, maxFlushTime, RAW_GROUP_SELECTOR, executor, onError) {
-                    protoMR.send(it, RAW.value)
-                }.also { registerResource("proto message batcher", it::close) }
-            }
-
             val aliasToService = mutableMapOf<String, Holder>()
             sessions.forEach { sessionAlias, sessionSettings ->
                 val stateManager = load<IStateManager>(BasicStateManager::class.java)
@@ -138,18 +116,30 @@ class Application(
                 val onRequest: (RawHttpRequest) -> Unit
                 val onResponse: (RawHttpRequest, RawHttpResponse<*>) -> Unit
 
+                val onError: (Throwable) -> Unit = {
+                    eventBatcher.storeEvent(clientEventId, "Batching problem: ${it.message}", "Message batching problem", it)
+                }
+
                 if (useTransport) {
+                    val transportMB = MessageBatcher(
+                            maxBatchSize,
+                            maxFlushTime,
+                            book,
+                            GROUP_SELECTOR,
+                            executor,
+                            onError
+                        ) { batch ->
+                            transportMR.send(batch)
+                            if (!sessionSettings.publishSentEvents) return@MessageBatcher
+                            eventBatcher.publishSentEvents(clientEventId, batch)
+                        }.also { registerResource("transport message batcher $sessionAlias", it::close) }
+
                     onRequest = { request: RawHttpRequest ->
-                        val rawMessage = outgoingLock.withLock {
+                        outgoingLock.withLock {
                             request.toTransportMessage(sessionAlias, outgoingSequence()).also {
                                 transportMB.onMessage(it, sessionGroup)
                             }
                         }
-                        eventBatcher.storeEvent(
-                            rawMessage.eventId?.toProto() ?: rootEventId,
-                            "Sent HTTP request",
-                            "Send message"
-                        )
                     }
                     onResponse = { request: RawHttpRequest, response: RawHttpResponse<*> ->
                         incomingLock.withLock {
@@ -161,21 +151,27 @@ class Application(
                         stateManager.onResponse(response)
                     }
                 } else {
+                    val protoMB = RawMessageBatcher(
+                        maxBatchSize,
+                        maxFlushTime,
+                        RAW_GROUP_SELECTOR,
+                        executor,
+                        onError
+                    ) { batch ->
+                        protoMR.send(batch, RAW.value)
+                        if (!sessionSettings.publishSentEvents) return@RawMessageBatcher
+                        eventBatcher.publishSentEvents(clientEventId, batch)
+                    }.also { registerResource("proto message batcher $sessionAlias", it::close) }
                     val connectionId = com.exactpro.th2.common.grpc.ConnectionID.newBuilder()
                         .setSessionAlias(sessionAlias)
                         .setSessionGroup(sessionGroup)
                         .build()
 
                     onRequest = { request: RawHttpRequest ->
-                        val rawMessage = outgoingLock.withLock {
+                        outgoingLock.withLock {
                             request.toProtoMessage(connectionId, outgoingSequence())
                                 .also(protoMB::onMessage)
                         }
-                        eventBatcher.storeEvent(
-                            if (rawMessage.hasParentEventId()) rawMessage.parentEventId else rootEventId,
-                            "Sent HTTP request",
-                            "Send message"
-                        )
                     }
                     onResponse = { request: RawHttpRequest, response: RawHttpResponse<*> ->
                         incomingLock.withLock {

src/main/kotlin/com/exactpro/th2/http/client/Settings.kt

@@ -44,6 +44,7 @@ data class InternalSessionSettings(
     val validateCertificates: Boolean? = null,
     @param:JsonDeserialize(converter = CertificateConverter::class) val clientCertificate: X509Certificate? = null,
     @param:JsonDeserialize(converter = PrivateKeyConverter::class) val certificatePrivateKey: PrivateKey? = null,
+    val publishSentEvents: Boolean? = null,
 )
 
 data class InternalSettings(
@@ -64,6 +65,7 @@ data class InternalSettings(
     @Deprecated("the parameter isn't used any more", level = DeprecationLevel.ERROR) val batcherThreads: Int = 2,
     val maxBatchSize: Int = 1000,
     val maxFlushTime: Long = 1000,
+    val publishSentEvents: Boolean = true,
     val sessions: Map<String, InternalSessionSettings> = emptyMap(),
 )
 
@@ -79,6 +81,7 @@ data class SessionSettings(
     val validateCertificates: Boolean,
     val clientCertificate: X509Certificate?,
     val certificatePrivateKey: PrivateKey?,
+    val publishSentEvents: Boolean,
 ) {
     val certificate: Certificate? = clientCertificate?.run {
         val key = requireNotNull(certificatePrivateKey) {
@@ -140,6 +143,7 @@ fun getSettings(getFunc: (Class<InternalSettings>, ObjectMapper) -> InternalSett
                 validateCertificates = settings.validateCertificates,
                 clientCertificate = settings.clientCertificate,
                 certificatePrivateKey = settings.certificatePrivateKey,
+                publishSentEvents = settings.publishSentEvents,
             ))
         } else {
             settings.sessions.mapValues { (key, value) ->
@@ -157,6 +161,7 @@ fun getSettings(getFunc: (Class<InternalSettings>, ObjectMapper) -> InternalSett
                     validateCertificates = value.validateCertificates ?: settings.validateCertificates,
                     clientCertificate = value.clientCertificate ?: settings.clientCertificate,
                     certificatePrivateKey = value.certificatePrivateKey ?: settings.certificatePrivateKey,
+                    publishSentEvents = value.publishSentEvents ?: settings.publishSentEvents,
                 )
             }
         }

src/main/kotlin/com/exactpro/th2/http/client/util/EventUtil.kt

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2026 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.http.client.util
+
+import com.exactpro.th2.common.event.Event
+import com.exactpro.th2.common.grpc.Direction.SECOND
+import com.exactpro.th2.common.grpc.EventID
+import com.exactpro.th2.common.grpc.MessageGroupBatch
+import com.exactpro.th2.common.grpc.MessageID
+import com.exactpro.th2.common.message.direction
+import com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.Direction.OUTGOING
+import com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.GroupBatch
+import com.exactpro.th2.common.utils.event.EventBatcher
+import com.exactpro.th2.common.utils.event.transport.toProto
+import com.exactpro.th2.common.utils.message.transport.toProto
+
+fun EventBatcher.publishSentEvents(rootEventId: EventID, batch: MessageGroupBatch) {
+    publishSentEvent(hashMapOf<EventID, MutableList<MessageID>>().apply {
+        for (group in batch.groupsList) {
+            val message = group.messagesList[0].rawMessage
+            if (message.direction != SECOND) continue
+            val eventId = message.eventId ?: rootEventId
+            this@apply.computeIfAbsent(eventId) { mutableListOf() } += message.metadata.id
+        }
+    })
+}
+
+fun EventBatcher.publishSentEvents(rootEventId: EventID, batch: GroupBatch) {
+    publishSentEvent(hashMapOf<EventID, MutableList<MessageID>>().apply {
+        for (group in batch.groups) {
+            val message = group.messages[0]
+            if (message.id.direction != OUTGOING) continue
+            val eventId = message.eventId?.toProto() ?: rootEventId
+            this@apply.computeIfAbsent(eventId) { mutableListOf() } += message.id.toProto(batch.book, batch.sessionGroup)
+        }
+    })
+}
+
+fun EventBatcher.publishSentEvent(eventMessages: Map<EventID, MutableList<MessageID>>) {
+    for ((eventId, messageIds) in eventMessages) {
+        storeEvent(
+            eventId,
+            "Sent HTTP request",
+            "Send message",
+            messageIds = messageIds
+        )
+    }
+}
+
+fun EventBatcher.storeEvent(
+    parentId: EventID,
+    name: String,
+    type: String,
+    cause: Throwable? = null,
+    messageIds: Iterable<MessageID> = emptySet(),
+): Event = Event.start().apply {
+    endTimestamp()
+    name(name)
+    type(type)
+    messageIds.forEach(this::messageID)
+    status(if (cause != null) Event.Status.FAILED else Event.Status.PASSED)
+
+    var error = cause
+
+    while (error != null) {
+        exception(error, true)
+        error = error.cause
+    }
+
+    onEvent(toProto(parentId))
+}

src/main/kotlin/com/exactpro/th2/http/client/util/MessageUtil.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 Exactpro (Exactpro Systems Limited)
+ * Copyright 2020-2026 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.
@@ -171,6 +171,9 @@ fun TransportMessageGroup.toRequest(): RawHttpRequest = when (messages.size) {
     else -> error("Message group contains more than 2 messages")
 }
 
+val RawMessage.eventId: EventID?
+    get() = if (hasParentEventId()) parentEventId else null
+
 private sealed interface IRequest {
     val parentEventId: EventID?
     val metadata: Map<String, String>