finos#1434 making sure web socket test times out if expected message …

…is not returned in time

vuu/src/main/scala/org/finos/vuu/core/CoreServerApiHandler.scala

@@ -19,7 +19,7 @@ class CoreServerApiHandler(val viewPortContainer: ViewPortContainer,
   override def process(msg: ViewPortRpcCall)(ctx: RequestContext): Option[ViewServerMessage] = {
     Try(viewPortContainer.callRpcService(msg.vpId, msg.rpcName, msg.params, msg.namedParams, ctx.session)(ctx)) match {
       case Success(action) =>
-        logger.info("Processed VP RPC call" + msg)
+        logger.info("Processed VP RPC call " + msg)
         vsMsg(ViewPortRpcResponse(msg.vpId, msg.rpcName, action))(ctx)
       case Failure(e) =>
         logger.info("Failed to process VP RPC call", e)
@@ -418,15 +418,15 @@ class CoreServerApiHandler(val viewPortContainer: ViewPortContainer,
   private def handleViewPortRpcRequest(msg: RpcRequest, viewPortId: String, ctx: RequestContext) = {
     val response = Try(viewPortContainer.handleRpcRequest(viewPortId, msg.rpcName, msg.params)(ctx)) match {
       case Success(functionResult) =>
-        logger.info(s"Processed VP RPC call ${ctx.requestId}" + msg)
+        logger.info(s"Processed Rpc Request ${ctx.requestId} " + msg)
         functionResult match {
           case RpcFunctionSuccess(data) =>
             RpcResponseNew(rpcName = msg.rpcName, result = RpcSuccessResult(data), NoneAction())
           case RpcFunctionFailure(errorCode, error, exception) =>
             createErrorRpcResponse(msg, error)
         }
       case Failure(e) =>
-        logger.info(s"Failed to process VP RPC call ${ctx.requestId}", e)
+        logger.info(s"Failed to process Rpc Request ${ctx.requestId}", e)
         createErrorRpcResponse(msg, e.toString)
     }
     vsMsg(response)(ctx)

vuu/src/main/scala/org/finos/vuu/net/ViewServerClient.scala

@@ -26,10 +26,12 @@ class WebSocketViewServerClient(ws: WebSocketClient, serializer: Serializer[Stri
 
     }
 
-    logger.info("Websocket should be up.")
+    logger.info(s"[WSClient] Websocket on ${ws.uri} should be up.")
   }
 
-  override def doStop(): Unit = {}
+  override def doStop(): Unit = {
+    logger.info(s"[WSClient] Websocket on ${ws.uri} stopping.")
+  }
 
   override def doInitialize(): Unit = {}
 

vuu/src/test/scala/org/finos/vuu/wsapi/helpers/TestStartUp.scala

@@ -1,21 +1,20 @@
 package org.finos.vuu.wsapi.helpers
 
-import com.typesafe.scalalogging.StrictLogging
 import org.finos.toolbox.jmx.{MetricsProvider, MetricsProviderImpl}
 import org.finos.toolbox.lifecycle.LifecycleContainer
 import org.finos.toolbox.time.Clock
-import org.finos.vuu.core.{VuuClientConnectionOptions, VuuSecurityOptions, VuuServer, VuuServerConfig, VuuThreadingOptions, VuuWebSocketOptions}
 import org.finos.vuu.core.module.{TableDefContainer, ViewServerModule}
-import org.finos.vuu.net.{AlwaysHappyLoginValidator, ViewServerClient, WebSocketViewServerClient}
+import org.finos.vuu.core._
 import org.finos.vuu.net.auth.AlwaysHappyAuthenticator
 import org.finos.vuu.net.http.VuuHttp2ServerOptions
 import org.finos.vuu.net.json.JsonVsSerializer
 import org.finos.vuu.net.ws.WebSocketClient
+import org.finos.vuu.net.{AlwaysHappyLoginValidator, ViewServerClient, WebSocketViewServerClient}
 
 class TestStartUp(moduleFactoryFunc: () => ViewServerModule)(
                   implicit val timeProvider: Clock,
                   implicit val lifecycle: LifecycleContainer,
-                  implicit val tableDefContainer: TableDefContainer) extends StrictLogging {
+                  implicit val tableDefContainer: TableDefContainer){
 
 
   def startServerAndClient(): TestVuuClient = {
@@ -62,7 +61,6 @@ class TestStartUp(moduleFactoryFunc: () => ViewServerModule)(
     //lifecycle registration is done in constructor of service classes, so sequence of create is important
     lifecycle.start()
 
-    logger.info(s"[TEST CLIENT] Starting vuu client at ${client.uri}")
     vuuClient
   }
 }

vuu/src/test/scala/org/finos/vuu/wsapi/helpers/TestVuuClient.scala

@@ -4,10 +4,12 @@ import com.typesafe.scalalogging.StrictLogging
 import org.finos.vuu.client.messages.{RequestId, TokenId}
 import org.finos.vuu.net._
 import org.scalatest.concurrent.TimeLimits.failAfter
+import org.scalatest.concurrent.{Signaler, ThreadSignaler}
 import org.scalatest.time.Span
 import org.scalatest.time.SpanSugar._
 
 import java.util.concurrent.ConcurrentHashMap
+import scala.annotation.tailrec
 import scala.language.postfixOps
 import scala.reflect.ClassTag
 
@@ -29,28 +31,46 @@ class TestVuuClient(vsClient: ViewServerClient) extends StrictLogging {
   def awaitForMsgWithBody[T <: AnyRef](implicit t: ClassTag[T]): Option[T] =
     awaitForMsg.map(msg => msg.body.asInstanceOf[T])
 
+
+  implicit val signaler: Signaler = ThreadSignaler
+
   def awaitForMsg[T <: AnyRef](implicit t: ClassTag[T]): Option[ViewServerMessage] = {
-    failAfter(timeout){
-      val msg = vsClient.awaitMsg
-      if (msg != null) { //null indicate error or timeout
-        if (isExpectedBodyType(t, msg))
-          Some(msg)
-        else
-          awaitForMsg
+    failAfter(timeout) {
+      getNextMessageUntilBodyIsExpectedType()
+    }
+  }
+
+  @tailrec
+  private def getNextMessageUntilBodyIsExpectedType[T <: AnyRef]()(implicit t: ClassTag[T]): Option[ViewServerMessage] = {
+    val msg = vsClient.awaitMsg
+    if (msg != null) { //null indicate error or timeout
+      if (isExpectedBodyType(t, msg)) {
+        Some(msg)
+      } else {
+        logger.info(s"Received ${msg.body.getClass} but was expecting ${t.runtimeClass}. Dismissing message and waiting for next one.")
+        getNextMessageUntilBodyIsExpectedType()
       }
-      else
-        None
+    }
+    else {
+      logger.info(s"Did not receive any message in response. Try waiting again.")
+      getNextMessageUntilBodyIsExpectedType()
     }
   }
 
   val responsesMap: ConcurrentHashMap[String, ViewServerMessage] = new ConcurrentHashMap
 
   def awaitForResponse(requestId: String): Option[ViewServerMessage] = {
+    failAfter(timeout) {
+      getNextMessageUntilResponseForRequestId(requestId)
+    }
+  }
 
+  @tailrec
+  private def getNextMessageUntilResponseForRequestId(requestId: String): Option[ViewServerMessage] = {
     lookupFromReceivedResponses(requestId)
       .map(msg => {
         logger.info(s"Found response for $requestId in cache")
-        return Some(msg)
+        Some(msg)
       })
 
     val msg = vsClient.awaitMsg
@@ -61,11 +81,11 @@ class TestVuuClient(vsClient: ViewServerClient) extends StrictLogging {
       } else {
         responsesMap.put(msg.requestId, msg)
         logger.info(s"Added response for $requestId in cache")
-        awaitForResponse(requestId)
+        getNextMessageUntilResponseForRequestId(requestId)
       }
     else {
       logger.error(s"Failed or timed out while waiting for response for $requestId")
-      None
+      getNextMessageUntilResponseForRequestId(requestId)
     }
   }