#355 better exception model

7mind · Aug 21, 2018 · c21eb51 · c21eb51
1 parent 67f6bb9
commit c21eb51
Show file tree

Hide file tree

Showing 6 changed files with 67 additions and 28 deletions.
diff --git a/...rc/main/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/RequestState.scala b/...rc/main/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/RequestState.scala
@@ -58,7 +58,7 @@ class RequestState[Or[+ _, + _] : BIO] {
 
       method <- maybeMethod match {
         case Some(m@PacketInfo(method, id)) =>
-          respond(id, RawResponse(data, method))
+          respond(id, RawResponse.GoodRawResponse(data, method))
           R.now(m)
 
         case None =>

diff --git a/...ain/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithHttp4sServer.scala b/...ain/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithHttp4sServer.scala
@@ -71,10 +71,14 @@ trait WithHttp4sServer {
                         id =>
                           sess.requestState.poll(id, pollingInterval, timeout)
                             .flatMap {
-                              case Some(value) =>
+                              case Some(value: RawResponse.GoodRawResponse) =>
                                 logger.debug(s"${request.method -> "method"}, $id: Have response: $value")
                                 codec.decode(value.data, value.method)
 
+                              case Some(value : RawResponse.BadRawResponse) =>
+                                logger.debug(s"${request.method -> "method"}, $id: Generic failure response: $value")
+                                BIO.terminate(new IRTGenericFailure(s"${request.method -> "method"}, $id: generic failure: $value"))
+
                               case None =>
                                 BIO.terminate(new TimeoutException(s"${request.method -> "method"}, $id: No response in $timeout"))
                             }
@@ -206,8 +210,12 @@ trait WithHttp4sServer {
         case RpcPacket(RPCPacketKind.BuzzResponse, data, _, id, _, _, _) =>
           context.requestState.handleResponse(id, data).flatMap(_ => BIO.point(None))
 
+        case RpcPacket(RPCPacketKind.BuzzFailure, data, _, Some(id), _, _, _) =>
+          context.requestState.respond(id, RawResponse.BadRawResponse())
+          BIO.fail(new IRTGenericFailure(s"Buzzer has returned failure: $data"))
+
         case k =>
-          BIO.fail(new UnsupportedOperationException(s"Can't handle $k"))
+          BIO.fail(new IRTMissingHandlerException(s"Can't handle $k", k))
       }
     }
 

diff --git a/...n/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithHttp4sWsClient.scala b/...n/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithHttp4sWsClient.scala
@@ -4,6 +4,7 @@ import java.net.URI
 import java.util.concurrent.TimeoutException
 
 import com.github.pshirshov.izumi.idealingua.runtime.bio.BIO._
+import com.github.pshirshov.izumi.idealingua.runtime.rpc
 import com.github.pshirshov.izumi.idealingua.runtime.rpc._
 import io.circe.parser.parse
 import io.circe.syntax._
@@ -85,15 +86,36 @@ trait WithHttp4sWsClient {
         case p@RpcPacket(RPCPacketKind.BuzzRequest, data, Some(id), _, Some(service), Some(method), _) =>
           val methodId = IRTMethodId(IRTServiceId(service), IRTMethodName(method))
           val packetInfo = PacketInfo(methodId, id)
-          for {
+
+          val responsePkt = for {
             maybeResponse <- buzzerMuxer.doInvoke(data, wsClientContextProvider.toContext(p), methodId)
-            maybePacket <- BIO.point(maybeResponse.map(r => RpcPacket.buzzerResponse(id, r)))
+            maybePacket <- BIO.now(maybeResponse.map(r => RpcPacket.buzzerResponse(id, r)))
+          } yield {
+            maybePacket
+          }
+
+          for {
+            maybePacket <- responsePkt.sandboxWith {
+              _.redeem(
+                {
+                  case Left(exception :: otherIssues) =>
+                    logger.error(s"${packetInfo -> null}: WS processing terminated, $exception, $otherIssues")
+                    BIO.point(Some(rpc.RpcPacket.buzzerFail(Some(id), exception.getMessage)))
+                  case Left(Nil) =>
+                    BIO.terminate(new IllegalStateException())
+                  case Right(exception) =>
+                    logger.error(s"${packetInfo -> null}: WS processing failed, $exception")
+                    BIO.point(Some(rpc.RpcPacket.buzzerFail(Some(id), exception.getMessage)))
+                }, {
+                  succ => BIO.point(succ)
+                }
+              )
+            }
             maybeEncoded <- BIO.syncThrowable(maybePacket.map(r => printer.pretty(r.asJson)))
             _ <- BIO.point {
               maybeEncoded match {
                 case Some(response) =>
                   logger.debug(s"${method -> "method"}, ${id -> "id"}: Prepared buzzer $response")
-
                   wsClient.send(response)
                 case None =>
               }
@@ -102,18 +124,13 @@ trait WithHttp4sWsClient {
             packetInfo
           }
 
-        case RpcPacket(RPCPacketKind.RpcFail, data, _, ref, _, _, _) =>
-          ref match {
-            case Some(value) =>
-              if (requestState.methodOf(value).nonEmpty) {
-                requestState.forget(value) // TODO: we should support exception passing in RequestState
-              }
-            case None =>
-          }
-
-          ref.foreach(requestState.forget)
+        case RpcPacket(RPCPacketKind.RpcFail, data, _, Some(ref), _, _, _) =>
+          requestState.respond(ref, RawResponse.BadRawResponse())
           BIO.fail(new IRTGenericFailure(s"RPC failure for $ref: $data"))
 
+        case RpcPacket(RPCPacketKind.RpcFail, data, _, None, _, _, _) =>
+          BIO.fail(new IRTGenericFailure(s"Missing ref in RPC failure: $data"))
+
         case RpcPacket(RPCPacketKind.Fail, data, _, _, _, _, _) =>
           BIO.fail(new IRTGenericFailure(s"Critical RPC failure: $data"))
 
@@ -165,10 +182,14 @@ trait WithHttp4sWsClient {
                     id =>
                       requestState.poll(id, pollingInterval, timeout)
                         .flatMap {
-                          case Some(value) =>
+                          case Some(value : RawResponse.GoodRawResponse) =>
                             logger.debug(s"${request.method -> "method"}, $id: Have response: $value")
                             codec.decode(value.data, value.method)
 
+                          case Some(value : RawResponse.BadRawResponse) =>
+                            logger.debug(s"${request.method -> "method"}, $id: Have response: $value")
+                            BIO.terminate(new IRTGenericFailure(s"${request.method -> "method"}, $id: generic failure: $value"))
+
                           case None =>
                             BIO.terminate(new TimeoutException(s"${request.method -> "method"}, $id: No response in $timeout"))
                         }

diff --git a/...com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithWebsocketClientContext.scala b/...com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/WithWebsocketClientContext.scala
@@ -133,4 +133,11 @@ trait WithWebsocketClientContext {
 
 }
 
-case class RawResponse(data: Json, method: IRTMethodId)
+sealed trait RawResponse
+
+object RawResponse {
+
+  case class GoodRawResponse(data: Json, method: IRTMethodId) extends RawResponse
+  case class BadRawResponse() extends RawResponse // This needs to be extended: https://github.com/pshirshov/izumi-r2/issues/355
+
+}
diff --git a/.../scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/Http4sTransportTest.scala b/.../scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/http4s/Http4sTransportTest.scala
@@ -20,7 +20,6 @@ import org.http4s.server.blaze._
 import org.scalatest.WordSpec
 import scalaz.zio
 
-import scala.concurrent.TimeoutException
 import scala.language.higherKinds
 
 class Http4sTransportTest extends WordSpec {
@@ -67,7 +66,7 @@ class Http4sTransportTest extends WordSpec {
     }
 
     disp.setupCredentials("user", "badpass")
-    intercept[TimeoutException] {
+    intercept[IRTGenericFailure] {
       BIOR.unsafeRun(greeterClient.alternative())
     }
     disp.close()

diff --git a/...-rpc-scala/src/main/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/packets.scala b/...-rpc-scala/src/main/scala/com/github/pshirshov/izumi/idealingua/runtime/rpc/packets.scala
@@ -102,12 +102,8 @@ object RpcPacket {
 
   implicit def enc0: Encoder[RpcPacket] = deriveEncoder
 
-  def rpcFail(ref: Option[RpcPacketId], cause: String): RpcPacket = {
-    RpcPacket(RPCPacketKind.RpcFail, Map("cause" -> cause).asJson, None, ref, None, None, Map.empty)
-  }
-
-  def rpcResponse(ref: RpcPacketId, data: Json): RpcPacket = {
-    RpcPacket(RPCPacketKind.RpcResponse, data, None, Some(ref), None, None, Map.empty)
+  def rpcCritical(data: String, cause: String): RpcPacket = {
+    RpcPacket(RPCPacketKind.Fail, Map("data" -> data, "cause" -> cause).asJson, None, None, None, None, Map.empty)
   }
 
   def rpcRequest(method: IRTMethodId, data: Json): RpcPacket = {
@@ -122,6 +118,14 @@ object RpcPacket {
     )
   }
 
+  def rpcResponse(ref: RpcPacketId, data: Json): RpcPacket = {
+    RpcPacket(RPCPacketKind.RpcResponse, data, None, Some(ref), None, None, Map.empty)
+  }
+
+  def rpcFail(ref: Option[RpcPacketId], cause: String): RpcPacket = {
+    RpcPacket(RPCPacketKind.RpcFail, Map("cause" -> cause).asJson, None, ref, None, None, Map.empty)
+  }
+
   def buzzerRequest(method: IRTMethodId, data: Json): RpcPacket = {
     RpcPacket(
       RPCPacketKind.BuzzRequest,
@@ -139,7 +143,7 @@ object RpcPacket {
 
   }
 
-  def rpcCritical(data: String, cause: String): RpcPacket = {
-    RpcPacket(RPCPacketKind.Fail, Map("data" -> data, "cause" -> cause).asJson, None, None, None, None, Map.empty)
+  def buzzerFail(ref: Option[RpcPacketId], cause: String): RpcPacket = {
+    RpcPacket(RPCPacketKind.BuzzFailure, Map("cause" -> cause).asJson, None, ref, None, None, Map.empty)
   }
 }