Progressing on the fs2-specific implementation

Author: Baccata

build.sc

@@ -22,13 +22,12 @@ object versions {
 }
 import versions._
 
-object lib extends Cross[LibModule](versions.crossMap.keys.toSeq: _*)
-class LibModule(versionKey: String) extends define.Module {
+object core extends Cross[CoreModule](versions.crossMap.keys.toSeq: _*)
+class CoreModule(versionKey: String) extends define.Module {
 
  val crossVersion = crossMap(versionKey)
  def crossPlatformIvyDeps: T[Agg[Dep]] = Agg(
- ivy"com.github.plokhotnyuk.jsoniter-scala::jsoniter-scala-macros::2.13.31",
- ivy"io.circe::circe-core::0.14.2"
+ ivy"com.github.plokhotnyuk.jsoniter-scala::jsoniter-scala-macros::2.13.31"
  )
 
  object jvm extends RpcCrossScalaModule {
@@ -63,19 +62,19 @@ class FS2Module(versionKey: String) extends define.Module {
  val crossVersion = crossMap(versionKey)
  def crossPlatformIvyDeps: T[Agg[Dep]] = Agg(
  ivy"com.github.plokhotnyuk.jsoniter-scala::jsoniter-scala-macros::2.13.31",
- ivy"io.circe::circe-core::0.14.2"
+ ivy"co.fs2::fs2-core:3.2.8"
  )
 
  object jvm extends RpcCrossScalaModule {
- def moduleDeps = Seq(lib(versionKey).jvm)
+ def moduleDeps = Seq(core(versionKey).jvm)
  def crossScalaVersion = crossVersion
  def targetPlatform: Platform = Platform.JVM
  def ivyDeps = T(super.ivyDeps() ++ crossPlatformIvyDeps())
  object test extends Tests
  }
 
  object js extends ScalaJSModule with RpcCrossScalaModule {
- def moduleDeps = Seq(lib(versionKey).js)
+ def moduleDeps = Seq(core(versionKey).js)
  def crossScalaVersion = crossVersion
  def targetPlatform: Platform = Platform.JS
  def scalaJSVersion = versions.scalaJSVersion

core/src/jsonrpclib/Channel.scala

@@ -0,0 +1,32 @@
+package jsonrpclib
+
+import jsonrpclib.EndpointTemplate.NotificationTemplate
+import jsonrpclib.EndpointTemplate.RequestResponseTemplate
+
+trait Channel[F[_]] {
+ def mountEndpoint(endpoint: Endpoint[F]): F[Unit]
+ def unmountEndpoint(method: String): F[Unit]
+
+ def notificationStub[In](method: String)(implicit inCodec: Codec[In]): In => F[Unit]
+ def requestResponseStub[In, Err, Out](
+ method: String
+ )(implicit inCodec: Codec[In], errCodec: ErrorCodec[Err], outCodec: Codec[Out]): In => F[Either[Err, Out]]
+
+ def clientStub[In, Err, Out](template: EndpointTemplate[In, Err, Out]): In => F[Either[Err, Out]]
+}
+
+object Channel {
+
+ protected[jsonrpclib] abstract class MonadicChannel[F[_]](implicit F: Monadic[F]) extends Channel[F] {
+
+ final def clientStub[In, Err, Out](template: EndpointTemplate[In, Err, Out]): In => F[Either[Err, Out]] =
+ template match {
+ case NotificationTemplate(method, inCodec) =>
+ val stub = notificationStub(method)(inCodec)
+ (in: In) => F.doFlatMap(stub(in))(unit => F.doPure(Right(unit)))
+ case RequestResponseTemplate(method, inCodec, errCodec, outCodec) =>
+ requestResponseStub(method)(inCodec, errCodec, outCodec)
+ }
+ }
+
+}

lib/src/Codec.scala renamed to core/src/jsonrpclib/Codec.scala

@@ -0,0 +1,5 @@
+package jsonrpclib
+
+final case class ConflictingMethodError(name: String) extends Exception {
+ override def getMessage(): String = s"Method $name is already handled"
+}

core/src/jsonrpclib/ConflictingMethodError.scala

@@ -0,0 +1,43 @@
+package jsonrpclib
+
+sealed trait Endpoint[F[_]] {
+ def method: String
+}
+
+object Endpoint {
+
+ def apply[F[_]](method: String): PartiallyAppliedEndpoint[F] = new PartiallyAppliedEndpoint[F](method)
+
+ class PartiallyAppliedEndpoint[F[_]](method: String) {
+ def apply[In, Err, Out](
+ run: In => F[Either[Err, Out]]
+ )(implicit inCodec: Codec[In], errCodec: ErrorCodec[Err], outCodec: Codec[Out]): Endpoint[F] =
+ RequestResponseEndpoint(method, run, inCodec, errCodec, outCodec)
+
+ def simple[In, Out](
+ run: In => F[Out]
+ )(implicit F: Monadic[F], inCodec: Codec[In], outCodec: Codec[Out]) =
+ apply[In, ErrorPayload, Out](in =>
+ F.doFlatMap(F.doAttempt(run(in))) {
+ case Left(error) => F.doPure(Left(ErrorPayload(0, error.getMessage(), None)))
+ case Right(value) => F.doPure(Right(value))
+ }
+ )
+
+ def notification[In](run: In => F[Unit])(implicit inCodec: Codec[In]): Endpoint[F] =
+ NotificationEndpoint(method, run, inCodec)
+
+ }
+
+ final case class NotificationEndpoint[F[_], In](method: String, run: In => F[Unit], inCodec: Codec[In])
+ extends Endpoint[F]
+
+ final case class RequestResponseEndpoint[F[_], In, Err, Out](
+ method: String,
+ run: In => F[Either[Err, Out]],
+ inCodec: Codec[In],
+ errCodec: ErrorCodec[Err],
+ outCodec: Codec[Out]
+ ) extends Endpoint[F]
+
+}

core/src/jsonrpclib/Endpoint.scala

@@ -1,17 +1,15 @@
 package jsonrpclib
 
-sealed trait EndpointTemplate
-
+sealed trait EndpointTemplate[In, Err, Out]
 object EndpointTemplate {
  final case class NotificationTemplate[In](
  method: String,
  inCodec: Codec[In]
- ) extends EndpointTemplate
-
+ ) extends EndpointTemplate[In, Nothing, Unit]
  final case class RequestResponseTemplate[In, Err, Out](
  method: String,
  inCodec: Codec[In],
  errCodec: ErrorCodec[Err],
  outCodec: Codec[Out]
- ) extends EndpointTemplate
+ ) extends EndpointTemplate[In, Err, Out]
 }

lib/src/EndpointTemplate.scala renamed to core/src/jsonrpclib/EndpointTemplate.scala

@@ -0,0 +1,15 @@
+package jsonrpclib
+
+trait ErrorCodec[E] {
+
+ def encode(a: E): ErrorPayload
+ def decode(error: ErrorPayload): Either[ProtocolError, E]
+
+}
+
+object ErrorCodec {
+ implicit val errorPayloadCodec: ErrorCodec[ErrorPayload] = new ErrorCodec[ErrorPayload] {
+ def encode(a: ErrorPayload): ErrorPayload = a
+ def decode(error: ErrorPayload): Either[ProtocolError, ErrorPayload] = Right(error)
+ }
+}

core/src/jsonrpclib/ErrorCodec.scala

@@ -3,7 +3,9 @@ package jsonrpclib
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
 import com.github.plokhotnyuk.jsoniter_scala.macros.JsonCodecMaker
 
-private[jsonrpclib] case class ErrorPayload(code: Int, message: String, data: Option[Payload])
+private[jsonrpclib] case class ErrorPayload(code: Int, message: String, data: Option[Payload]) extends Throwable {
+ override def getMessage(): String = message
+}
 
 private[jsonrpclib] object ErrorPayload {
 

lib/src/ErrorPayload.scala renamed to core/src/jsonrpclib/ErrorPayload.scala

@@ -0,0 +1,5 @@
+package jsonrpclib
+
+/** Errors that should not be sent back through the json rpc channel (such as invalid notifications)
+ */
+case class ErrorReport(method: String, payload: Payload, error: ProtocolError)

core/src/jsonrpclib/ErrorReport.scala

@@ -0,0 +1,20 @@
+package jsonrpclib
+
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext
+
+trait Monadic[F[_]] {
+ def doFlatMap[A, B](fa: F[A])(f: A => F[B]): F[B]
+ def doPure[A](a: A): F[A]
+ def doAttempt[A](fa: F[A]): F[Either[Throwable, A]]
+}
+
+object Monadic {
+ implicit def monadicFuture(implicit ec: ExecutionContext): Monadic[Future] = new Monadic[Future] {
+ def doFlatMap[A, B](fa: Future[A])(f: A => Future[B]): Future[B] = fa.flatMap(f)
+
+ def doPure[A](a: A): Future[A] = Future.successful(a)
+
+ def doAttempt[A](fa: Future[A]): Future[Either[Throwable, A]] = fa.map(Right(_)).recover(Left(_))
+ }
+}