fix: bump node-stream-pipes

spago.lock

@@ -19,13 +19,11 @@ workspace:
         - maybe: ">=6.0.0 <7.0.0"
         - node-buffer: ">=9.0.0 <10.0.0"
         - node-event-emitter: ">=3.0.0 <4.0.0"
-        - node-stream-pipes: ">=2.0.2 <3.0.0"
+        - node-stream-pipes: ">=2.1.0 <3.0.0"
         - node-streams: ">=9.0.0 <10.0.0"
         - nullable: ">=6.0.0 <7.0.0"
         - ordered-collections: ">=3.2.0 <4.0.0"
-        - pipes
         - prelude: ">=6.0.1 <7.0.0"
-        - profunctor
         - record: ">=4.0.0 <5.0.0"
         - simple-json: ">=9.0.0 <10.0.0"
         - tailrec: ">=6.1.0 <7.0.0"
@@ -633,8 +631,8 @@ packages:
       - effect
   node-stream-pipes:
     type: registry
-    version: 2.0.2
+    version: 2.1.0
-    integrity: sha256-IwkFgzWVwqjZkQRLYBGaRukKqYIw2I7wKHwIXRcdBWI=
+    integrity: sha256-pYBOQY4bGEZzI5UHsUxJAhsKqtmE6CC1sHmFqgj64V8=
     dependencies:
       - aff
       - arrays
@@ -662,7 +660,6 @@ packages:
       - parallel
       - pipes
       - prelude
-      - profunctor
       - st
       - strings
       - tailrec

spago.yaml

@@ -26,13 +26,11 @@ package:
     - maybe: ">=6.0.0 <7.0.0"
     - node-buffer: ">=9.0.0 <10.0.0"
     - node-event-emitter: ">=3.0.0 <4.0.0"
-    - node-stream-pipes: ">=2.0.2 <3.0.0"
+    - node-stream-pipes: ">=2.1.0 <3.0.0"
     - node-streams: ">=9.0.0 <10.0.0"
     - nullable: ">=6.0.0 <7.0.0"
     - ordered-collections: ">=3.2.0 <4.0.0"
-    - pipes: ">=8.0.0 <9.0.0"
     - prelude: ">=6.0.1 <7.0.0"
-    - profunctor: ">=6.0.1 <7.0.0"
     - record: ">=4.0.0 <5.0.0"
     - simple-json: ">=9.0.0 <10.0.0"
     - tailrec: ">=6.1.0 <7.0.0"

src/Pipes.CBOR.purs

@@ -7,21 +7,15 @@ import Control.Monad.Except (runExcept)
 import Control.Monad.Rec.Class (class MonadRec)
 import Data.Bifunctor (lmap)
 import Data.CBOR (class ReadCBOR, class WriteCBOR, readCBOR, writeCBOR)
-import Data.Either (Either)
 import Data.Maybe (Maybe)
-import Data.Profunctor as Pro
 import Data.Traversable (traverse)
 import Effect.Aff.Class (class MonadAff)
 import Effect.Exception (Error, error)
-import Foreign (Foreign, MultipleErrors)
 import Node.Buffer (Buffer)
 import Node.Stream.CBOR.Decode as CBOR.Decode
 import Node.Stream.CBOR.Encode as CBOR.Encode
-import Pipes.Async (AsyncPipe)
+import Pipes.Async (AsyncPipe, bindIO, mapIO)
-import Pipes.Core (Pipe)
-import Pipes.Node.Stream (TransformContext)
 import Pipes.Node.Stream as Pipes.Stream
-import Pipes.Prelude as Pipe
 
 -- | Transforms buffer chunks of a CBOR file to parsed values
 -- | of type `a`.
@@ -31,14 +25,12 @@ decode
   => MonadAff m
   => MonadThrow Error m
   => ReadCBOR a
-  => AsyncPipe (TransformContext Buffer Foreign) m (Maybe Buffer) (Maybe (Either MultipleErrors a))
+  => AsyncPipe (Maybe Buffer) (Maybe a) m Unit
 decode = do
   let
-    parser = Pipes.Stream.fromTransform $ CBOR.Decode.toObjectStream <$> CBOR.Decode.make {}
+    decoder = Pipes.Stream.fromTransformEffect $ CBOR.Decode.toObjectStream <$> CBOR.Decode.make {}
-  Pro.rmap (map (runExcept <<< readCBOR @a)) parser
+    parse = liftEither <<< lmap (error <<< show) <<< runExcept <<< readCBOR @a
-
+  bindIO pure (traverse parse) decoder
-decodeError :: forall m a r. MonadThrow Error m => Pipe (Maybe (Either MultipleErrors a)) (Maybe a) m r
-decodeError = Pipe.mapM (traverse liftEither <<< map (lmap $ error <<< show))
 
 -- | Encode purescript values as CBOR buffers
 encode
@@ -47,9 +39,9 @@ encode
   => MonadThrow Error m
   => MonadRec m
   => WriteCBOR a
-  => AsyncPipe (TransformContext Foreign Buffer) m (Maybe a) (Maybe Buffer)
+  => AsyncPipe (Maybe a) (Maybe Buffer) m Unit
 encode =
   let
-    p = Pipes.Stream.fromTransform $ CBOR.Encode.toObjectStream <$> CBOR.Encode.make {}
+    p = Pipes.Stream.fromTransformEffect $ CBOR.Encode.toObjectStream <$> CBOR.Encode.make {}
   in
-    Pro.lcmap (map writeCBOR) p
+    mapIO (map writeCBOR) identity p

test/Test/Pipes.CSV.purs

@@ -2,23 +2,17 @@ module Test.Pipes.CBOR where
 
 import Prelude
 
-import Control.Monad.Cont (lift)
-import Control.Monad.Error.Class (liftEither)
 import Control.Monad.Gen (chooseInt)
-import Data.Bifunctor (lmap)
 import Data.DateTime (DateTime)
 import Data.List ((:))
 import Data.List as List
 import Data.Maybe (Maybe(..), fromJust)
 import Data.Newtype (wrap)
 import Data.PreciseDateTime (fromRFC3339String, toDateTimeLossy)
-import Data.Traversable (traverse)
 import Data.Tuple.Nested ((/\))
 import Effect (Effect)
-import Effect.Aff (delay)
 import Effect.CBOR as CBOR
 import Effect.Class (liftEffect)
-import Effect.Exception (error)
 import Node.Buffer (Buffer)
 import Node.Buffer as Buffer
 import Node.Encoding (Encoding(..))
@@ -28,7 +22,7 @@ import Pipes.Async (debug, (>-/->))
 import Pipes.CBOR as Pipes.CBOR
 import Pipes.Collect as Pipes.Collect
 import Pipes.Node.Stream as Pipes.Stream
-import Pipes.Prelude (mapM, toListM) as Pipes
+import Pipes.Prelude (toListM) as Pipes
 import Test.QuickCheck.Gen (randomSample')
 import Test.Spec (Spec, before, describe, it)
 import Test.Spec.Assertions (shouldEqual)
@@ -67,7 +61,6 @@ spec =
         rows <- Pipes.toListM
           $ (yield (Just buf) *> yield Nothing)
               >-/-> debug "cbor" Pipes.CBOR.decode
-              >-> Pipes.CBOR.decodeError
 
         rows `shouldEqual` ((Just exp) : Nothing : List.Nil)
       before
@@ -87,7 +80,6 @@ spec =
             Pipes.Collect.toArray
               $ Pipes.Stream.withEOS (yield bytes)
                   >-/-> Pipes.CBOR.decode @(Array {id :: Int})
-                  >-> Pipes.CBOR.decodeError
                   >-> Pipes.Stream.unEOS
 
           rows `shouldEqual` [(\id -> { id }) <$> nums]