Adds installation instructions and examples

Author: Baccata

.scalafmt.conf

@@ -1,3 +1,17 @@
 version = "3.5.3"
 runner.dialect = scala213
 maxColumn = 120
+fileOverride {
+  "glob:**/fs2/src/**" {
+     runner.dialect = scala213source3
+  }
+  "glob:**/fs2/test/src/**" {
+     runner.dialect = scala213source3
+  }
+  "glob:**/core/test/src-jvm-native/**" {
+     runner.dialect = scala213source3
+  }
+  "glob:**/core/src/**" {
+     runner.dialect = scala213source3
+  }
+}

README.md

@@ -1,23 +1,41 @@
 [![CI](https://github.com/neandertech/jsonrpclib/actions/workflows/ci.yml/badge.svg)](https://github.com/neandertech/jsonrpclib/actions/workflows/ci.yml)
 
+[![jsonrpclib-fs2 Scala version support](https://index.scala-lang.org/neandertech/jsonrpclib/jsonrpclib-fs2/latest-by-scala-version.svg?platform=jvm](https://index.scala-lang.org/neandertech/jsonrpclib/jsonrpclib-fs2)
+
+[![jsonrpclib-fs2 Scala version support](https://index.scala-lang.org/neandertech/jsonrpclib/jsonrpclib-fs2/latest-by-scala-version.svg?platform=sjs1](https://index.scala-lang.org/neandertech/jsonrpclib/jsonrpclib-fs2)
+
+
 # jsonrpclib
 
-This is a cross-platform, cross-scala-version [jsonrpc](https://www.jsonrpc.org/) library that provides construct for bidirectional communication using
-the jsonrpc protocol.
+This is a cross-platform, cross-scala-version library that provides construct for bidirectional communication using the [jsonrpc](https://www.jsonrpc.org/) protocol. It is built on top of [fs2](https://fs2.io/#/) and [jsoniter-scala](https://github.com/plokhotnyuk/jsoniter-scala)
+
+This library does not enforce any transport, and can work on top of stdin/stdout or other channels.
+
+## Installation
+
+The dependencies below are following [cross-platform semantics](http://youforgotapercentagesignoracolon.com/).
+Adapt according to your needs
+
+### SBT
 
-This library does not enforce any transport, and works as long as you can provide input/output byte streams.
+```scala
+libraryDependencies += "tech.neander" %%% "jsonrpclib-fs2" % version
+```
 
+### Mill
 
-## Dev Notes
+```scala
+override def ivyDeps = super.ivyDeps() ++ Agg(ivy"tech.neander::jsonrpclib-fs2::$version")
+```
 
-### Scala-native
+### Scala-cli
 
-See
-* https://github.com/scala-native/scala-native/blob/63d07093f6d0a6e9de28cd8f9fb6bc1d6596c6ec/test-interface/src/main/scala/scala/scalanative/testinterface/NativeRPC.scala
+```scala
+//> using lib "tech.neander::jsonrpclib-fs2:<VERSION>"
+```
 
+## Usage
 
-### Scala-js
+**/!\ Please be aware that this library is in its early days and offers strictly no guarantee with regards to backward compatibility**
 
-See
-* https://github.com/scala-js/scala-js-js-envs/blob/main/nodejs-env/src/main/scala/org/scalajs/jsenv/nodejs/ComSupport.scala#L245
-* https://github.com/scala-js/scala-js/blob/0708917912938714d52be1426364f78a3d1fd269/test-bridge/src/main/scala/org/scalajs/testing/bridge/JSRPC.scala
+See the examples folder

build.sc

@@ -1,3 +1,5 @@
+import mill.define.Target
+import mill.util.Jvm
 import $ivy.`com.lihaoyi::mill-contrib-bloop:$MILL_VERSION`
 import $ivy.`io.github.davidgregory084::mill-tpolecat::0.3.1`
 import $ivy.`de.tototec::de.tobiasroeser.mill.vcs.version::0.1.4`
@@ -20,6 +22,7 @@ object versions {
   val scalaNativeVersion = "0.4.4"
   val munitVersion = "0.7.29"
   val munitNativeVersion = "1.0.0-M6"
+  val fs2 = "3.2.11"
 
   val scala213 = "2.13"
   val scala212 = "2.12"
@@ -59,7 +62,7 @@ object fs2 extends RPCCrossPlatformModule { cross =>
 
   override def crossPlatformModuleDeps = Seq(core)
   def crossPlatformIvyDeps: T[Agg[Dep]] = Agg(
-    ivy"co.fs2::fs2-core::3.2.8"
+    ivy"co.fs2::fs2-core::${versions.fs2}"
   )
 
   object jvm extends mill.Cross[JvmModule](scala213, scala3)
@@ -74,6 +77,26 @@ object fs2 extends RPCCrossPlatformModule { cross =>
 
 }
 
+object examples extends mill.define.Module {
+
+  object server extends ScalaModule {
+    def ivyDeps = Agg(ivy"co.fs2::fs2-io:${versions.fs2}")
+    def moduleDeps = Seq(fs2.jvm(versions.scala213))
+    def scalaVersion = versions.scala213Version
+  }
+
+  object client extends ScalaModule {
+    def ivyDeps = Agg(ivy"co.fs2::fs2-io:${versions.fs2}", ivy"eu.monniot::fs2-process:0.4.4")
+    def moduleDeps = Seq(fs2.jvm(versions.scala213))
+    def scalaVersion = versions.scala213Version
+    def forkEnv: Target[Map[String, String]] = T {
+      val assembledServer = server.assembly()
+      super.forkEnv() ++ Map("SERVER_JAR" -> assembledServer.path.toString())
+    }
+  }
+
+}
+
 // #############################################################################
 //  COMMON SETUP
 // #############################################################################

devnotes.md

@@ -0,0 +1,15 @@
+# Dev Notes
+
+In case somebody wants to implement future-based channels, here are some source of inspiration :
+
+### Scala-native
+
+See
+* https://github.com/scala-native/scala-native/blob/63d07093f6d0a6e9de28cd8f9fb6bc1d6596c6ec/test-interface/src/main/scala/scala/scalanative/testinterface/NativeRPC.scala
+
+
+### Scala-js
+
+See
+* https://github.com/scala-js/scala-js-js-envs/blob/main/nodejs-env/src/main/scala/org/scalajs/jsenv/nodejs/ComSupport.scala#L245
+* https://github.com/scala-js/scala-js/blob/0708917912938714d52be1426364f78a3d1fd269/test-bridge/src/main/scala/org/scalajs/testing/bridge/JSRPC.scala

examples/client/src/examples/client/ClientMain.scala

@@ -0,0 +1,47 @@
+package examples.server
+
+import jsonrpclib.CallId
+import jsonrpclib.fs2._
+import cats.effect._
+import fs2.io._
+import eu.monniot.process.Process
+import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
+import com.github.plokhotnyuk.jsoniter_scala.macros.JsonCodecMaker
+import jsonrpclib.Endpoint
+import cats.syntax.all._
+import fs2.Stream
+import jsonrpclib.StubTemplate
+
+object ClientMain extends IOApp.Simple {
+
+  // Reserving a method for cancelation.
+  val cancelTemplate = CancelTemplate.make[CallId]("$/cancel", identity, identity)
+
+  // Creating a datatype that'll serve as a request (and response) of an endpoint
+  case class IntWrapper(value: Int)
+  object IntWrapper {
+    implicit val jcodec: JsonValueCodec[IntWrapper] = JsonCodecMaker.make
+  }
+
+  type IOStream[A] = fs2.Stream[IO, A]
+  def log(str: String): IOStream[Unit] = Stream.eval(IO.consoleForIO.errorln(str))
+
+  def run: IO[Unit] = {
+    // Using errorln as stdout is used by the RPC channel
+    val run = for {
+      _ <- log("Starting client")
+      serverJar <- sys.env.get("SERVER_JAR").liftTo[IOStream](new Exception("SERVER_JAR env var does not exist"))
+      serverProcess <- Stream.resource(Process.spawn[IO]("java", "-jar", serverJar))
+      fs2Channel <- FS2Channel.lspCompliant[IO](serverProcess.stdout, serverProcess.stdin)
+      // Opening the stream to be able to send and receive data
+      _ <- fs2Channel.openStream
+      // Creating a `IntWrapper => IO[IntWrapper]` stub that can call the server
+      increment = fs2Channel.simpleStub[IntWrapper, IntWrapper]("increment")
+      result <- Stream.eval(IO.pure(IntWrapper(0)).flatMap(increment))
+      _ <- log(s"Client received $result")
+      _ <- log("Terminating client")
+    } yield ()
+    run.compile.drain
+  }
+
+}

examples/server/src/examples/server/ServerMain.scala

@@ -0,0 +1,40 @@
+package examples.server
+
+import jsonrpclib.CallId
+import jsonrpclib.fs2._
+import cats.effect._
+import fs2.io._
+import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
+import com.github.plokhotnyuk.jsoniter_scala.macros.JsonCodecMaker
+import jsonrpclib.Endpoint
+
+object ServerMain extends IOApp.Simple {
+
+  // Reserving a method for cancelation.
+  val cancelTemplate = CancelTemplate.make[CallId]("$/cancel", identity, identity)
+
+  // Creating a datatype that'll serve as a request (and response) of an endpoint
+  case class IntWrapper(value: Int)
+  object IntWrapper {
+    implicit val jcodec: JsonValueCodec[IntWrapper] = JsonCodecMaker.make
+  }
+
+  // Implementing an incrementation endpoint
+  val increment = Endpoint[IO]("increment").simple { in: IntWrapper =>
+    IO.consoleForIO.errorln(s"Server received $in") >>
+      IO.pure(in.copy(value = in.value + 1))
+  }
+
+  def run: IO[Unit] = {
+    // Using errorln as stdout is used by the RPC channel
+    IO.consoleForIO.errorln("Starting server") >>
+      FS2Channel
+        .lspCompliant[IO](fs2.io.stdin[IO](bufSize = 512), fs2.io.stdout[IO])
+        .flatMap(_.withEndpointStream(increment)) // mounting an endpoint onto the channel
+        .flatMap(_.openStreamForever) // starts the communication
+        .compile
+        .drain
+        .guarantee(IO.consoleForIO.errorln("Terminating server"))
+  }
+
+}

fs2/src/jsonrpclib/fs2/FS2Channel.scala

@@ -19,21 +19,30 @@ import scala.util.Try
 import _root_.fs2.concurrent.SignallingRef
 
 trait FS2Channel[F[_]] extends Channel[F] {
-  def withEndpoint(endpoint: Endpoint[F])(implicit F: Functor[F]): Resource[F, Unit] =
-    Resource.make(mountEndpoint(endpoint))(_ => unmountEndpoint(endpoint.method))
+  def withEndpoint(endpoint: Endpoint[F])(implicit F: Functor[F]): Resource[F, FS2Channel[F]] =
+    Resource.make(mountEndpoint(endpoint))(_ => unmountEndpoint(endpoint.method)).map(_ => this)
 
-  def withEndpoints(endpoint: Endpoint[F], rest: Endpoint[F]*)(implicit F: Monad[F]): Resource[F, Unit] =
-    (endpoint :: rest.toList).traverse_(withEndpoint)
+  def withEndpointStream(endpoint: Endpoint[F])(implicit F: MonadCancelThrow[F]): Stream[F, FS2Channel[F]] =
+    Stream.resource(withEndpoint(endpoint))
+
+  def withEndpoints(endpoint: Endpoint[F], rest: Endpoint[F]*)(implicit F: Monad[F]): Resource[F, FS2Channel[F]] =
+    (endpoint :: rest.toList).traverse_(withEndpoint).as(this)
+
+  def withEndpointStream(endpoint: Endpoint[F], rest: Endpoint[F]*)(implicit
+      F: MonadCancelThrow[F]
+  ): Stream[F, FS2Channel[F]] =
+    Stream.resource(withEndpoints(endpoint, rest: _*))
 
   def open: Resource[F, Unit]
   def openStream: Stream[F, Unit]
+  def openStreamForever: Stream[F, Nothing]
 }
 
 object FS2Channel {
 
   def lspCompliant[F[_]: Concurrent](
       byteStream: Stream[F, Byte],
-      byteSink: Pipe[F, Byte, Nothing],
+      byteSink: Pipe[F, Byte, Unit],
       bufferSize: Int = 512,
       maybeCancelTemplate: Option[CancelTemplate] = None
   ): Stream[F, FS2Channel[F]] = internals.LSP.writeSink(byteSink, bufferSize).flatMap { sink =>
@@ -120,6 +129,7 @@ object FS2Channel {
 
     def open: Resource[F, Unit] = Resource.make[F, Unit](isOpen.set(true))(_ => isOpen.set(false))
     def openStream: Stream[F, Unit] = Stream.resource(open)
+    def openStreamForever: Stream[F, Nothing] = openStream.evalMap(_ => F.never)
 
     protected[fs2] def cancel(callId: CallId): F[Unit] = state.get.map(_.runningCalls.get(callId)).flatMap {
       case None        => F.unit

fs2/src/jsonrpclib/fs2/internals/LSP.scala

@@ -14,7 +14,7 @@ import java.nio.charset.StandardCharsets
 object LSP {
 
   def writeSink[F[_]: Concurrent](
-      writePipe: fs2.Pipe[F, Byte, Nothing],
+      writePipe: fs2.Pipe[F, Byte, Unit],
       bufferSize: Int
   ): Stream[F, Payload => F[Unit]] =
     Stream.eval(Queue.bounded[F, Payload](bufferSize)).flatMap { queue =>