Using pauseWhen

Baccata · Baccata · commit 3d83715eafef · 2022-08-04T15:56:17.000+02:00
diff --git a/fs2/src/jsonrpclib/fs2/FS2Channel.scala b/fs2/src/jsonrpclib/fs2/FS2Channel.scala
@@ -49,8 +49,7 @@ object FS2Channel {
       impl = new Impl(awaitingSink, 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)
+        payloadStream.pauseWhen(isOpen.map(b => !b)).evalMap(impl.handleReceivedPayload)
       }
     } yield impl
   }

Original file line number	Diff line number	Diff line change
`@@ -49,8 +49,7 @@ object FS2Channel {`
`49`	`49`	`impl = new Impl(awaitingSink, ref, isOpen, supervisor)`
`50`	`50`	`_ <- Stream(()).concurrently {`
`51`	`51`	`// Gatekeeping the pull until the channel is actually marked as open`
`52`		`- val wait = isOpen.waitUntil(identity)`
`53`		`- payloadStream.evalTap(_ => wait).evalMap(impl.handleReceivedPayload)`
	`52`	`+ payloadStream.pauseWhen(isOpen.map(b => !b)).evalMap(impl.handleReceivedPayload)`
`54`	`53`	`}`
`55`	`54`	`} yield impl`
`56`	`55`	`}`