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,67 +42,79 @@ import scala.scalajs.js.|
4142
4243private [fs2] trait ioplatform {
4344
44- @ deprecated(" Use readReadableResource for safer variant " " 3.1.4"
45+ @ deprecated(" Use suspendReadableAndRead instead " " 3.1.4"
4546 def readReadable [F [_]](
4647 readable : F [Readable ],
4748 destroyIfNotEnded : Boolean = true ,
4849 destroyIfCanceled : Boolean = true
4950 )(implicit
5051 F : Async [F ]
51- ): Stream [F , Byte ] =
52- Stream .resource(readReadableResource(readable, destroyIfNotEnded, destroyIfCanceled)).flatten
52+ ): Stream [F , Byte ] = Stream
53+ .eval(readable)
54+ .flatMap(r => Stream .resource(suspendReadableAndRead(destroyIfNotEnded, destroyIfCanceled)(r)))
55+ .flatMap(_._2)
5356
54- /** Reads all bytes from the specified `Readable`.
55- * Note that until the resource is opened, it is not safe to use the `Readable`
56- * without risking loss of data or events (e.g., termination, error).
57+ /** Suspends the creation of a `Readable` and a `Stream` that reads all bytes from that `Readable`.
5758 */
58- def readReadableResource [F [_]](
59- readable : F [Readable ],
59+ def suspendReadableAndRead [F [_], R <: Readable ](
6060 destroyIfNotEnded : Boolean = true ,
6161 destroyIfCanceled : Boolean = true
62- )(implicit
63- F : Async [F ]
64- ): Resource [F , Stream [F , Byte ]] =
62+ )(thunk : => R )(implicit F : Async [F ]): Resource [F , (R , Stream [F , Byte ])] =
6563 (for {
66- readable <- Resource .makeCase(readable.map(_.asInstanceOf [streamMod.Readable ])) {
67- case (readable, Resource .ExitCase .Succeeded ) =>
68- F .delay {
69- if (! readable.readableEnded & destroyIfNotEnded)
70- readable.destroy()
71- }
72- case (readable, Resource .ExitCase .Errored (ex)) =>
73- F .delay(readable.destroy(js.Error (ex.getMessage())))
74- case (readable, Resource .ExitCase .Canceled ) =>
75- if (destroyIfCanceled)
76- F .delay(readable.destroy())
77- else
78- F .unit
79- }
8064 dispatcher <- Dispatcher [F ]
8165 queue <- Queue .synchronous[F , Option [Unit ]].toResource
8266 error <- F .deferred[Throwable ].toResource
83- _ <- registerListener0(readable, nodeStrings.readable)(_.on_readable(_, _)) { () =>
84- dispatcher.unsafeRunAndForget(queue.offer(Some (())))
85- }
86- _ <- registerListener0(readable, nodeStrings.end)(_.on_end(_, _)) { () =>
87- dispatcher.unsafeRunAndForget(queue.offer(None ))
88- }
89- _ <- registerListener0(readable, nodeStrings.close)(_.on_close(_, _)) { () =>
90- dispatcher.unsafeRunAndForget(queue.offer(None ))
91- }
92- _ <- registerListener[js.Error ](readable, nodeStrings.error)(_.on_error(_, _)) { e =>
93- dispatcher.unsafeRunAndForget(error.complete(js.JavaScriptException (e)))
94- }
95- } yield (Stream
96- .fromQueueNoneTerminated(queue)
97- .concurrently(Stream .eval(error.get.flatMap(F .raiseError[Unit ]))) >>
98- Stream
99- .evalUnChunk(
100- F .delay(
101- Option (readable.read().asInstanceOf [bufferMod.global.Buffer ])
102- .fold(Chunk .empty[Byte ])(_.toChunk)
103- )
104- )).adaptError { case IOException (ex) => ex }).adaptError { case IOException (ex) => ex }
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 ])) {
75+ case (readable, Resource .ExitCase .Succeeded ) =>
76+ SyncIO {
77+ if (! readable.readableEnded & destroyIfNotEnded)
78+ readable.destroy()
79+ }
80+ case (readable, Resource .ExitCase .Errored (ex)) =>
81+ SyncIO (readable.destroy(js.Error (ex.getMessage())))
82+ case (readable, Resource .ExitCase .Canceled ) =>
83+ if (destroyIfCanceled)
84+ SyncIO (readable.destroy())
85+ else
86+ SyncIO .unit
87+ }
88+ _ <- registerListener0(readable, nodeStrings.readable)(_.on_readable(_, _)) { () =>
89+ dispatcher.unsafeRunAndForget(queue.offer(Some (())))
90+ }(SyncIO .syncForSyncIO)
91+ _ <- registerListener0(readable, nodeStrings.end)(_.on_end(_, _)) { () =>
92+ dispatcher.unsafeRunAndForget(queue.offer(None ))
93+ }(SyncIO .syncForSyncIO)
94+ _ <- registerListener0(readable, nodeStrings.close)(_.on_close(_, _)) { () =>
95+ dispatcher.unsafeRunAndForget(queue.offer(None ))
96+ }(SyncIO .syncForSyncIO)
97+ _ <- registerListener[js.Error ](readable, nodeStrings.error)(_.on_error(_, _)) { e =>
98+ dispatcher.unsafeRunAndForget(error.complete(js.JavaScriptException (e)))
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
108+ .fromQueueNoneTerminated(queue)
109+ .concurrently(Stream .eval(error.get.flatMap(F .raiseError[Unit ]))) >>
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 }
105118
106119 /** `Pipe` that converts a stream of bytes to a stream that will emit a single `Readable`,
107120 * that ends whenever the resulting stream terminates.
0 commit comments