Merge pull request #3 from neandertech/cancelation

Add logic to propagate CE-cancelation via channels

Author: Baccata

core/src/jsonrpclib/internals/CallId.scala renamed to core/src/jsonrpclib/CallId.scala

@@ -1,4 +1,4 @@
-package jsonrpclib.internals
+package jsonrpclib
 
 import com.github.plokhotnyuk.jsoniter_scala.core._
 import scala.util.Try

core/src/jsonrpclib/RPCCoreBis.scala

@@ -11,7 +11,7 @@ import scala.util.Try
 abstract class FutureBasedChannel(endpoints: List[Endpoint[Future]])(implicit ec: ExecutionContext)
     extends MessageDispatcher[Future] {
 
-  override def createPromise[A](): Future[(Try[A] => Future[Unit], () => Future[A])] = Future.successful {
+  override def createPromise[A](callId: CallId): Future[(Try[A] => Future[Unit], () => Future[A])] = Future.successful {
     val promise = Promise[A]()
     val fulfill: Try[A] => Future[Unit] = (a: Try[A]) => Future.successful(promise.complete(a))
     val future: () => Future[A] = () => promise.future

core/src/jsonrpclib/internals/FutureBaseChannel.scala

@@ -12,12 +12,12 @@ private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F
 
   import F._
 
-  protected def background[A](fa: F[A]): F[Unit]
+  protected def background[A](maybeCallId: Option[CallId], fa: F[A]): F[Unit]
   protected def reportError(params: Option[Payload], error: ProtocolError, method: String): F[Unit]
   protected def getEndpoint(method: String): F[Option[Endpoint[F]]]
   protected def sendMessage(message: Message): F[Unit]
   protected def nextCallId(): F[CallId]
-  protected def createPromise[A](): F[(Try[A] => F[Unit], () => F[A])]
+  protected def createPromise[A](callId: CallId): F[(Try[A] => F[Unit], () => F[A])]
   protected def storePendingCall(callId: CallId, handle: OutputMessage => F[Unit]): F[Unit]
   protected def removePendingCall(callId: CallId): F[Option[OutputMessage => F[Unit]]]
 
@@ -34,7 +34,7 @@ private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F
       val encoded = inCodec.encode(input)
       doFlatMap(nextCallId()) { callId =>
         val message = InputMessage.RequestMessage(method, callId, Some(encoded))
-        doFlatMap(createPromise[Either[Err, Out]]()) { case (fulfill, future) =>
+        doFlatMap(createPromise[Either[Err, Out]](callId)) { case (fulfill, future) =>
           val pc = createPendingCall(errCodec, outCodec, fulfill)
           doFlatMap(storePendingCall(callId, pc))(_ => doFlatMap(sendMessage(message))(_ => future()))
         }
@@ -45,7 +45,7 @@ private[jsonrpclib] abstract class MessageDispatcher[F[_]](implicit F: Monadic[F
     Codec.decode[Message](Some(payload)).map {
       case im: InputMessage =>
         doFlatMap(getEndpoint(im.method)) {
-          case Some(ep) => background(executeInputMessage(im, ep))
+          case Some(ep) => background(im.maybeCallId, executeInputMessage(im, ep))
           case None =>
             im.maybeCallId match {
               case None =>

core/src/jsonrpclib/internals/MessageDispatcher.scala

@@ -0,0 +1,43 @@
+package jsonrpclib.fs2
+
+import jsonrpclib.Codec
+import jsonrpclib.CallId
+
+/** A cancelation template that represents the RPC method by which cancelation
+  *
+  * @param template
+  *   : the notification template for the cancelation method
+  * @param fromCallId
+  *   : a function to create a cancelation request out of a call id
+  * @param toCallId
+  *   : a function to extract a call id from a cancelation request
+  */
+trait CancelTemplate {
+  type C
+  def method: String
+  def codec: Codec[C]
+  def fromCallId(callId: CallId): C
+  def toCallId(cancelRequest: C): CallId
+
+}
+
+object CancelTemplate {
+
+  def make[CancelRequest: Codec](
+      cancelMethod: String,
+      toId: CancelRequest => CallId,
+      fromId: CallId => CancelRequest
+  ): CancelTemplate =
+    new CancelTemplate {
+      type C = CancelRequest
+
+      def method: String = cancelMethod
+
+      def codec: Codec[CancelRequest] = implicitly[Codec[CancelRequest]]
+
+      def fromCallId(callId: CallId): CancelRequest = fromId(callId)
+
+      def toCallId(cancelRequest: CancelRequest): CallId = toId(cancelRequest)
+    }
+
+}