Adds a function to manually open the channel

Author: Baccata

fs2/src/jsonrpclib/fs2/FS2Channel.scala

@@ -14,13 +14,16 @@ import jsonrpclib.internals.MessageDispatcher
 import jsonrpclib.internals._
 
 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 withEndpoints(endpoint: Endpoint[F], rest: Endpoint[F]*)(implicit F: Monad[F]): Resource[F, Unit] =
     (endpoint :: rest.toList).traverse_(withEndpoint)
+
+  def open: Resource[F, Unit]
 }
 
 object FS2Channel {
@@ -42,16 +45,22 @@ object FS2Channel {
     val endpointsMap = startingEndpoints.map(ep => ep.method -> ep).toMap
     for {
       supervisor <- Stream.resource(Supervisor[F])
-      ref <- Ref[F].of(State[F](Map.empty, endpointsMap, 0)).toStream
-      impl = new Impl(payloadSink, ref, supervisor)
-      _ <- Stream(()).concurrently(payloadStream.evalMap(impl.handleReceivedPayload))
+      ref <- Ref[F].of(State[F](Map.empty, endpointsMap, 0, false)).toStream
+      isOpen <- SignallingRef[F].of(false).toStream
+      impl = new Impl(payloadSink, ref, isOpen, supervisor)
+      _ <- Stream(()).concurrently {
+        // Gatekeeping the pull until the channel is actually marked as open
+        val wait = isOpen.waitUntil(identity)
+        payloadStream.evalTap(_ => wait).evalMap(impl.handleReceivedPayload)
+      }
     } yield impl
   }
 
   private case class State[F[_]](
       pendingCalls: Map[CallId, OutputMessage => F[Unit]],
       endpoints: Map[String, Endpoint[F]],
-      counter: Long
+      counter: Long,
+      isOpen: Boolean
   ) {
     def nextCallId: (State[F], CallId) = (this.copy(counter = counter + 1), CallId.NumberId(counter))
     def storePendingCall(callId: CallId, handle: OutputMessage => F[Unit]): State[F] =
@@ -67,11 +76,15 @@ object FS2Channel {
       }
     def removeEndpoint(method: String): State[F] =
       copy(endpoints = endpoints.removed(method))
+
+    def open: State[F] = copy(isOpen = true)
+    def close: State[F] = copy(isOpen = false)
   }
 
   private class Impl[F[_]](
       private val sink: Payload => F[Unit],
       private val state: Ref[F, FS2Channel.State[F]],
+      private val isOpen: SignallingRef[F, Boolean],
       supervisor: Supervisor[F]
   )(implicit F: Concurrent[F])
       extends MessageDispatcher[F]
@@ -88,6 +101,8 @@ 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))
+
     protected def background[A](fa: F[A]): F[Unit] = supervisor.supervise(fa).void
     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))

fs2/src/jsonrpclib/fs2/package.scala

@@ -3,13 +3,19 @@ package jsonrpclib
 import _root_.fs2.Stream
 import cats.MonadThrow
 import cats.Monad
+import cats.effect.kernel.Resource
+import cats.effect.kernel.MonadCancel
 
 package object fs2 {
 
   private[jsonrpclib] implicit class EffectOps[F[_], A](private val fa: F[A]) extends AnyVal {
     def toStream: Stream[F, A] = Stream.eval(fa)
   }
 
+  private[jsonrpclib] implicit class ResourceOps[F[_], A](private val fa: Resource[F, A]) extends AnyVal {
+    def asStream(implicit F: MonadCancel[F, Throwable]): Stream[F, A] = Stream.resource(fa)
+  }
+
   implicit def catsMonadic[F[_]: MonadThrow]: Monadic[F] = new Monadic[F] {
     def doFlatMap[A, B](fa: F[A])(f: A => F[B]): F[B] = Monad[F].flatMap(fa)(f)
 

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

@@ -21,7 +21,7 @@ object FS2ChannelSpec extends SimpleIOSuite {
   }
 
   def testRes(name: TestName)(run: Stream[IO, Expectations]): Unit =
-    test(name)(run.compile.lastOrError)
+    test(name)(run.compile.lastOrError.timeout(10.second))
 
   testRes("Round trip") {
     val endpoint: Endpoint[IO] = Endpoint[IO]("inc").simple((int: IntWrapper) => IO(IntWrapper(int.int + 1)))
@@ -31,8 +31,10 @@ object FS2ChannelSpec extends SimpleIOSuite {
       stdin <- Queue.bounded[IO, Payload](10).toStream
       serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer)
       clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), stdin.offer)
-      _ <- Stream.resource(serverSideChannel.withEndpoint(endpoint))
+      _ <- serverSideChannel.withEndpoint(endpoint).asStream
       remoteFunction = clientSideChannel.simpleStub[IntWrapper, IntWrapper]("inc")
+      _ <- serverSideChannel.open.asStream
+      _ <- clientSideChannel.open.asStream
       result <- remoteFunction(IntWrapper(1)).toStream
     } yield {
       expect.same(result, IntWrapper(2))
@@ -44,9 +46,11 @@ object FS2ChannelSpec extends SimpleIOSuite {
     for {
       stdout <- Queue.bounded[IO, Payload](10).toStream
       stdin <- Queue.bounded[IO, Payload](10).toStream
-      _ <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer)
+      serverSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdin), stdout.offer)
       clientSideChannel <- FS2Channel[IO](Stream.fromQueueUnterminated(stdout), stdin.offer)
       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)))
@@ -65,8 +69,10 @@ object FS2ChannelSpec extends SimpleIOSuite {
       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))
-      _ <- Stream.resource(serverSideChannel.withEndpoint(endpoint))
+      _ <- serverSideChannel.withEndpoint(endpoint).asStream
       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