Change UX to be more compositional

Author: Baccata

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

@@ -39,11 +39,11 @@ object ClientMain extends IOApp.Simple {
       // Starting the server
       rp <- ChildProcess.spawn[IO]("java", "-jar", serverJar)
       // Creating a channel that will be used to communicate to the server
-      fs2Channel <- FS2Channel
-        .lspCompliant[IO](rp.stdout, rp.stdin, cancelTemplate = cancelEndpoint.some)
-        .concurrently(rp.stderr.through(fs2.io.stderr))
-      // Opening the stream to be able to send and receive data
-      _ <- fs2Channel.openStream
+      fs2Channel <- FS2Channel[IO](cancelTemplate = cancelEndpoint.some)
+      _ <- Stream(())
+        .concurrently(fs2Channel.output.through(lsp.encodePayloads).through(rp.stdin))
+        .concurrently(rp.stdout.through(lsp.decodePayloads).through(fs2Channel.input))
+        .concurrently(rp.stderr.through(fs2.io.stderr[IO]))
       // Creating a `IntWrapper => IO[IntWrapper]` stub that can call the server
       increment = fs2Channel.simpleStub[IntWrapper, IntWrapper]("increment")
       result1 <- Stream.eval(increment(IntWrapper(0)))

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

@@ -29,10 +29,14 @@ object ServerMain extends IOApp.Simple {
   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], cancelTemplate = cancelEndpoint.some)
+      FS2Channel[IO](cancelTemplate = Some(cancelEndpoint))
         .flatMap(_.withEndpointStream(increment)) // mounting an endpoint onto the channel
-        .flatMap(_.openStreamForever) // starts the communication
+        .flatMap(channel =>
+          fs2.Stream
+            .eval(IO.never) // running the server forever
+            .concurrently(stdin[IO](512).through(lsp.decodePayloads).through(channel.input))
+            .concurrently(channel.output.through(lsp.encodePayloads).through(stdout[IO]))
+        )
         .compile
         .drain
         .guarantee(IO.consoleForIO.errorln("Terminating server"))

fs2/src/jsonrpclib/fs2/FS2Channel.scala

@@ -16,50 +16,45 @@ import jsonrpclib.internals.MessageDispatcher
 import jsonrpclib.internals._
 
 import scala.util.Try
-import _root_.fs2.concurrent.SignallingRef
 
 trait FS2Channel[F[_]] extends Channel[F] {
+
+  def input: Pipe[F, Payload, Unit]
+  def output: Stream[F, Payload]
+
   def withEndpoint(endpoint: Endpoint[F])(implicit F: Functor[F]): Resource[F, FS2Channel[F]] =
     Resource.make(mountEndpoint(endpoint))(_ => unmountEndpoint(endpoint.method)).map(_ => this)
 
   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)
+    withEndpoints(endpoint +: rest)
+
+  def withEndpoints(endpoints: Seq[Endpoint[F]])(implicit F: Monad[F]): Resource[F, FS2Channel[F]] =
+    endpoints.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]
+  def withEndpointsStream(endpoints: Seq[Endpoint[F]])(implicit F: MonadCancelThrow[F]): Stream[F, FS2Channel[F]] =
+    Stream.resource(withEndpoints(endpoints))
+
 }
 
 object FS2Channel {
 
-  def lspCompliant[F[_]: Concurrent](
-      byteStream: Stream[F, Byte],
-      byteSink: Pipe[F, Byte, Unit],
-      bufferSize: Int = 512,
-      cancelTemplate: Option[CancelTemplate] = None
-  ): Stream[F, FS2Channel[F]] = internals.LSP.writeSink(byteSink, bufferSize).flatMap { sink =>
-    apply[F](internals.LSP.readStream(byteStream), sink, cancelTemplate)
-  }
-
   def apply[F[_]: Concurrent](
-      payloadStream: Stream[F, Payload],
-      payloadSink: Payload => F[Unit],
+      bufferSize: Int = 2048,
       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, cancelTemplate)
+      queue <- cats.effect.std.Queue.bounded[F, Payload](bufferSize).toStream
+      impl = new Impl(queue, ref, supervisor, cancelTemplate)
 
       // Creating a bespoke endpoint to receive cancelation requests
       maybeCancelEndpoint: Option[Endpoint[F]] = cancelTemplate.map { ct =>
@@ -71,10 +66,6 @@ object FS2Channel {
       }
       // mounting the cancelation endpoint
       _ <- maybeCancelEndpoint.traverse_(ep => impl.mountEndpoint(ep)).toStream
-      _ <- Stream(()).concurrently {
-        // Gatekeeping the pull until the channel is actually marked as open
-        payloadStream.pauseWhen(isOpen.map(b => !b)).evalMap(impl.handleReceivedPayload)
-      }
     } yield impl
   }
 
@@ -107,15 +98,17 @@ object FS2Channel {
   }
 
   private class Impl[F[_]](
-      private val sink: Payload => F[Unit],
+      private val queue: cats.effect.std.Queue[F, Payload],
       private val state: Ref[F, FS2Channel.State[F]],
-      private val isOpen: SignallingRef[F, Boolean],
       supervisor: Supervisor[F],
       maybeCancelTemplate: Option[CancelTemplate]
   )(implicit F: Concurrent[F])
       extends MessageDispatcher[F]
       with FS2Channel[F] {
 
+    def output: Stream[F, Payload] = Stream.fromQueueUnterminated(queue)
+    def input: Pipe[F, Payload, Unit] = _.evalMap(handleReceivedPayload)
+
     def mountEndpoint(endpoint: Endpoint[F]): F[Unit] = state
       .modify(s =>
         s.mountEndpoint(endpoint) match {
@@ -127,10 +120,6 @@ object FS2Channel {
 
     def unmountEndpoint(method: String): F[Unit] = state.update(_.removeEndpoint(method))
 
-    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
       case Some(fiber) => fiber.cancel
@@ -147,7 +136,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 sendMessage(message: Message): F[Unit] = queue.offer(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 {

fs2/src/jsonrpclib/fs2/internals/LSP.scala renamed to fs2/src/jsonrpclib/fs2/lsp.scala

@@ -1,33 +1,26 @@
-package jsonrpclib.fs2.internals
+package jsonrpclib.fs2
 
 import cats.MonadThrow
-import cats.effect.Concurrent
-import cats.effect.std.Queue
 import cats.implicits._
 import fs2.Chunk
 import fs2.Stream
+import fs2.Pipe
 import jsonrpclib.Payload
 
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets
 
-object LSP {
+object lsp {
 
-  def writeSink[F[_]: Concurrent](
-      writePipe: fs2.Pipe[F, Byte, Unit],
-      bufferSize: Int
-  ): Stream[F, Payload => F[Unit]] =
-    Stream.eval(Queue.bounded[F, Payload](bufferSize)).flatMap { queue =>
-      val payloads = fs2.Stream.fromQueueUnterminated(queue, bufferSize)
-      Stream(queue.offer(_)).concurrently(payloads.map(writeChunk).flatMap(Stream.chunk(_)).through(writePipe))
-    }
+  def encodePayloads[F[_]]: Pipe[F, Payload, Byte] =
+    (_: Stream[F, Payload]).map(writeChunk).flatMap(Stream.chunk(_))
 
   /** Split a stream of bytes into payloads by extracting each frame based on information contained in the headers.
     *
     * See https://microsoft.github.io/language-server-protocol/specifications/lsp/3.17/specification/#contentPart
     */
-  def readStream[F[_]: MonadThrow](bytes: Stream[F, Byte]): Stream[F, Payload] =
-    bytes
+  def decodePayloads[F[_]: MonadThrow]: Pipe[F, Byte, Payload] =
+    (_: Stream[F, Byte])
       .scanChunks(ScanState.starting) { case (state, chunk) =>
         val (ns, maybeResult) = loop(state.concatChunk(chunk))
         (ns, Chunk(maybeResult))

fs2/test/src/jsonrpclib/fs2/FS2ChannelSpec.scala

@@ -1,8 +1,6 @@
 package jsonrpclib.fs2
 
 import cats.effect.IO
-import cats.effect.implicits._
-import cats.effect.std.Queue
 import cats.syntax.all._
 import com.github.plokhotnyuk.jsoniter_scala.core.JsonValueCodec
 import com.github.plokhotnyuk.jsoniter_scala.macros.JsonCodecMaker
@@ -28,18 +26,28 @@ object FS2ChannelSpec extends SimpleIOSuite {
   def testRes(name: TestName)(run: Stream[IO, Expectations]): Unit =
     test(name)(run.compile.lastOrError.timeout(10.second))
 
+  type ClientSideChannel = FS2Channel[IO]
+  def setup(endpoints: Endpoint[IO]*) = setupAux(endpoints, None)
+  def setup(cancelTemplate: CancelTemplate, endpoints: Endpoint[IO]*) = setupAux(endpoints, Some(cancelTemplate))
+  def setupAux(endpoints: Seq[Endpoint[IO]], cancelTemplate: Option[CancelTemplate]): Stream[IO, ClientSideChannel] = {
+    for {
+      serverSideChannel <- FS2Channel[IO](cancelTemplate = cancelTemplate)
+      clientSideChannel <- FS2Channel[IO](cancelTemplate = cancelTemplate)
+      _ <- serverSideChannel.withEndpointsStream(endpoints)
+      _ <- Stream(())
+        .concurrently(clientSideChannel.output.through(serverSideChannel.input))
+        .concurrently(serverSideChannel.output.through(clientSideChannel.input))
+    } yield {
+      clientSideChannel
+    }
+  }
+
   testRes("Round trip") {
     val endpoint: Endpoint[IO] = Endpoint[IO]("inc").simple((int: IntWrapper) => IO(IntWrapper(int.int + 1)))
 
     for {
-      stdout <- Queue.bounded[IO, Payload](10).toStream
-      stdin <- Queue.bounded[IO, Payload](10).toStream
-      serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer)
-      clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), stdin.offer)
-      _ <- serverSideChannel.withEndpoint(endpoint).asStream
+      clientSideChannel <- setup(endpoint)
       remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc")
-      _ <- serverSideChannel.open.asStream
-      _ <- clientSideChannel.open.asStream
       result <- remoteFunction(IntWrapper(1)).toStream
     } yield {
       expect.same(result, IntWrapper(2))
@@ -49,13 +57,8 @@ object FS2ChannelSpec extends SimpleIOSuite {
   testRes("Endpoint not mounted") {
 
     for {
-      stdout <- Queue.bounded[IO, Payload](10).toStream
-      stdin <- Queue.bounded[IO, Payload](10).toStream
-      serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer)
-      clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), stdin.offer)
+      clientSideChannel <- setup()
       remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc")
-      _ <- serverSideChannel.open.asStream
-      _ <- clientSideChannel.open.asStream
       result <- remoteFunction(IntWrapper(1)).attempt.toStream
     } yield {
       expect.same(result, Left(ErrorPayload(-32601, "Method inc not found", None)))
@@ -70,14 +73,8 @@ object FS2ChannelSpec extends SimpleIOSuite {
       }
 
     for {
-      stdout <- Queue.bounded[IO, Payload](10).toStream
-      stdin <- Queue.bounded[IO, Payload](10).toStream
-      serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), payload => stdout.offer(payload))
-      clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), payload => stdin.offer(payload))
-      _ <- serverSideChannel.withEndpoint(endpoint).asStream
+      clientSideChannel <- setup(endpoint)
       remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc")
-      _ <- serverSideChannel.open.asStream
-      _ <- clientSideChannel.open.asStream
       timedResults <- (1 to 10).toList.map(IntWrapper(_)).parTraverse(remoteFunction).timed.toStream
     } yield {
       val (time, results) = timedResults
@@ -95,14 +92,8 @@ object FS2ChannelSpec extends SimpleIOSuite {
         IO.never.as(int).onCancel(canceledPromise.complete(int).void)
       )
 
-      stdin <- Queue.bounded[IO, Payload](10).toStream
-      stdout <- Queue.bounded[IO, Payload](10).toStream
-      serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer, Some(cancelTemplate))
-      clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), stdin.offer, Some(cancelTemplate))
-      _ <- serverSideChannel.withEndpoint(endpoint).asStream
+      clientSideChannel <- setup(cancelTemplate, endpoint)
       remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("never")
-      _ <- serverSideChannel.open.asStream
-      _ <- clientSideChannel.open.asStream
       // Timeing-out client-call to verify cancelation progagates to server
       _ <- IO.race(remoteFunction(IntWrapper(23)), IO.sleep(1.second)).toStream
       result <- canceledPromise.get.toStream