2222package fs2
2323package io
2424
25+ import cats .effect .SyncIO
2526import cats .effect .kernel .Async
2627import cats .effect .kernel .Resource
2728import cats .effect .std .Dispatcher
@@ -41,58 +42,83 @@ import scala.scalajs.js.|
4142
4243private [fs2] trait ioplatform {
4344
45+ @ deprecated(" Use suspendReadableAndRead instead" " 3.1.4"
4446 def readReadable [F [_]](
4547 readable : F [Readable ],
4648 destroyIfNotEnded : Boolean = true ,
4749 destroyIfCanceled : Boolean = true
4850 )(implicit
4951 F : Async [F ]
50- ): Stream [F , Byte ] =
51- Stream
52- .resource(for {
53- readable <- Resource .makeCase(readable.map(_.asInstanceOf [streamMod.Readable ])) {
52+ ): Stream [F , Byte ] = Stream
53+ .eval(readable)
54+ .flatMap(r => Stream .resource(suspendReadableAndRead(destroyIfNotEnded, destroyIfCanceled)(r)))
55+ .flatMap(_._2)
56+
57+ /** Suspends the creation of a `Readable` and a `Stream` that reads all bytes from that `Readable`.
58+ */
59+ def suspendReadableAndRead [F [_], R <: Readable ](
60+ destroyIfNotEnded : Boolean = true ,
61+ destroyIfCanceled : Boolean = true
62+ )(thunk : => R )(implicit F : Async [F ]): Resource [F , (R , Stream [F , Byte ])] =
63+ (for {
64+ dispatcher <- Dispatcher [F ]
65+ queue <- Queue .synchronous[F , Option [Unit ]].toResource
66+ error <- F .deferred[Throwable ].toResource
67+ // Implementation Note: why suspend in `SyncIO` and then `unsafeRunSync()` inside `F.delay`?
68+ // In many cases creating a `Readable` starts async side-effects (e.g. negotiating TLS handshake or opening a file handle).
69+ // Furthermore, these side-effects will invoke the listeners we register to the `Readable`.
70+ // Therefore, it is critical that the listeners are registered to the `Readable` _before_ these async side-effects occur:
71+ // in other words, before we next yield (cede) to the event loop. Because an arbitrary effect `F` (particularly `IO`) may cede at any time,
72+ // our only recourse is to suspend the entire creation/listener registration process within a single atomic `delay`.
73+ readableResource = for {
74+ readable <- Resource .makeCase(SyncIO (thunk).map(_.asInstanceOf [streamMod.Readable ])) {
5475 case (readable, Resource .ExitCase .Succeeded ) =>
55- F .delay {
76+ SyncIO {
5677 if (! readable.readableEnded & destroyIfNotEnded)
5778 readable.destroy()
5879 }
5980 case (readable, Resource .ExitCase .Errored (ex)) =>
60- F .delay (readable.destroy(js.Error (ex.getMessage())))
81+ SyncIO (readable.destroy(js.Error (ex.getMessage())))
6182 case (readable, Resource .ExitCase .Canceled ) =>
6283 if (destroyIfCanceled)
63- F .delay (readable.destroy())
84+ SyncIO (readable.destroy())
6485 else
65- F .unit
86+ SyncIO .unit
6687 }
67- dispatcher <- Dispatcher [F ]
68- queue <- Queue .synchronous[F , Option [Unit ]].toResource
69- error <- F .deferred[Throwable ].toResource
7088 _ <- registerListener0(readable, nodeStrings.readable)(_.on_readable(_, _)) { () =>
7189 dispatcher.unsafeRunAndForget(queue.offer(Some (())))
72- }
90+ }( SyncIO .syncForSyncIO)
7391 _ <- registerListener0(readable, nodeStrings.end)(_.on_end(_, _)) { () =>
7492 dispatcher.unsafeRunAndForget(queue.offer(None ))
75- }
93+ }( SyncIO .syncForSyncIO)
7694 _ <- registerListener0(readable, nodeStrings.close)(_.on_close(_, _)) { () =>
7795 dispatcher.unsafeRunAndForget(queue.offer(None ))
78- }
96+ }( SyncIO .syncForSyncIO)
7997 _ <- registerListener[js.Error ](readable, nodeStrings.error)(_.on_error(_, _)) { e =>
8098 dispatcher.unsafeRunAndForget(error.complete(js.JavaScriptException (e)))
81- }
82- } yield (readable, queue, error))
83- .flatMap { case (readable, queue, error) =>
84- Stream
99+ }(SyncIO .syncForSyncIO)
100+ } yield readable
101+ readable <- Resource
102+ .make(F .delay {
103+ readableResource.allocated.unsafeRunSync()
104+ }) { case (_, close) => close.to[F ] }
105+ .map(_._1)
106+ stream =
107+ (Stream
85108 .fromQueueNoneTerminated(queue)
86109 .concurrently(Stream .eval(error.get.flatMap(F .raiseError[Unit ]))) >>
87- Stream .evalUnChunk(
88- F .delay (
89- Option (readable.read(). asInstanceOf [bufferMod.global. Buffer ])
90- .fold( Chunk .empty[ Byte ])(_.toChunk )
91- )
92- )
93- }
94- .adaptError { case IOException (ex) => ex }
110+ Stream
111+ .evalUnChunk (
112+ F .delay(
113+ Option (readable.read(). asInstanceOf [bufferMod.global. Buffer ] )
114+ .fold( Chunk .empty[ Byte ])(_.toChunk )
115+ )
116+ )).adaptError { case IOException (ex) => ex }
117+ } yield (readable. asInstanceOf [ R ], stream)) .adaptError { case IOException (ex) => ex }
95118
119+ /** `Pipe` that converts a stream of bytes to a stream that will emit a single `Readable`,
120+ * that ends whenever the resulting stream terminates.
121+ */
96122 def toReadable [F [_]](implicit F : Async [F ]): Pipe [F , Byte , Readable ] =
97123 in =>
98124 Stream
@@ -111,9 +137,13 @@ private[fs2] trait ioplatform {
111137 }
112138 .adaptError { case IOException (ex) => ex }
113139
140+ /** Like [[toReadable ]] but returns a `Resource` rather than a single element stream.
141+ */
114142 def toReadableResource [F [_]: Async ](s : Stream [F , Byte ]): Resource [F , Readable ] =
115143 s.through(toReadable).compile.resource.lastOrError
116144
145+ /** Writes all bytes to the specified `Writable`.
146+ */
117147 def writeWritable [F [_]](
118148 writable : F [Writable ],
119149 endAfterUse : Boolean = true
@@ -147,9 +177,17 @@ private[fs2] trait ioplatform {
147177 }
148178 .adaptError { case IOException (ex) => ex }
149179
180+ /** Take a function that emits to a `Writable` effectfully,
181+ * and return a stream which, when run, will perform that function and emit
182+ * the bytes recorded in the `Writable` as an fs2.Stream
183+ */
150184 def readWritable [F [_]: Async ](f : Writable => F [Unit ]): Stream [F , Byte ] =
151185 Stream .empty.through(toDuplexAndRead(f))
152186
187+ /** Take a function that reads and writes from a `Duplex` effectfully,
188+ * and return a pipe which, when run, will perform that function,
189+ * write emitted bytes to the duplex, and read emitted bytes from the duplex
190+ */
153191 def toDuplexAndRead [F [_]: Async ](f : Duplex => F [Unit ]): Pipe [F , Byte , Byte ] =
154192 in =>
155193 Stream .resource(mkDuplex(in)).flatMap { case (duplex, out) =>
0 commit comments