fix: asyncpipe is profunctor

spago.lock

@@ -29,6 +29,7 @@ workspace:
         - parallel: ">=6.0.0 <7.0.0"
         - pipes: ">=8.0.0 <9.0.0"
         - prelude: ">=6.0.1 <7.0.0"
+        - profunctor
         - st: ">=6.2.0 <7.0.0"
         - strings: ">=6.0.1 <7.0.0"
         - tailrec: ">=6.1.0 <7.0.0"

spago.yaml

@@ -36,6 +36,7 @@ package:
     - parallel: ">=6.0.0 <7.0.0"
     - pipes: ">=8.0.0 <9.0.0"
     - prelude: ">=6.0.1 <7.0.0"
+    - profunctor: ">=6.0.1 <7.0.0"
     - st: ">=6.2.0 <7.0.0"
     - strings: ">=6.0.1 <7.0.0"
     - tailrec: ">=6.1.0 <7.0.0"

src/Pipes.Async.purs

@@ -21,6 +21,7 @@ import Data.Either (Either(..), either)
 import Data.Generic.Rep (class Generic)
 import Data.Maybe (Maybe(..), isNothing)
 import Data.Newtype (unwrap)
+import Data.Profunctor (class Profunctor)
 import Data.Show.Generic (genericShow)
 import Data.Time.Duration (Milliseconds)
 import Data.Traversable (traverse_)
@@ -92,7 +93,7 @@ instance Show a => Show (ReadResult a) where show = genericShow
 -- |    - Attempt to read a chunk
 -- | - `x -> m ReadSignal`
 -- |    - Block until the pipe is readable again (or reading must stop)
-data AsyncPipe x a b m =
+data AsyncPipe x m a b =
   AsyncPipe
     (m x)
     (x -> a -> m WriteResult)
@@ -100,9 +101,22 @@ data AsyncPipe x a b m =
     (x -> m (ReadResult b))
     (x -> m ReadSignal)
 
+instance Monad m => Functor (AsyncPipe x m a) where
+  map f (AsyncPipe init w aw r ar) = AsyncPipe init w aw (map (map f) <<< r) ar
+
+instance Monad m => Profunctor (AsyncPipe x m) where
+  dimap :: forall a b c d. (a -> b) -> (c -> d) -> _ b c -> _ a d
+  dimap ab cd (AsyncPipe init w aw r ar) =
+    AsyncPipe
+      init
+      (\x -> w x <<< ab)
+      aw
+      (map (map cd) <<< r)
+      ar
+
 -- | Wraps all fields of an `AsyncPipe` with logging to debug
 -- | behavior and timing.
-debug :: forall x a b m. MonadAff m => String -> AsyncPipe x a b m -> AsyncPipe x a b m
+debug :: forall x a b m. MonadAff m => String -> AsyncPipe x m a b -> AsyncPipe x m a b
 debug c (AsyncPipe init write awaitWrite read awaitRead) =
   let
     logL m = liftEffect $ log $ "[" <> c <> "] " <> m
@@ -188,7 +202,7 @@ debug c (AsyncPipe init write awaitWrite read awaitRead) =
 -- | * `read` will pass chunks to `parse` as fast as `parse` allows
 -- | * `parse` will parse chunks and yield them to `encode` as soon as they're ready
 -- | * `encode` will encode chunks and yield them to `write` as soon as they're ready
-sync :: forall x a b f p e m. MonadError e m => Alternative p => Parallel p m => MonadFork f m => MonadAff m => AsyncPipe x (Maybe a) (Maybe b) m -> Pipe (Maybe a) (Maybe b) m Unit
+sync :: forall x a b f p e m. MonadError e m => Alternative p => Parallel p m => MonadFork f m => MonadAff m => AsyncPipe x m (Maybe a) (Maybe b) -> Pipe (Maybe a) (Maybe b) m Unit
 sync (AsyncPipe init write awaitWrite read awaitRead) =
   let
     liftPipe :: forall r. (Proxy _ _ _ _ m) r -> ExceptT (Step _ _) (Proxy _ _ _ _ m) r
@@ -243,7 +257,7 @@ pipeAsync
   => MonadAff m
   => MonadBracket Error f m
   => Producer (Maybe a) m Unit
-  -> AsyncPipe x (Maybe a) (Maybe b) m
+  -> AsyncPipe x m (Maybe a) (Maybe b)
   -> Producer (Maybe b) m Unit
 pipeAsync prod (AsyncPipe init write awaitWrite read awaitRead) =
   do

src/Pipes.Node.Stream.purs

@@ -113,7 +113,7 @@ fromTransform
    . MonadThrow Error m
   => MonadAff m
   => Effect (O.Transform a b)
-  -> AsyncPipe (TransformContext a b) (Maybe a) (Maybe b) m
+  -> AsyncPipe (TransformContext a b) m (Maybe a) (Maybe b)
 fromTransform t =
   let
     init = do

src/Pipes.Node.Zlib.purs

@@ -17,29 +17,29 @@ import Pipes.Node.Stream (TransformContext, fromTransform)
 
 type X = TransformContext Buffer Buffer
 
-fromZlib :: forall r m. MonadAff m => MonadThrow Error m => Effect (ZlibStream r) -> AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+fromZlib :: forall r m. MonadAff m => MonadThrow Error m => Effect (ZlibStream r) -> AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 fromZlib z =
   fromTransform do
     raw <- liftEffect $ Zlib.toDuplex <$> z
     pure $ O.unsafeCoerceTransform raw
 
-gzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+gzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 gzip = fromZlib Zlib.createGzip
 
-gunzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+gunzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 gunzip = fromZlib Zlib.createGunzip
 
-unzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+unzip :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 unzip = fromZlib Zlib.createUnzip
 
-inflate :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+inflate :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 inflate = fromZlib Zlib.createInflate
 
-deflate :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+deflate :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 deflate = fromZlib Zlib.createDeflate
 
-brotliCompress :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+brotliCompress :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 brotliCompress = fromZlib Zlib.createBrotliCompress
 
-brotliDecompress :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X (Maybe Buffer) (Maybe Buffer) m
+brotliDecompress :: forall m. MonadAff m => MonadThrow Error m => AsyncPipe X m (Maybe Buffer) (Maybe Buffer)
 brotliDecompress = fromZlib Zlib.createBrotliDecompress