Bi-directional GRPC non interactive implementation

Issue-ID: CCSDK-1747
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: I236bf6d4edaf983ca4162b5ce064736ac690b504

2 files changed, 33 insertions, 8 deletions

diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt
index c9ff23573..29d24c6ad 100644
--- a/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/services/execution/StreamingRemoteExecutionServiceTest.kt
@@ -36,6 +36,8 @@ import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType
 import org.onap.ccsdk.cds.controllerblueprints.core.logger
 import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput
 import java.util.*
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
 
 
 class StreamingRemoteExecutionServiceTest {
@@ -69,25 +71,48 @@ class StreamingRemoteExecutionServiceTest {
 
             val spyStreamingRemoteExecutionService = spyk(streamingRemoteExecutionService)
             /** To test with real server, uncomment below line */
-            coEvery { spyStreamingRemoteExecutionService.createGrpcChannel(any()) } returns channel
+            coEvery() { spyStreamingRemoteExecutionService.createGrpcChannel(any()) } returns channel
+
+            /** Test Send and Receive non interactive transaction */
+            val nonInteractiveDeferred = arrayListOf<Deferred<*>>()
+            repeat(2) { count ->
+                val requestId = "1234-$count"
+                val request = getRequest(requestId)
+                val invocationId = request.commonHeader.subRequestId
+                val deferred = async {
+                    val response = spyStreamingRemoteExecutionService.sendNonInteractive(tokenAuthGrpcClientProperties,
+                            invocationId, request, 1000L)
+                    assertNotNull(response, "failed to get non interactive response")
+                    assertEquals(response.commonHeader.requestId, requestId,
+                            "failed to match non interactive response id")
+                    assertEquals(response.status.eventType, EventType.EVENT_COMPONENT_EXECUTED,
+                            "failed to match non interactive response type")
+                }
+                nonInteractiveDeferred.add(deferred)
 
-            val responseFlowsDeferred = arrayListOf<Deferred<*>>()
+            }
+            nonInteractiveDeferred.awaitAll()
 
-            repeat(1) { count ->
+            /** Test Send and Receive interactive transaction */
+            val responseFlowsDeferred = arrayListOf<Deferred<*>>()
+            repeat(2) { count ->
                 val requestId = "12345-$count"
-                val responseFlow = spyStreamingRemoteExecutionService.openSubscription(tokenAuthGrpcClientProperties, requestId)
+                val request = getRequest(requestId)
+                val invocationId = request.commonHeader.requestId
+                val responseFlow = spyStreamingRemoteExecutionService
+                        .openSubscription(tokenAuthGrpcClientProperties, invocationId)
 
                 val deferred = async {
                     responseFlow.collect {
-                        log.info("Received $count-response (${it.commonHeader.subRequestId}) : ${it.status.eventType}")
+                        log.info("Received $count-response ($invocationId) : ${it.status.eventType}")
                         if (it.status.eventType == EventType.EVENT_COMPONENT_EXECUTED) {
-                            spyStreamingRemoteExecutionService.cancelSubscription(it.commonHeader.requestId)
+                            spyStreamingRemoteExecutionService.cancelSubscription(invocationId)
                         }
                     }
                 }
                 responseFlowsDeferred.add(deferred)
                 /** Sending Multiple messages with same requestId  and different subRequestId */
-                spyStreamingRemoteExecutionService.send(getRequest(requestId))
+                spyStreamingRemoteExecutionService.send(invocationId, request)
             }
             responseFlowsDeferred.awaitAll()
             streamingRemoteExecutionService.closeChannel(tokenAuthGrpcClientProperties)
diff --git a/ms/blueprintsprocessor/modules/services/execution-service/src/test/resources/logback-test.xml b/ms/blueprintsprocessor/modules/services/execution-service/src/test/resources/logback-test.xml
index 703a52642..afe10b39d 100644
--- a/ms/blueprintsprocessor/modules/services/execution-service/src/test/resources/logback-test.xml
+++ b/ms/blueprintsprocessor/modules/services/execution-service/src/test/resources/logback-test.xml
@@ -19,7 +19,7 @@
         <!-- encoders are assigned the type
              ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
         <encoder>
-            <pattern>%d{HH:mm:ss.SSS} %-5level %logger{100} - %msg%n</pattern>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n</pattern>
         </encoder>
     </appender>