[TH2-5073] Ability to send RawMessage directly to conn components. (#121

)
th2-net · Sep 19, 2023 · b44e6eb · b44e6eb
1 parent 5b0275e
commit b44e6eb
Show file tree

Hide file tree

Showing 4 changed files with 152 additions and 4 deletions.
diff --git a/README.md b/README.md
@@ -1,4 +1,4 @@
-# th2 act template (5.0.0)
+# th2 act template (5.1.0)
 
 ## Overview
 
@@ -17,9 +17,14 @@ Most of them consists of the next steps:
 3. Sends the passed business message to Codec via mq pin
 4. Waits the specific business message from Codec during specified timeout
 5. Returns responded business message with checkpoint
-
 ![picture](scheme.png)
 
+### Sending raw messages
+It is also possible to send raw messages ( containing metadata and having payload in original format ) directly to conn components without waiting while message will be encoded by codec and then sent to conn component.
+For that you need to:
+1. Send rpc request using `SendRawMessageRequest` and `sendRawMessage` rpc that is defined in [th2-grpc-act-template](https://github.com/th2-net/th2-grpc-act-template/blob/master/src/main/proto/th2_grpc_act_template/act_template.proto "act_template.proto")
+2. Transport group with `RawMessage` will be published to pins with `send_raw` attribute.
+
 ## Custom resources for infra-mgr
 
 ```yaml
@@ -60,6 +65,28 @@ spec:
             - field-name: session_alias
               expected-value: conn2_session_alias
               operation: EQUAL
+    - name: to_conn1_transport
+      connection-type: mq
+      attributes:
+        - publish
+        - transport-group
+        - send_raw
+      filters:
+        - metadata:
+            - field-name: session_alias
+              expected-value: conn1_session_alias
+              operation: EQUAL
+    - name: to_conn2_transport
+      connection-type: mq
+      attributes:
+        - publish
+        - transport-group
+        - send_raw
+      filters:
+        - metadata:
+            - field-name: session_alias
+              expected-value: conn1_session_alias
+              operation: EQUAL
 ```
 
 ## Descriptor gradle plugin
@@ -71,6 +98,10 @@ the `protobuf-description-base64` label. Such descriptors can be used to interac
 
 ## Release Notes
 
+### 5.1.0
++ Added RPC to send raw messages.
++ Updated grpc-act-template to `4.2.0`
+
 ### 5.0.0
 
 + Migrated to th2 transport protocol

diff --git a/build.gradle b/build.gradle
@@ -94,7 +94,7 @@ dependencies {
 
     implementation 'com.exactpro.th2:common:5.4.0-dev'
     implementation 'com.exactpro.th2:common-utils:2.2.0-dev'
-    implementation 'com.exactpro.th2:grpc-act-template:4.1.0'
+    implementation 'com.exactpro.th2:grpc-act-template:4.2.0'
     implementation 'com.exactpro.th2:grpc-check1:4.2.0-dev'
 
     implementation "com.fasterxml.jackson.core:jackson-core"

diff --git a/gradle.properties b/gradle.properties
@@ -1,4 +1,4 @@
-release_version=5.0.0
+release_version=5.1.0
 nexus_url=
 nexus_user=
 nexus_password=

diff --git a/src/main/java/com/exactpro/th2/act/ActHandler.java b/src/main/java/com/exactpro/th2/act/ActHandler.java
@@ -20,13 +20,16 @@
 import com.exactpro.th2.act.grpc.PlaceMessageRequestOrBuilder;
 import com.exactpro.th2.act.grpc.PlaceMessageResponse;
 import com.exactpro.th2.act.grpc.SendMessageResponse;
+import com.exactpro.th2.act.grpc.SendRawMessageRequest;
+import com.exactpro.th2.act.grpc.SendRawMessageResponse;
 import com.exactpro.th2.act.impl.MessageResponseMonitor;
 import com.exactpro.th2.act.rules.FieldCheckRule;
 import com.exactpro.th2.check1.grpc.Check1Service;
 import com.exactpro.th2.check1.grpc.CheckpointRequest;
 import com.exactpro.th2.check1.grpc.CheckpointResponse;
 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.event.IBodyData;
 import com.exactpro.th2.common.event.bean.TreeTable;
 import com.exactpro.th2.common.event.bean.builder.CollectionBuilder;
@@ -42,6 +45,8 @@
 import com.exactpro.th2.common.grpc.MessageID;
 import com.exactpro.th2.common.grpc.MessageMetadata;
 import com.exactpro.th2.common.grpc.MessageOrBuilder;
+import com.exactpro.th2.common.grpc.RawMessageMetadata;
+import com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.RawMessage;
 import com.exactpro.th2.common.grpc.RequestStatus;
 import com.exactpro.th2.common.grpc.Value;
 import com.exactpro.th2.common.schema.message.MessageRouter;
@@ -57,6 +62,7 @@
 import io.grpc.Context;
 import io.grpc.Deadline;
 import io.grpc.stub.StreamObserver;
+import io.netty.buffer.ByteBufUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -78,6 +84,7 @@
 import static com.exactpro.th2.common.grpc.RequestStatus.Status.SUCCESS;
 import static com.exactpro.th2.common.schema.message.impl.rabbitmq.transport.Direction.INCOMING;
 import static com.exactpro.th2.common.utils.message.MessageHolderUtilsKt.getTreeTable;
+import static com.exactpro.th2.common.utils.message.MessageUtilsKt.toTransport;
 import static com.exactpro.th2.common.utils.message.transport.MessageUtilsKt.toBatch;
 import static com.exactpro.th2.common.utils.message.transport.MessageUtilsKt.toGroup;
 import static com.exactpro.th2.common.utils.message.transport.MessageUtilsKt.toTreeTable;
@@ -90,6 +97,7 @@
 
 public class ActHandler extends ActImplBase {
     private static final int DEFAULT_RESPONSE_TIMEOUT = 10_000;
+    private static final String SEND_RAW_QUEUE_ATTRIBUTE = "send_raw";
     private static final Logger LOGGER = LoggerFactory.getLogger(ActHandler.class);
 
     private final Check1Service check1Service;
@@ -169,6 +177,48 @@ public void placeOrderCancelReplaceRequest(PlaceMessageRequest request, StreamOb
         }
     }
 
+    @Override
+    public void sendRawMessage(SendRawMessageRequest request, StreamObserver<SendRawMessageResponse> responseObserver) {
+        long startPlaceMessage = System.currentTimeMillis();
+        try {
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("Sending  message request: " + shortDebugString(request));
+            }
+
+            String actName = "sendRawMessage";
+            // FIXME store parent with fail in case of children fail
+            EventID parentId = createAndStoreParentEvent(request, actName, PASSED);
+
+            Checkpoint checkpoint = registerCheckPoint(parentId);
+
+            if (Context.current().isCancelled()) {
+                LOGGER.warn("'{}' request cancelled by client", actName);
+                sendRawMessageErrorResponse(responseObserver, "Request has been cancelled by client");
+                return;
+            }
+
+            try {
+                sendRawMessage(request.getRaw(), parentId);
+            } catch (Exception ex) {
+                createAndStoreErrorEvent("sendMessage", ex.getMessage(), Instant.now(), parentId);
+                throw ex;
+            }
+
+            SendRawMessageResponse response = SendRawMessageResponse.newBuilder()
+                .setStatus(RequestStatus.newBuilder().setStatus(SUCCESS).build())
+                .setCheckpointId(checkpoint)
+                .build();
+            responseObserver.onNext(response);
+            responseObserver.onCompleted();
+
+        } catch (Exception e) {
+            LOGGER.error("Failed to send a message. Message = {}", request.getRaw(), e);
+            sendRawMessageErrorResponse(responseObserver, "Send message failed. Error: " + e.getMessage());
+        } finally {
+            LOGGER.debug("Sending the message has been finished in {}", System.currentTimeMillis() - startPlaceMessage);
+        }
+    }
+
     @Override
     public void sendMessage(PlaceMessageRequest request, StreamObserver<SendMessageResponse> responseObserver) {
         long startPlaceMessage = System.currentTimeMillis();
@@ -390,6 +440,26 @@ private EventID createAndStoreParentEvent(PlaceMessageRequestOrBuilder request,
         }
     }
 
+    private EventID createAndStoreParentEvent(SendRawMessageRequest request, String actName, Status status) throws IOException {
+        long startTime = System.currentTimeMillis();
+
+        Event event = start()
+            .name(actName + ' ' + request.getRaw().getMetadata().getId().getConnectionId().getSessionAlias())
+            .type(actName)
+            .status(status)
+            .endTimestamp(); // FIXME set properly as is in the last child
+
+        com.exactpro.th2.common.grpc.Event protoEvent = event.toProto(request.getParentEventId());
+        //FIXME process response
+        try {
+            eventBatchMessageRouter.send(EventBatch.newBuilder().addEvents(event.toProto(request.getParentEventId())).build(), "publish", "event");
+            LOGGER.debug("createAndStoreParentEvent for {} in {} ms", actName, System.currentTimeMillis() - startTime);
+            return protoEvent.getId();
+        } catch (IOException e) {
+            throw new RuntimeException("Can not send event = " + protoEvent.getId().getId(), e);
+        }
+    }
+
     private void createAndStoreNoResponseEvent(
             String actName, NoResponseBodySupplier noResponseBodySupplier,
             Instant start,
@@ -507,6 +577,29 @@ private void sendMessage(Message message, EventID parentEventId) throws IOExcept
         }
     }
 
+    private void sendRawMessage(com.exactpro.th2.common.grpc.RawMessage rawMessage, EventID parentEventId) throws IOException {
+        try {
+            LOGGER.debug("Sending the message started");
+            RawMessageMetadata metadata = rawMessage.getMetadata();
+            RawMessage transportRawMessage = RawMessage.builder()
+                .setId(toTransport(metadata.getId()))
+                .setProtocol(metadata.getProtocol())
+                .setMetadata(metadata.getPropertiesMap())
+                .setEventId(EventUtilsKt.toTransport(parentEventId))
+                .setBody(rawMessage.getBody().toByteArray())
+                .build();
+            MessageID messageID = metadata.getId();
+            messageRouter.send(toBatch(toGroup(transportRawMessage), messageID.getBookName(), messageID.getConnectionId().getSessionGroup()), SEND_RAW_QUEUE_ATTRIBUTE);
+
+            EventBatch eventBatch = EventBatch.newBuilder()
+                .addEvents(createSendRawMessageEvent(transportRawMessage, parentEventId))
+                .build();
+            eventBatchMessageRouter.send(eventBatch, "publish", "event");
+        } finally {
+            LOGGER.debug("Sending the message ended");
+        }
+    }
+
     private com.exactpro.th2.common.grpc.Event createSendMessageEvent(ParsedMessage message, EventID parentEventId) throws IOException {
         Event event = start()
                 .name("Send '" + message.getType() + "' message to connectivity");
@@ -517,6 +610,16 @@ private com.exactpro.th2.common.grpc.Event createSendMessageEvent(ParsedMessage
         return event.toProto(parentEventId);
     }
 
+    private com.exactpro.th2.common.grpc.Event createSendRawMessageEvent(RawMessage message, EventID parentEventId) throws IOException {
+        Event event = start()
+            .name("Sent raw message to connectivity");
+        com.exactpro.th2.common.event.bean.Message messageBean = EventUtils.createMessageBean(ByteBufUtil.hexDump(message.getBody()));
+        event.status(Status.PASSED);
+        event.bodyData(messageBean);
+        event.type("Outgoing message");
+        return event.toProto(parentEventId);
+    }
+
     private void createAndStoreErrorEvent(String actName, String message, Instant start, EventID parentEventId) throws IOException {
         Event errorEvent = Event.from(start)
                 .endTimestamp()
@@ -593,6 +696,20 @@ private void sendMessageErrorResponse(
         LOGGER.debug("Error response : {}", message);
     }
 
+    private void sendRawMessageErrorResponse(
+        StreamObserver<SendRawMessageResponse> responseObserver,
+        String message
+    ) {
+        responseObserver.onNext(SendRawMessageResponse.newBuilder()
+            .setStatus(RequestStatus.newBuilder()
+                .setStatus(ERROR)
+                .setMessage(message)
+                .build())
+            .build());
+        responseObserver.onCompleted();
+        LOGGER.debug("Error response : {}", message);
+    }
+
     private Checkpoint registerCheckPoint(EventID parentEventId) {
         LOGGER.debug("Registering the checkpoint started");
         CheckpointResponse response = check1Service.createCheckpoint(CheckpointRequest.newBuilder()