packages feed

piped (empty) → 0.1.0.0

raw patch · 20 files changed

+1457/−0 lines, 20 filesdep +basedep +bench-showdep +conduitsetup-changed

Dependencies added: base, bench-show, conduit, gauge, microlens-platform, mtl, piped, quickcheck-instances, split, tasty, tasty-discover, tasty-golden, tasty-hunit, tasty-quickcheck

Files

+ ChangeLog.md view
@@ -0,0 +1,3 @@+# Changelog for piped++## Unreleased changes
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Author name here (c) 2019++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ piped.cabal view
@@ -0,0 +1,106 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.31.1.+--+-- see: https://github.com/sol/hpack+--+-- hash: be93d4d46d6f3b7018cd97af12bc628bf189c2940205fea9d2bc5949051b0c11++name:           piped+version:        0.1.0.0+synopsis:       Conduit with a smaller core+description:    Please see the README on GitHub at <https://github.com/ssadler/piped#readme>+category:       Control+homepage:       https://github.com/ssadler/piped#readme+bug-reports:    https://github.com/ssadler/piped/issues+author:         Scott Sadler+maintainer:     example@example.com+copyright:      2019 Scott Sadler+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    ChangeLog.md++source-repository head+  type: git+  location: https://github.com/ssadler/piped++flag bench+  description: Enable build benchmark+  manual: True+  default: False++library+  exposed-modules:+      Piped+      Piped.Compose+      Piped.Extras+      Piped.Internal+      Piped.Prelude+      Piped.Resume+  other-modules:+      Paths_piped+  hs-source-dirs:+      src+  build-depends:+      base >=4.7 && <5+    , mtl+  default-language: Haskell2010++executable piped-bench+  main-is: Bench.hs+  other-modules:+      BenchPipes+      BenchYields+      BenchCompare+      Paths_piped+  hs-source-dirs:+      test-bench+      test-common+  default-extensions: LambdaCase FlexibleContexts+  ghc-options: -rtsopts -O2 -fprof-auto -fwarn-unused-imports+  build-depends:+      base >=4.7 && <5+    , bench-show+    , conduit+    , gauge+    , mtl+    , piped+    , split+  if (flag(bench))+    buildable: True+  else+    buildable: False+  default-language: Haskell2010++test-suite piped-test+  type: exitcode-stdio-1.0+  main-is: Discover.hs+  other-modules:+      BenchSpec+      ComposeSpec+      ConduitSpec+      PipeLike+      TestUtils+      BenchCompare+      Paths_piped+  hs-source-dirs:+      test+      test-common+  default-extensions: RankNTypes MultiParamTypeClasses FunctionalDependencies FlexibleInstances ConstraintKinds TypeSynonymInstances AllowAmbiguousTypes LambdaCase FlexibleContexts+  ghc-options: -threaded -rtsopts -with-rtsopts=-N -fwarn-unused-imports+  build-depends:+      base >=4.7 && <5+    , conduit+    , gauge+    , microlens-platform+    , mtl+    , piped+    , quickcheck-instances+    , tasty+    , tasty-discover+    , tasty-golden+    , tasty-hunit+    , tasty-quickcheck+  default-language: Haskell2010
+ src/Piped.hs view
@@ -0,0 +1,31 @@++module Piped (++  -- Core+  +    Pipe+  , await+  , yield+  , (.|)+  , (|.)+  , runPipe+  , leftover++  -- Barebones API+  +  , awaitForever+  , awaitMaybe+  , awaitJust+  , yieldM+  , sourceList+  , sinkList+  , sinkNull++  , Void++  ) where++import Piped.Extras+import Piped.Compose+import Piped.Internal+
+ src/Piped/Compose.hs view
@@ -0,0 +1,100 @@++-- | Pipe composition can be supply or demand driven, which refers to how the pipeline is+--   initiated. Supply driven initiates the left side first.+--+--   When a pipe terminates, one side must return a value. Unless the pipeline is run+--   in resumable mode, the other sides may never "complete" in that they do not+--   return a value. If they allocate resources, MonadResource or similar should be+--   used to handle cleanup (see 'Piped.Extras.bracketPipe').+--+--   Right hand termination works by indicating that no more values are available.+--+--   In a mode where either side can return a value, no special termination logic is invoked,+--   execution ends the first time a side returns.+--+--   Left termination is not provided; since there is no way to indicate to the left side+--   that the right has terminated, and potential solutions involve discarding values.+--   However, the "either" modes are also suitable for left termination.+--++module Piped.Compose+  (+  -- ** Operators+  --+    (.|)+  , (|.)++  -- ** Demand driven+  --+  -- | The right hand side is run first. The left hand side is only+  --   invoked by calling `await`.+  --+  , composeDemand+  , composeDemandEither+  +  -- ** Supply driven+  --+  -- | The left hand side is run first. If it returns immediately, the right+  --   hand side is invoked only in the case of `composeSupply`.+  --   +  , composeSupply+  , composeSupplyEither+  ) where++import Piped.Internal+++-- | Demand driven; same as 'composeDemand+--+(.|) :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b+(.|) = composeDemand+{-# INLINE (.|) #-}+++-- | Supply driven; same as 'composeSupplyEither+--+(|.) :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a+(|.) = composeSupplyEither+{-# INLINE (|.) #-}+++-- | The right side is run first, only the right side may return a value.+--+composeDemand :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b+composeDemand (Pipe f1) (Pipe f2) =+  Pipe $ \rest l r ->+    f2 (\_ -> rest termLeft) (Await $ f1 (\_ r _ -> terminate r) l) r+{-# INLINE composeDemand #-}+++-- | The right side is run first, either side may return a value.+--+composeDemandEither :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a+composeDemandEither (Pipe f1) (Pipe f2) =+  Pipe $ \rest l r ->+    f2 (\_ -> rest termLeft) (Await $ f1 (\l -> rest l . termRight) l) r+{-# INLINE composeDemandEither #-}+++-- | The left side is run first, only the right side may return a value.+--+composeSupply :: Monad m => Pipe i e m () -> Pipe e o m b -> Pipe i o m b+composeSupply (Pipe f1) (Pipe f2) =+  Pipe $ \rest l r ->+    f1 (\_ r _ -> terminate r) l $+      Yield+        (           f2 (\_ -> rest termLeft) termLeft             r)+        (\i left -> f2 (\_ -> rest termLeft) (addLeftover i left) r)+{-# INLINE composeSupply #-}+++-- | The left side is run first, either side may return a value.+--+composeSupplyEither :: Monad m => Pipe i e m a -> Pipe e o m a -> Pipe i o m a+composeSupplyEither (Pipe f1) (Pipe f2) =+  Pipe $ \rest l r ->+    f1 (\l r -> rest l $ termRight r) l $+      Yield+        (           f2 (\_ -> rest termLeft) termLeft             r)+        (\i left -> f2 (\_ -> rest termLeft) (addLeftover i left) r)+{-# INLINE composeSupplyEither #-}
+ src/Piped/Extras.hs view
@@ -0,0 +1,84 @@+-- | Convenience methods, plus 'bracketPipe'.+--++module Piped.Extras where++import Data.Function++import Piped.Internal+++-- | Yield all values from a foldable data structure such as a list.+--+sourceList :: (Monad m, Foldable t) => t o -> Pipe () o m ()+sourceList = foldMap yield+{-# INLINE sourceList #-}++-- | Return all input values as a list.+--+sinkList :: Pipe i Void m [i]+sinkList = Pipe $+  \rest l r ->+    fix2 l id $+      \go l fxs ->+        let term = rest termLeft r $ fxs []+         in runAwait l term $ \i l -> go l $ fxs . (i:)+{-# INLINE sinkList #-}++-- | Consume all values using given function+--+awaitForever :: Monad m => (i -> Pipe i o m a) -> Pipe i o m ()+awaitForever f = Pipe $+  \rest ->+    fix $+      \next await yield -> +        runAwait await (rest termLeft yield ()) $+          \i await ->+            unPipe (f i) (\l r _ -> next l r) await yield+{-# INLINE awaitForever #-}++-- | Discard all input values+--+sinkNull :: Monad m => Pipe i Void m ()+sinkNull = Pipe $+  \rest l r -> do+    let term = rest termLeft r ()+        f l = runAwait l term $ \_ -> f+     in f l+{-# INLINE sinkNull #-}++-- | Basically `await >>= maybe ...`, but avoids the conversion to a maybe type.+--+awaitMaybe :: Pipe i o m a -> (i -> Pipe i o m a) -> Pipe i o m a+awaitMaybe def act = Pipe $+  \rest l r ->+    runAwait l (unPipe def rest termLeft r)+               (\i l -> unPipe (act i) rest l r)+{-# INLINE awaitMaybe #-}+++-- | Act on a value, if there is no value then return ().+--+awaitJust :: Monad m => (i -> Pipe i o m ()) -> Pipe i o m ()+awaitJust act = Pipe $+  \rest l r ->+    runAwait l (rest termLeft r ()) $+      \i l -> unPipe (act i) rest l r+{-# INLINE awaitJust #-}+++-- | Yield the result of a monadic action.+--+yieldM :: Monad m => Pipe i o m o -> Pipe i o m ()+yieldM = (>>= yield)+{-# INLINE yieldM #-}+++-- | This is the same as [bracketP](http://hackage.haskell.org/package/conduit-1.3.1.1/docs/Data-Conduit.html#v:bracketP) from Conduit, however it's not specialised to MonadResource.+--+bracketPipe :: Monad m => m b -> (b -> m ()) -> (b -> Pipe i o m a) -> Pipe i o m a+bracketPipe alloc free inside = Pipe $+  \rest l r -> do+    b <- alloc+    let rest' l r a = free b >> rest l r a+    unPipe (inside b) rest' l r
+ src/Piped/Internal.hs view
@@ -0,0 +1,189 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE UndecidableInstances #-}++-- | The internal API and core datatypes of Pipe.+--+module Piped.Internal+  (+  -- ** Pipe+  --+  -- | Piped is implemented using a type of codensity transform, that uses left and right continuations, rather than sum types, to propagate state transitions.+  --+    Pipe(..)+  , await+  , yield+  , runPipe+  , leftover++  -- ** Continuations+  --+  -- | The state of a pipeline is encoded in a recursive data structure (Await / Yield)+  --   which holds continuations that promise a final return value of @r@.+  --+  , Await(..)+  , Yield(..)+  , Await'+  , Yield'+  , runAwait+  , runYield+  , termLeft+  , termRight+  , voidRight+  , addLeftover+  , Void++  -- ** Miscellania+  --+  , fix1+  , fix2+  ) where+++import Data.Void++import Control.Monad+import Control.Monad.Reader+import Control.Monad.State+++-- | The upstream continuation; holds a callback that provides a value.+--+newtype Await i m a = Await { unAwait :: Await' i m a }+++-- | The downstream continuation; holds a callback that accepts a value,+--   plus a default return value in case there is no more input.+--+data Yield i m a = Yield+  { terminate :: m a+  , unYield   :: Yield' i m a+  }++-- | The type of an upstream continuation+--+type Await' i m a = Yield i m a -> m a++-- | The type of a downstream continuation+--+type Yield' i m a = i -> Await i m a -> m a++-- | Run an Await+--+runAwait :: Await i m a -> m a -> Yield' i m a -> m a+runAwait (Await awt) a yld = awt $ Yield a yld+{-# INLINE runAwait #-}++-- | Run a Yield+--+runYield :: Yield i m a -> i -> Await' i m a -> m a+runYield (Yield _ a) i = a i . Await+{-# INLINE runYield #-}++-- | An Await that terminates when called.+--+termLeft :: Await i m a+termLeft = Await terminate+{-# INLINE termLeft #-}++-- | A Yield that terminates when called.+--+termRight :: Yield i1 m a -> Yield i2 m a+termRight r = Yield (terminate r) (\_ _ -> terminate r)+{-# INLINE termRight #-}++-- | A Void output+--+voidRight :: Yield Void m a+voidRight = Yield (error "Void") (\i _ -> absurd i)+{-# INLINE voidRight #-}++-- | Wrap an Await with a leftover value+--+addLeftover :: i -> Await i m a -> Await i m a+addLeftover i await = Await $ \y -> unYield y i await+++-- | The Pipe datatype, that represents a stage in a pipeline with inputs of type @i@ and outputs of type @o@.+--+newtype Pipe i o m a =+  Pipe { unPipe :: forall r. (Await i m r -> Yield o m r -> a -> m r) -> Await i m r -> Yield o m r -> m r }++instance Monad m => Functor (Pipe i o m) where+  fmap f (Pipe p) = Pipe $+    \rest l r -> p (\l r -> rest l r . f) l r++instance Monad m => Applicative (Pipe i o m) where+  pure = return+  (<*>) = ap++instance Monad m => Monad (Pipe i o m) where+  return x = Pipe $ \f l r -> f l r x+  Pipe f >>= g = Pipe $+    \rest -> f (\l r a -> unPipe (g a) rest l r)++instance MonadTrans (Pipe i o) where+  lift mf = Pipe $ \f l r -> mf >>= f l r++instance MonadIO m => MonadIO (Pipe i o m) where+  liftIO = lift . liftIO++instance MonadReader r m => MonadReader r (Pipe i o m) where+  ask = lift ask+  local f (Pipe p) = Pipe $ \rest l r -> local f $ p rest l r++instance MonadState s m => MonadState s (Pipe i o m) where+  get = lift get+  put = lift . put++instance Monad m => Semigroup (Pipe i o m a) where+  (<>) = (>>)++instance Monad m => Monoid (Pipe i o m ()) where+  mempty = pure ()+++-- | Await a value. Nothing indicates that there are no more values.+--+await :: Monad m => Pipe i o m (Maybe i)+await = Pipe $+  \rest l r ->+    let term = rest termLeft r Nothing+     in runAwait l term $ \i l -> rest l r $ Just i+{-# INLINE await #-}+++-- | Yield a value to downstream.+--+yield :: o -> Pipe i o m ()+yield i = Pipe $+  \rest a y -> runYield y i $ \y -> rest a y ()+{-# INLINE yield #-}+++-- | Run pipe to completion.+--+runPipe :: Monad m => Pipe () Void m r -> m r+runPipe pipe = unPipe pipe (\_ _ -> pure) termLeft voidRight+{-# INLINE runPipe #-}+++-- | Push a value back into the incoming pipeline+--+leftover :: i -> Pipe i o m ()+leftover i = Pipe $+  \rest l r -> rest (addLeftover i l) r ()+{-# INLINE leftover #-}+++-- `fix` providing one value.+fix1 :: a -> ((a -> b) -> a -> b) -> b+fix1 a f = fix f a+{-# INLINE fix1 #-}+++-- `fix` providing two values.+fix2 :: a -> b -> ((a -> b -> c) -> a -> b -> c) -> c+fix2 a b f = fix f a b+{-# INLINE fix2 #-}
+ src/Piped/Prelude.hs view
@@ -0,0 +1,142 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE BangPatterns #-}++-- | Basic functionality, closely mirroring the list / Foldable api from Prelude.+--+-- This module should be imported qualified i.e. `import Piped.Prelude as P`.+--+module Piped.Prelude where++import Control.Monad+import Control.Monad.Trans++import Prelude hiding (foldl, scanl, mapM_, last, take, drop, zipWith)++import Piped+import Piped.Internal+++-- | Yield only elements satisfying a predicate.+--+filter :: Monad m => (i -> Bool) -> Pipe i i m ()+filter f = awaitForever $ \i -> if f i then yield i else pure ()+{-# INLINE filter #-}++-- | Yield values while they satisfy a predicate, then return.+--+takeWhile :: Monad m => (i -> Bool) -> Pipe i i m ()+takeWhile f = go where+  go = awaitJust $ \i -> if f i then yield i >> go else leftover i+{-# INLINE takeWhile #-}++-- | Drop values while they satisfy a predicate, then return.+--+--   Note: this does not yield any values, and should be combined with (>>).+--+dropWhile :: Monad m => (i -> Bool) -> Pipe i o m ()+dropWhile f = go where+  go = awaitJust $ \i -> if f i then go else leftover i+{-# INLINE dropWhile #-}++-- | Equivalent to `map id`.+--+identity :: Monad m => Pipe i i m ()+identity = awaitForever yield+{-# INLINE identity #-}++-- | `map` with an accumulator.+--+scanl :: forall i o m. (o -> i -> o) -> o -> Pipe i o m ()+scanl f s = Pipe $+  \rest ->+    fix1 s $+      \next !s l r ->+        runYield r s $ \r -> +          runAwait l (rest l r ()) $+            \i l -> next (f s i) l r+{-# INLINE scanl #-}++-- | Left fold over input values.+--+foldl :: Monad m => (a -> i -> a) -> a -> Pipe i o m a+foldl f start = Pipe $+  \rest ->+    fix1 start $+      \next !s l r ->+        runAwait l (rest termLeft r s) $+          \i l -> next (f s i) l r+{-# INLINE foldl #-}++-- | Drop n values.+--+--   Note: This will not yield values and should be combined with `>>`+--+drop :: Monad m => Int -> Pipe i o m ()+drop 0 = pure ()+drop n = awaitJust (\_ -> drop $ n-1)+{-# INLINE drop #-}++-- | Take n values.+--+take :: Monad m => Int -> Pipe i i m ()+take 0 = pure ()+take n = awaitJust $ \i -> yield i >> take (n-1)+{-# INLINE take #-}++-- | Map a pure function over input values.+--+map :: Monad m => (i -> o) -> Pipe i o m ()+map f = awaitForever $ yield . f+{-# INLINE map #-}++-- | Map a monadic function over input values.+--+mapM :: Monad m => (i -> m o) -> Pipe i o m ()+mapM f = awaitForever $ \i -> lift (f i) >>= yield+{-# INLINE mapM #-}++-- | Map a monadic function over input values but don't yield anything.+--+mapM_ :: Monad m => (i -> m ()) -> Pipe i () m ()+mapM_ f = awaitForever $ lift . f+{-# INLINE mapM_ #-}++-- | Skip the first value+--+tail :: Monad m => Pipe i i m ()+tail = await >> identity+{-# INLINE tail #-}++-- | Return the last value.+--+last :: Monad m => Pipe i o m (Maybe i)+last = go Nothing where+  go x = awaitMaybe (pure x) (go . Just)+{-# INLINE last #-}++-- | Zip two pipes together.+--+zip :: Monad m => Pipe () o m () -> Pipe () o' m () -> Pipe () (o, o') m ()+zip = zipWith (,)+{-# INLINE zip #-}++-- | Zip two pipes together with a pure function.+--+zipWith :: Monad m => (o -> o' -> a) -> Pipe () o m () -> Pipe () o' m () -> Pipe () a m ()+zipWith f (Pipe f1) (Pipe f2) = Pipe go where+  go rest _ r = loop (side f1) (side f2) r+    where+      exit y = rest termLeft y ()+      side f = Await $ \y -> f (\_ y () -> terminate y) termLeft y+      loop a1 a2 yield =+        runAwait a1 (exit yield) $ \i1 a1 ->+          runAwait a2 (exit yield) $ \i2 a2 ->+            runYield yield (f i1 i2) $ \yield ->+              loop a1 a2 yield++-- | Concatenate foldable inputs to a single stream of outputs.+--+concat :: (Foldable t, Monad m) => Pipe (t i) i m ()+concat = awaitForever $ foldMap yield+{-# INLINE concat #-}
+ src/Piped/Resume.hs view
@@ -0,0 +1,62 @@++module Piped.Resume+  ( createResumableSource+  , createResumableSink+  , runResumableSource+  , runResumableSink+  , ResumableSource(..)+  , ResumableSink(..)+  , ResumableResult(..)+  ) where+++import Piped.Internal+++-- | Either a resumable source or sink, plus the result of the pipe that finished.+--+data ResumableResult e m a b =+    ResumeSource (ResumableSource e m a b) b+  | ResumeSink   (ResumableSink   e m a b) a+++-- | A source that may be resumed+--+newtype ResumableSource o m a b = ResumableSource (Await o m (ResumableResult o m a b))+++-- | A sink that may be resumed+--+newtype ResumableSink   i m a b = ResumableSink   (Yield i m (ResumableResult i m a b))+++-- | Create a resumable source from a Pipe+--+createResumableSource :: Monad m => Pipe () o m a -> ResumableSource o m a b+createResumableSource (Pipe f) = ResumableSource $+  Await $ f (\l r a -> pure $ ResumeSink (ResumableSink r) a) termLeft+++-- | Create a resumable sink from a Pipe+--+createResumableSink :: Monad m => Pipe i Void m b -> ResumableSink i m a b+createResumableSink (Pipe f) = ResumableSink $+   Yield+     (           f resumeLeft termLeft             voidRight)+     (\i left -> f resumeLeft (addLeftover i left) voidRight)+  where+    resumeLeft l _ b  = pure $ ResumeSource (ResumableSource l) b+++-- | Run a resumable source+--+runResumableSource :: Monad m => ResumableSource i m a b -> Pipe i Void m b -> m (ResumableResult i m a b)+runResumableSource (ResumableSource source) (Pipe f) =+  f (\l _ b -> pure $ ResumeSource (ResumableSource l) b) source voidRight+++-- | Run a resumable sink+--+runResumableSink :: Monad m => Pipe () o m a -> ResumableSink o m a b -> m (ResumableResult o m a b)+runResumableSink (Pipe f) (ResumableSink sink) =+  f (\_ r a -> pure $ ResumeSink (ResumableSink r) a) termLeft sink
+ test-bench/Bench.hs view
@@ -0,0 +1,52 @@++module Main where++import BenchShow+import Gauge.Main+import Control.Exception+import Data.List (sortOn)+import Data.List.Split+import System.Environment (getArgs)++import BenchYields+import BenchPipes+import BenchCompare+++main = do+  args <- getArgs+  if take 1 args == ["charts"]+     then renderCharts $ args !! 1+     else do+       defaultMain+        [ benchPipes+        , benchYields+        , benchCompare+        ]+++renderCharts file = do+  stdgraph "mixed_naive" $ drop 1+  stdgraph "n_stages" $ drop 1+  stdgraph "mixed_optimised" $ drop 1+  +  graph file "yielding_vs_monadic" $+    defaultConfig+      { classifyBenchmark = \name ->+          case splitOn "/" name of+            ["pipes", b, c] -> Just (c, b)+            _               -> Nothing+      }++  where+    stdgraph name f =+      handle (\e -> print (e::SomeException)) $+        graph file name $+          defaultConfig+            { classifyBenchmark = \s ->+                case f (splitOn "/" s) of+                  [a, b, c] | a == name -> Just (c, b)+                  _                     -> Nothing+            , title = Just name+            , selectGroups = sortOn $ (/="piped") . fst+            }
+ test-bench/BenchPipes.hs view
@@ -0,0 +1,149 @@+{-# LANGUAGE BangPatterns #-}++module BenchPipes where++import Data.Function (fix)++import Control.Monad++import Piped+import Piped.Internal+import Piped.Prelude as P++import Gauge.Main++import Prelude hiding (scanl, foldl, dropWhile, takeWhile)+++benchPipes = bgroup "pipes"+  [ bgroup "sinkNull"  [ bench "monadic" $ whnfAppIO (runSink sinkNullM) 1000+                       , bench "yielding" $ whnfAppIO (runSink sinkNull) 1000+                       ]+  , bgroup "sinkList"  [ bench "monadic" $ nfAppIO (runSink sinkListM) 1000+                       , bench "yielding" $ nfAppIO (runSink sinkList) 1000+                       ]++  , bgroup "foldl"     [ bench "monadic" $ nfAppIO bFoldlM 10000+                       , bench "yielding" $ nfAppIO bFoldlY 10000+                       ]++  , bgroup "scanl"     [ bench "monadic" $ nfAppIO bScanlM 10000+                       , bench "yielding" $ nfAppIO bScanlY 10000+                       ]++  , bgroup "dropWhile" [ bench "monadic" $ nfAppIO (bDoWhile dropWhile) 10000+                       , bench "yielding" $ nfAppIO (bDoWhile dropWhileY) 10000+                       ]++  , bgroup "takeWhile" [ bench "monadic" $ nfAppIO (bDoWhile takeWhile) 10000+                       , bench "yielding" $ nfAppIO (bDoWhile takeWhileY) 10000+                       ]+  , bgroup "identity"  [ bench "monadic" $ nfAppIO (bIdentity identity) 1000+                       , bench "yielding" $ nfAppIO (bIdentity identityY) 1000+                       ]+  -- , bgroup "maybes"    [ bench "awaitMaybe" $ nfAppIO (bMaybes $ awaitMaybe mempty) 10000+  --                      , bench "awaitJust" $ nfAppIO (bMaybes $ awaitJust) 10000+  --                      ]+  , bgroup "map"       [ bench "monadic" $ nfAppIO (bSimple $ P.map (+1)) 10000+                       , bench "yielding" $ nfAppIO (bSimple $ pmapMon (+1)) 10000+                       ]+  , bgroup "filter"    [ bench "monadic" $ nfAppIO (bSimple $ P.filter even) 10000+                       , bench "yielding" $ nfAppIO (bSimple $ pFilter even) 10000+                       ]+  ]++source :: Int -> Pipe () Int IO ()+source n = Prelude.mapM_ yield [0..n]++runSink :: Pipe Int Void IO a -> Int -> IO a+runSink sink n =+  runPipe $ source n .| sink++sinkNullM :: Monad m => Pipe i Void m ()+sinkNullM = awaitForever (\_ -> pure ())++sinkListM :: Monad m => Pipe i Void m [i]+sinkListM = let f = awaitMaybe (pure []) (\i -> (i:) <$> f) in f+{-# INLINE sinkListM #-}++bFoldl p n = runPipe $ source n .| (p (+) 0 >> pure ())+bFoldlM n = runPipe $ source n .| (foldl' (+) 0 >> pure ())+bFoldlY n = runPipe $ source n .| (foldl (+) 0 >> pure ())++foldl' :: Monad m => (a -> i -> a) -> a -> Pipe i o m a+foldl' f s =+  let next !s = awaitMaybe (pure s) (next . f s) in next s+{-# INLINE foldl' #-}+-- foldl' :: Monad m => (a -> i -> a) -> a -> Pipe i o m a+-- foldl' f start = Pipe $+--   \rest ->+--     fix1 start $+--       \next !s l r ->+--         runAwait l (rest termLeft r s) $+--           \i l -> next (f s i) l r+-- {-# INLINE foldl' #-}++bScanlM = runSink $ scanlM (+) 0 .| sinkNull+bScanlY = runSink $ scanl (+) 0 .| sinkNull++scanlM :: Monad m => (a -> b -> a) -> a -> Pipe b a m ()+scanlM f =+  fix $ \next !s ->+    yield s >> awaitJust (\i -> next $ f s i)+{-# INLINE scanlM #-}++bDoWhile p n = runPipe $ source n .| p (\_ -> True) .| sinkNull++dropWhileY :: Monad m => (i -> Bool) -> Pipe i i m ()+dropWhileY f = Pipe $+  \rest ->+    fix $ \next l r ->+      runAwait l (rest termLeft r ()) $ \i l ->+        if f i+           then next l r+           else let l' = addLeftover i l+                 in unPipe identity rest l' r+{-# INLINE dropWhileY #-}++takeWhileY :: Monad m => (i -> Bool) -> Pipe i i m ()+takeWhileY f = Pipe $+  \rest ->+    fix $ \next l r ->+      runAwait l (rest termLeft r ()) $ \i l ->+        if f i+           then runYield r i $ next l+           else rest l r ()+{-# INLINE takeWhileY #-}++bIdentity :: Pipe Int Int IO () -> Int -> IO ()+bIdentity p n = runPipe $ source n .| ((iterate (.| p) p) !! 10) .| sinkNull++identityY :: Pipe i i m ()+identityY = Pipe $+  \rest ->+    fix $ \next l r ->+      runAwait l (rest termLeft r ()) $ \i l ->+        runYield r i $ next l+{-# INLINE identityY #-}++bMaybes p n = runPipe $ source n .| go+  where go = p $ (\_ -> go :: Pipe Int Void IO ())++bSimple p n = runPipe $ source n .| p .| sinkNull++pFilter :: (i -> Bool) -> Pipe i i m ()+pFilter f = Pipe $+  \rest -> fix $+    \next l r ->+      runAwait l (rest termLeft r ()) $+        \i l -> if f i+                   then runYield r i $ \r -> next l r+                   else next l r+{-# INLINE pFilter #-}++pmapMon :: (i -> o) -> Pipe i o m ()+pmapMon f = Pipe $+  \rest -> fix $+    \next l r ->+      runAwait l (rest termLeft r ()) $+        \i l -> runYield r (f i) $ \r -> next l r
+ test-bench/BenchYields.hs view
@@ -0,0 +1,114 @@++module BenchYields where++import Gauge.Main++-- Benchmark different data structures for continuation passing+--+-- The requirement is for two coroutines to pass continuations+-- back and forth, so a recursive data structure is required.+-- It needs to be able to pass a value as well as the+-- contination in one direction, and indicate that no further+-- value is available in the other.+--+-- Some results:+--+-- cont/maybe                               time                 16.66 μs+-- cont/term                                time                 14.80 μs+-- cont/stall                               time                 13.88 μs+-- cont/term4                               time                 21.43 μs+++benchYields =+  bgroup "cont" [ bench "maybe"  $ whnf runMaybes 1000+                , bench "term"   $ whnf runTerm 1000+                , bench "stall"  $ whnf runStall 1000+                , bench "term4"  $ whnf runTerm4 1000+                ]+++-- The first data structure wraps the data in a Maybe in order to indicate+-- when no more data is available.++newtype Await i a = Await { unAwait :: Yield i a -> a }+newtype Yield i a = Yield { unYield :: Maybe i -> Await i a -> a }++runMaybes n = unAwait (left n) right+  where+    left n =+      Await $ \yield -> do+        let v = if n == 0 then Nothing else Just ()+         in unYield yield v $ left (n-1)+      +    right = Yield $ \mi await -> maybe () (\_ -> unAwait await right) mi+++--------------------------++-- This is the preferred version; the yield function wrapper also+-- contains a thunk which returns an output directly. It's faster+-- than the above and can also express the Maybe interface.++newtype Await2 i a = Await2 { unAwait2 :: Await2' i a }++data Yield2 i a = Yield2+  { terminate :: a+  , unYield2 :: Yield2' i a+  }++type Await2' i a = Yield2 i a -> a+type Yield2' i a = i -> Await2 i a -> a++runTerm n = unAwait2 (left n) right+  where+    left n =+      Await2 $ \yield ->+        if n == 0 then terminate yield+                  else unYield2 yield () $ left (n-1)+      +    right = Yield2 () $ \() await -> unAwait2 await right+++--------------------------++-- This is slightly faster than the previous one, but since it does not allow+-- to specify termination, any node can terminate and that's not always desirable.++newtype Await3 i a = Await3 { unAwait3 :: Yield3 i a -> a }+newtype Yield3 i a = Yield3 { unYield3 :: i -> Await3 i a -> a }++runStall n = unAwait3 (left n) right+  where+    left n =+      Await3 $ \yield -> do+        if n == 0 then ()+                  else unYield3 yield () $ left (n-1)+      +    right = Yield3 $ \() await -> unAwait3 await right+++--------------------------++-- This one is just slow, its the same concept as Yield2 but uses an extra +-- function to select between yielding and termination, to avoid the `data`+-- record, but actually the `data` record hardly incurs any penalty as it turns out.++newtype Await4 i a = Await4 { unAwait4 :: Await4' i a }++newtype Yield4 i a = Yield4+  { unYield4 :: Yield4' i a+  }++type Await4' i a = Yield4 i a -> a+type Yield4' i a = (a -> (i -> Await4 i a -> a) -> a) -> a++runTerm4 n = unAwait4 (left n) right+  where+    left n =+      Await4 $ \yield ->+        unYield4 yield $+          \term onval -> +            if n == 0 then term else onval () (left (n-1))+      +    right = Yield4 $ \f -> f () $ \() await -> unAwait4 await right+
+ test-common/BenchCompare.hs view
@@ -0,0 +1,150 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}++module BenchCompare where++import Control.Monad+import Data.Function+import Data.Void+import Prelude hiding (map, foldl, takeWhile, dropWhile, scanl, filter)++import Gauge.Main++import qualified Conduit as C+import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.List as CL+import qualified Piped as P+import qualified Piped.Prelude as P+++benchCompare = bgroup "compare"+  [ bgroup "n_stages" $+      [ compare (show (n+2), p_stages n, c_stages n) | n <- [0, 2, 4, 6]+      ]+  , bgroup  "mixed_optimised" $+      [ compare ("map",           p_map,           c_map)+      , compare ("foldl",         p_foldl,         c_foldl)+      , compare ("takeall", p_takeall, c_takeall)+      , compare ("dropall", p_dropall, c_dropall)+      , compare ("scanl",   p_scanl,   c_scanl)+      , compare ("filter_all",    p_filter_all,    c_filter_all)+      , compare ("filter_none",   p_filter_none,   c_filter_none)+      , compare ("sinklist",      p_sinklist,      c_sinklist)+      , compare ("sinknull",      p_sinknull,      c_sinknull)+      ]+  , bgroup "mixed_naive" $+      [ compare ("map",           p_n_map,           c_n_map)+      , compare ("foldl",         p_n_foldl,         c_n_foldl)+      , compare ("takeall", p_n_takeall, c_n_takeall)+      , compare ("dropall", p_n_dropall, c_n_dropall)+      , compare ("scanl",   p_n_scanl,   c_n_scanl)+      , compare ("filter_all",    p_n_filter_all,    c_n_filter_all)+      , compare ("filter_none",   p_n_filter_none,   c_n_filter_none)+      , compare ("sinklist",      p_n_sinklist,      c_n_sinklist)+      , compare ("sinknull",      p_n_sinknull,      c_n_sinknull)+      ]+  ]++  where+    iter = 100000+    runC :: (Integer -> C.ConduitT () o IO a) -> Integer -> IO ()+    runC c n = C.runConduit $ (c n >> pure ()) C..| C.sinkNull+    runP :: (Integer -> P.Pipe () o IO a) -> Integer -> IO ()+    runP p n = P.runPipe    $ (p n >> pure ()) P..| P.sinkNull+    compare :: (String, Integer -> P.Pipe () o IO a, Integer -> C.ConduitT () o IO a) -> Benchmark+    compare (name, p, c) = bgroup name $+      [ bench "conduit" $ whnfAppIO (runC c) iter+      , bench "piped"   $ whnfAppIO (runP p) iter+      ]+++c_stages s n = source_C n C..| go s C..| C.sinkNull where+  go 0 = ident_C+  go s = ident_C C..| go (s-1)++p_stages s n = source_P n P..| go s P..| P.sinkNull where+  go 0 = ident_P+  go s = ident_P P..| go (s-1)+++c_foldl, c_n_foldl :: Integer -> C.ConduitT () () IO Integer+p_foldl, p_n_foldl :: Integer -> P.Pipe () () IO Integer+c_sinklist, c_n_sinklist :: Integer -> C.ConduitT () Void IO [Integer]+p_sinklist, p_n_sinklist :: Integer -> P.Pipe () Void IO [Integer]++-- Optimised++c_map           n = CL.sourceList [0..n] C..| C.map (+1)+c_foldl         n = CL.sourceList [0..n] C..| C.foldl (+) 0+c_takeall       n = CL.sourceList [0..n] C..| C.takeWhile (<n)+c_dropall       n = CL.sourceList [0..n] C..| (C.dropWhile (<n) >> ident_C)+c_scanl         n = CL.sourceList [0..n] C..| C.scanl (+) 0+c_filter_all    n = CL.sourceList [0..n] C..| C.filter (<0)+c_filter_none   n = CL.sourceList [0..n] C..| C.filter (>=0)+c_sinklist      n = CL.sourceList [0..quot n 2] C..| C.sinkList+c_sinknull      n = CL.sourceList [0..n]++p_map           n = P.sourceList [0..n] P..| P.map (+1)+p_foldl         n = P.sourceList [0..n] P..| P.foldl (+) 0+p_takeall       n = P.sourceList [0..n] P..| P.takeWhile (<n)+p_dropall       n = P.sourceList [0..n] P..| (P.dropWhile (<n) >> ident_P)+p_scanl         n = P.sourceList [0..n] P..| P.scanl (+) 0+p_filter_all    n = P.sourceList [0..n] P..| P.filter (<0)+p_filter_none   n = P.sourceList [0..n] P..| P.filter (>=0)+p_sinklist      n = P.sourceList [0..quot n 2] P..| P.sinkList+p_sinknull      n = P.sourceList [0..n]++-- Naive++c_n_map           n = source_C n C..| map_C (+1)+c_n_foldl         n = source_C n C..| foldl_C (+) 0+c_n_takeall       n = source_C n C..| takeWhile_C (<n)+c_n_dropall       n = source_C n C..| (dropWhile_C (<n) >> ident_C)+c_n_scanl         n = source_C n C..| scanl_C (+) 0+c_n_filter_all    n = source_C n C..| filter_C (<0)+c_n_filter_none   n = source_C n C..| filter_C (>=0)+c_n_sinklist      n = source_C (quot n 2) C..| sinkList_C+c_n_sinknull      n = source_C n++p_n_map           n = source_P n P..| map_P (+1)+p_n_foldl         n = source_P n P..| foldl_P (+) 0+p_n_takeall       n = source_P n P..| takeWhile_P (<n)+p_n_dropall       n = source_P n P..| (dropWhile_P (<n) >> ident_P)+p_n_scanl         n = source_P n P..| scanl_P (+) 0+p_n_filter_all    n = source_P n P..| filter_P (<0)+p_n_filter_none   n = source_P n P..| filter_P (>=0)+p_n_sinklist      n = source_P (quot n 2) P..| sinkList_P+p_n_sinknull      n = source_P n++++ident_C = C.awaitForever C.yield+source_C n = CL.sourceList [0..n]+map_C f = C.awaitForever $ C.yield . f+awaitMaybe_C d f = C.await >>= maybe d f+foldl_C f = go where+  go (!s) = awaitMaybe_C (pure s) (\i -> go (f s i))+takeWhile_C f = go where+  go = awaitMaybe_C (pure ()) (\i -> if f i then C.yield i >> go else pure ())+dropWhile_C f = go where+  go = awaitMaybe_C (pure ()) (\i -> if f i then go else C.yield i >> ident_C)+scanl_C f = go where+  go (!s) = C.yield s >> awaitMaybe_C (pure ()) (\i -> go $ f s i)+filter_C f = C.awaitForever $ \i -> if f i then C.yield i else pure ()+sinkList_C = awaitMaybe_C (pure []) (\i -> (i:) <$> sinkList_C)+++ident_P = P.awaitForever P.yield+source_P n = P.sourceList [0..n]+map_P f = P.awaitForever $ P.yield . f+awaitMaybe_P d f = P.await >>= maybe d f+foldl_P f = go where+  go (!s) = awaitMaybe_P (pure s) (\i -> go (f s i))+takeWhile_P f = go where+  go = awaitMaybe_P (pure ()) (\i -> if f i then P.yield i >> go else pure ())+dropWhile_P f = go where+  go = awaitMaybe_P (pure ()) (\i -> if f i then go else P.yield i >> ident_P)+scanl_P f = go where+  go (!s) = P.yield s >> awaitMaybe_P (pure ()) (\i -> go $ f s i)+filter_P f = P.awaitForever $ \i -> if f i then P.yield i else pure ()+sinkList_P = awaitMaybe_P (pure []) (\i -> (i:) <$> sinkList_P)
+ test/BenchSpec.hs view
@@ -0,0 +1,49 @@++module BenchSpec where++import Data.Conduit as C+import Data.Conduit.Combinators as CL+import Piped as P++import Test.Tasty+import Test.Tasty.HUnit++import BenchCompare+++test_bench :: TestTree+test_bench = testGroup "Verify that benchmark pipes output same results" $+  [ testGroup "n_stages" $+      [ -- compare (show (n+2), p_stages n, c_stages n) | n <- [0, 2, 4, 6]+      ]+  , testGroup  "mixed_optimised" $+      [ compare ("map",           p_map,           c_map)+      , compare ("foldl",         p_foldl,         c_foldl)+      , compare ("takeall", p_takeall, c_takeall)+      , compare ("dropall", p_dropall, c_dropall)+      , compare ("scanl",   p_scanl,   c_scanl)+      , compare ("filter_all",    p_filter_all,    c_filter_all)+      , compare ("filter_none",   p_filter_none,   c_filter_none)+      , compare ("sinklist",      p_sinklist,      c_sinklist)+      , compare ("sinknull",      p_sinknull,      c_sinknull)+      ]+  , testGroup "mixed_naive" $+      [ compare ("map",           p_n_map,           c_n_map)+      , compare ("foldl",         p_n_foldl,         c_n_foldl)+      , compare ("takeall", p_n_takeall, c_n_takeall)+      , compare ("dropall", p_n_dropall, c_n_dropall)+      , compare ("scanl",   p_n_scanl,   c_n_scanl)+      , compare ("filter_all",    p_n_filter_all,    c_n_filter_all)+      , compare ("filter_none",   p_n_filter_none,   c_n_filter_none)+      , compare ("sinklist",      p_n_sinklist,      c_n_sinklist)+      , compare ("sinknull",      p_n_sinknull,      c_n_sinknull)+      ]+  ]+  where+    n = 100++    compare :: (Eq o, Show o) => (String, Integer -> Pipe () o IO a, Integer -> ConduitT () o IO a) -> TestTree+    compare (s, p, c) = testCase s $ do+      cr <- runConduit ((c n >> pure ()) C..| CL.sinkList)+      pr <- runPipe    ((p n >> pure ()) P..| P.sinkList)+      pr @?= cr
+ test/ComposeSpec.hs view
@@ -0,0 +1,31 @@++module ComposeSpec where++import Piped+import Piped.Compose++import TestUtils+++test_4 = testCase "composeSupply init correct order" $ do+  let o = runPipeS [] $ composeSupply (alloc 1 >> yield 1) (alloc 2 >> await)+  o @?= (Just 1, [1, 2])++test_5 = testCase "composeSupplyEither left run first" $ do+  let o = runPipeI $ composeSupplyEither (pure 1) (pure 2)+  o @?= 1++test_55 = testCase "composeSupplyEither right can return" $ do+  let o = runPipeI $ composeSupplyEither (yield () >> pure 1) (pure (2::Int))+  o @?= 2++test_6 = testCase "composeDemand left doesnt run" $+  runPipe $ composeDemand (error "failure") (pure ())++test_8 = testCase "composeDemandEither right run first" $ do+  let o = runPipeI $ composeDemandEither (pure 1) (pure 2)+  o @?= 2++test_88 = testCase "composeDemandEither left can return" $ do+  let o = runPipeI $ composeDemandEither (pure 1) (await >> pure 2)+  o @?= 1
+ test/ConduitSpec.hs view
@@ -0,0 +1,54 @@++module ConduitSpec where++import qualified Data.Conduit as C+import qualified Piped as P+import Control.Monad.State++import Test.Tasty.HUnit++import PipeLike as PL+++-- | Below tests polymorphic pipeline definitions using Piped and Conduit.+--   The application of () is because for some reason GHC will allow +--   polymorphic definitions without signatures if an argument is passed,+--   event a useless one.++test_1 = test "demand driven" p p+  where+    p _ = undefined .| (pure ())++test_2 = test "sinkList correct order" p p+  where+    p _ = mapM_ yield [True, False] .| sinkList++test_3 = test "dropWhile doesn't consume" p p+  where+    p _ = sourceList [1..10] .| (PL.dropWhile (<5) >> leftover (-1) >> sinkList)++test_4 = test "takeWhile doesn't drop values" p p+  where+    p _ = sourceList [1..10] .| (PL.takeWhile (<5) >> PL.map (+1)) .| sinkList++test_44 = test "take doesn't drop values" p p+  where+    p _ = sourceList [1..10] .| PL.take 5 .| sinkList++test_45 = test "drop doesn't yield values" p p+  where+    p _ = sourceList [1..10] .| (PL.drop 5 >> PL.map (+1)) .| sinkList++test_5 = test "recursive pipes don't do funny things" p p+  where+    p _ = go 2+    go 0 = await >> pure True+    go n = yield () .| (await >> go (n-1))++test_6 = test "foldl works as expected" p p+  where+    p _ = sourceList [True, False, True] .| PL.foldl (flip (:)) []+++test s p c = testCompare s (p ()) (c ())+testCompare s c p = testCase s $ join $ (@?=) <$> P.runPipe p <*> C.runConduit c
+ test/Discover.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF tasty-discover -optF --tree-display #-}
+ test/PipeLike.hs view
@@ -0,0 +1,87 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE LambdaCase #-}++module PipeLike+  ( module PipeLike+  , C.ConduitT+  , P.Pipe+  , C.runConduit+  , P.runPipe+  , Void+  ) where++import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.List as C (sourceList)+import qualified Piped as P+import qualified Piped.Prelude as P++import Data.Void+++class (Monad m, Monad (p i o m)) => PipeLike p i o m where+  await :: p i o m (Maybe i)+  yield :: o -> p i o m ()+  (.|) :: p i e m () -> p e o m b -> p i o m b+  infixr 5 .|+  sourceList :: [o] -> p () o m ()+  sinkList :: p i Void m [i]+  foldl :: (a -> i -> a) -> a -> p i o m a+  scanl :: (a -> b -> a) -> a -> p b a m ()+  sinkNull :: p i Void m ()+  awaitForever :: (i -> p i o m ()) -> p i o m ()+  map :: (a -> b) -> p a b m ()+  mapMC :: (i -> m o) -> p i o m ()+  mapMC_ :: (i -> m ()) -> p i () m ()+  take :: Int -> p i i m ()+  drop :: Int -> p i o m ()+  takeWhile :: (i -> Bool) -> p i i m ()+  dropWhile :: Monad m => (i -> Bool) -> p i o m ()+  filter :: (i -> Bool) -> p i i m ()+  leftover :: i -> p i o m ()++instance Monad m => PipeLike C.ConduitT i o m where+  await = C.await+  yield = C.yield+  (.|) = (C..|)+  sourceList = C.sourceList+  sinkList = C.sinkList+  foldl = C.foldl+  scanl = C.scanl+  sinkNull = C.sinkNull+  awaitForever = C.awaitForever+  take = C.take+  drop = C.drop+  takeWhile = C.takeWhile+  map = C.map+  mapMC = C.mapM+  mapMC_ = C.mapM_+  dropWhile = C.dropWhile+  filter = C.filter+  leftover = C.leftover++instance Monad m => PipeLike P.Pipe i o m where+  await = P.await+  yield = P.yield+  (.|) = (P..|)+  sourceList = P.sourceList+  sinkList = P.sinkList+  foldl = P.foldl+  scanl = P.scanl+  sinkNull = P.sinkNull+  awaitForever = P.awaitForever+  take = P.take+  drop = P.drop+  takeWhile = P.takeWhile+  map = P.map+  mapMC = P.mapM+  mapMC_ = P.mapM_+  dropWhile = P.dropWhile+  filter = P.filter+  leftover = P.leftover
+ test/TestUtils.hs view
@@ -0,0 +1,21 @@++module TestUtils+  ( module TestUtils+  , module T+  ) where++import Data.IORef as T+import Test.Tasty as T+import Test.Tasty.HUnit as T++import Control.Monad.State as T+import Control.Monad.Identity as T++import Piped as T+import Piped.Internal as T+++runPipeI = runIdentity . runPipe+runPipeS s p = runState (runPipe p) s+alloc n = modify (++[n])+