It works ... but client program doesn't terminate

Author: Baccata

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

@@ -4,18 +4,23 @@ 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
+import cats.effect.std.Dispatcher
+import scala.sys.process.ProcessIO
+import cats.effect.implicits._
+import scala.sys.process.{Process => SProcess}
+import java.io.OutputStream
+import java.io.InputStream
 
 object ClientMain extends IOApp.Simple {
 
   // Reserving a method for cancelation.
-  val cancelTemplate = CancelTemplate.make[CallId]("$/cancel", identity, identity)
+  val cancelEndpoint = 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)
@@ -27,21 +32,75 @@ object ClientMain extends IOApp.Simple {
   def log(str: String): IOStream[Unit] = Stream.eval(IO.consoleForIO.errorln(str))
 
   def run: IO[Unit] = {
+    import scala.concurrent.duration._
     // 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)
+      // Starting the server
+      (serverStdin, serverStdout, serverStderr) <- Stream.resource(process("java", "-jar", serverJar))
+      pipeErrors = serverStderr.through(fs2.io.stderr)
+      // Creating a channel that will be used to communicate to the server
+      fs2Channel <- FS2Channel
+        .lspCompliant[IO](serverStdout, serverStdin, cancelTemplate = cancelEndpoint.some)
+        .concurrently(pipeErrors)
       // 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))
+      result <- Stream.eval(increment(IntWrapper(0)))
       _ <- log(s"Client received $result")
       _ <- log("Terminating client")
     } yield ()
-    run.compile.drain
+    run.compile.drain.timeout(2.second)
   }
 
+  /** Wraps the spawning of a subprocess into fs2 friendly semantics
+    */
+  import scala.concurrent.duration._
+  def process(command: String*) = for {
+    dispatcher <- Dispatcher[IO]
+    stdinPromise <- IO.deferred[fs2.Pipe[IO, Byte, Unit]].toResource
+    stdoutPromise <- IO.deferred[fs2.Stream[IO, Byte]].toResource
+    stderrPromise <- IO.deferred[fs2.Stream[IO, Byte]].toResource
+    makeProcessBuilder = IO(sys.process.stringSeqToProcess(command))
+    makeProcessIO = IO(
+      new ProcessIO(
+        in = { (outputStream: OutputStream) =>
+          val pipe = writeOutputStreamFlushingChunks(IO(outputStream))
+          val fulfil = stdinPromise.complete(pipe)
+          dispatcher.unsafeRunSync(fulfil)
+        },
+        out = { (inputStream: InputStream) =>
+          val stream = fs2.io.readInputStream(IO(inputStream), 512)
+          val fulfil = stdoutPromise.complete(stream)
+          dispatcher.unsafeRunSync(fulfil)
+        },
+        err = { (inputStream: InputStream) =>
+          val stream = fs2.io.readInputStream(IO(inputStream), 512)
+          val fulfil = stderrPromise.complete(stream)
+          dispatcher.unsafeRunSync(fulfil)
+        }
+      )
+    )
+    makeProcess = (makeProcessBuilder, makeProcessIO).flatMapN { case (b, io) => IO.blocking(b.run(io)) }
+    _ <- Resource.make(makeProcess)((runningProcess) => IO.blocking(runningProcess.destroy()))
+    pipes <- (stdinPromise.get, stdoutPromise.get, stderrPromise.get).tupled.toResource
+  } yield pipes
+
+  /** Adds a flush after each chunk
+    */
+  def writeOutputStreamFlushingChunks[F[_]](
+      fos: F[OutputStream],
+      closeAfterUse: Boolean = true
+  )(implicit F: Sync[F]): fs2.Pipe[F, Byte, Nothing] =
+    s => {
+      def useOs(os: OutputStream): Stream[F, Nothing] =
+        s.chunks.foreach(c => F.interruptible(os.write(c.toArray)) >> F.blocking(os.flush()))
+
+      val os =
+        if (closeAfterUse) Stream.bracket(fos)(os => F.blocking(os.close()))
+        else Stream.eval(fos)
+      os.flatMap(os => useOs(os) ++ Stream.exec(F.blocking(os.flush())))
+    }
 }

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

@@ -7,11 +7,12 @@ import fs2.io._
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
 import com.github.plokhotnyuk.jsoniter_scala.macros.JsonCodecMaker
 import jsonrpclib.Endpoint
+import cats.syntax.all._
 
 object ServerMain extends IOApp.Simple {
 
   // Reserving a method for cancelation.
-  val cancelTemplate = CancelTemplate.make[CallId]("$/cancel", identity, identity)
+  val cancelEndpoint = 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)
@@ -29,7 +30,7 @@ object ServerMain extends IOApp.Simple {
     // 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])
+        .lspCompliant[IO](fs2.io.stdin[IO](bufSize = 512), fs2.io.stdout[IO], cancelTemplate = cancelEndpoint.some)
         .flatMap(_.withEndpointStream(increment)) // mounting an endpoint onto the channel
         .flatMap(_.openStreamForever) // starts the communication
         .compile

fs2/src/jsonrpclib/fs2/FS2Channel.scala

@@ -44,28 +44,28 @@ object FS2Channel {
       byteStream: Stream[F, Byte],
       byteSink: Pipe[F, Byte, Unit],
       bufferSize: Int = 512,
-      maybeCancelTemplate: Option[CancelTemplate] = None
+      cancelTemplate: Option[CancelTemplate] = None
   ): Stream[F, FS2Channel[F]] = internals.LSP.writeSink(byteSink, bufferSize).flatMap { sink =>
-    apply[F](internals.LSP.readStream(byteStream), sink, maybeCancelTemplate)
+    apply[F](internals.LSP.readStream(byteStream), sink, cancelTemplate)
   }
 
   def apply[F[_]: Concurrent](
       payloadStream: Stream[F, Payload],
       payloadSink: Payload => F[Unit],
-      maybeCancelTemplate: Option[CancelTemplate] = None
+      cancelTemplate: Option[CancelTemplate] = None
   ): Stream[F, FS2Channel[F]] = {
     for {
       supervisor <- Stream.resource(Supervisor[F])
       ref <- Ref[F].of(State[F](Map.empty, Map.empty, Map.empty, 0)).toStream
       isOpen <- SignallingRef[F].of(false).toStream
       awaitingSink = isOpen.waitUntil(identity) >> payloadSink(_: Payload)
-      impl = new Impl(awaitingSink, ref, isOpen, supervisor, maybeCancelTemplate)
+      impl = new Impl(awaitingSink, ref, isOpen, supervisor, cancelTemplate)
 
       // Creating a bespoke endpoint to receive cancelation requests
-      maybeCancelEndpoint: Option[Endpoint[F]] = maybeCancelTemplate.map { cancelTemplate =>
-        implicit val codec: Codec[cancelTemplate.C] = cancelTemplate.codec
-        Endpoint[F](cancelTemplate.method).notification[cancelTemplate.C] { request =>
-          val callId = cancelTemplate.toCallId(request)
+      maybeCancelEndpoint: Option[Endpoint[F]] = cancelTemplate.map { ct =>
+        implicit val codec: Codec[ct.C] = ct.codec
+        Endpoint[F](ct.method).notification[ct.C] { request =>
+          val callId = ct.toCallId(request)
           impl.cancel(callId)
         }
       }
@@ -148,6 +148,7 @@ object FS2Channel {
     protected def reportError(params: Option[Payload], error: ProtocolError, method: String): F[Unit] = ???
     protected def getEndpoint(method: String): F[Option[Endpoint[F]]] = state.get.map(_.endpoints.get(method))
     protected def sendMessage(message: Message): F[Unit] = sink(Codec.encode(message))
+
     protected def nextCallId(): F[CallId] = state.modify(_.nextCallId)
     protected def createPromise[A](callId: CallId): F[(Try[A] => F[Unit], () => F[A])] = Deferred[F, Try[A]].map {
       promise =>