orion/purescript-node-stream-pipes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: introduced transform bug

Browse Source
This commit is contained in:
orion 2024-06-21 13:21:19 -05:00
parent 4b91ab7d5c
commit d76f55e267
Signed by: orion
GPG Key ID: 6D4165AE4C928719
2 changed files with 14 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/Node.Stream.Object.purs
View File

@ -172,9 +172,10 @@ unsafeFromStringWritable :: forall r. Stream.Writable r -> Writable String
unsafeFromStringWritable = unsafeCoerce unsafeFromStringWritable = unsafeCoerce
awaitReadableOrClosed :: forall s a. Read s a => s -> Aff Unit awaitReadableOrClosed :: forall s a. Read s a => s -> Aff Unit
awaitReadableOrClosed s = do awaitReadableOrClosed s = Aff.supervise do
fiber <- fiber <-
Aff.forkAff $ parOneOf Aff.forkAff
$ parOneOf
[ onceAff0 readableH s $> Right unit [ onceAff0 readableH s $> Right unit
, onceAff0 closeH s $> Right unit , onceAff0 closeH s $> Right unit
, Left <$> onceAff1 errorH s , Left <$> onceAff1 errorH s
@ -189,14 +190,20 @@ awaitReadableOrClosed s = do
Aff.killFiber (error "") fiber Aff.killFiber (error "") fiber
awaitFinished :: forall s a. Write s a => s -> Aff Unit awaitFinished :: forall s a. Write s a => s -> Aff Unit
awaitFinished s = do awaitFinished s = Aff.supervise do
fiber <- Aff.forkAff $ onceAff0 finishH s fiber <- Aff.forkAff $ onceAff0 finishH s
finished <- liftEffect $ isWritableFinished s finished <- liftEffect $ isWritableFinished s
if not finished then Aff.joinFiber fiber else Aff.killFiber (error "") fiber if not finished then Aff.joinFiber fiber else Aff.killFiber (error "") fiber
awaitWritableOrClosed :: forall s a. Write s a => s -> Aff Unit awaitWritableOrClosed :: forall s a. Write s a => s -> Aff Unit
awaitWritableOrClosed s = do awaitWritableOrClosed s = Aff.supervise do
fiber <- Aff.forkAff $ parOneOf [ onceAff0 drainH s $> Right unit, onceAff0 closeH s $> Right unit, Left <$> onceAff1 errorH s ] fiber <-
Aff.forkAff
$ parOneOf
[ onceAff0 drainH s $> Right unit
, onceAff0 closeH s $> Right unit
, Left <$> onceAff1 errorH s
]
closed <- liftEffect $ isClosed s closed <- liftEffect $ isClosed s
writeEnded <- liftEffect $ isWritableEnded s writeEnded <- liftEffect $ isWritableEnded s
writable <- liftEffect $ isWritable s writable <- liftEffect $ isWritable s

3
src/Pipes.Node.Stream.purs
View File

@ -7,6 +7,7 @@ import Control.Monad.Rec.Class (class MonadRec, Step(..), tailRecM, whileJust)
import Control.Monad.ST.Class (liftST) import Control.Monad.ST.Class (liftST)
import Control.Monad.ST.Ref as STRef import Control.Monad.ST.Ref as STRef
import Control.Monad.Trans.Class (lift) import Control.Monad.Trans.Class (lift)
import Control.Parallel (parOneOf)
import Data.Maybe (Maybe(..), maybe) import Data.Maybe (Maybe(..), maybe)
import Data.Traversable (for_, traverse, traverse_) import Data.Traversable (for_, traverse, traverse_)
import Data.Tuple.Nested ((/\)) import Data.Tuple.Nested ((/\))
@ -118,7 +119,7 @@ fromTransform t = do
ended <- liftEffect $ O.isWritableEnded t ended <- liftEffect $ O.isWritableEnded t
if needsDrain then do if needsDrain then do
yieldWhileReadable yieldWhileReadable
liftAff $ O.awaitWritableOrClosed t liftAff $ parOneOf [O.awaitWritableOrClosed t, O.awaitReadableOrClosed t]
pure $ Loop unit pure $ Loop unit
else if ended then else if ended then
cleanup $> Done unit cleanup $> Done unit