conduino 0.2.2.0 → 0.2.3.0

raw patch · 6 files changed

+256/−60 lines, 6 filesdep +textdep ~transformersPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: text

Dependency ranges changed: transformers

API changes (from Hackage documentation)

Files

• CHANGELOG.md +15/−0
• conduino.cabal +2/−1
• src/Data/Conduino.hs +62/−16
• src/Data/Conduino/Combinators.hs +74/−6
• src/Data/Conduino/Internal.hs +63/−10
• src/Data/Conduino/Lift.hs +40/−27

@@ -1,6 +1,21 @@ Changelog ========= +Version 0.2.3.0+---------------++*March 25, 2022*++<https://github.com/mstksg/conduino/releases/tag/v0.2.3.0>++*   Add `sourceHandleLinesText` for `Text` output+*   `sourceHandleLines` now continues through blank lines, but `stdinLines`+    retains the same behavior+*   `passthrough` pipe manipulation+*   More efficient `runExceptP`, `runCatchP`+*   More explicit inlining+*   `yield` is strict by default.  Use `yieldLazy` for original lazy behavior.+ Version 0.2.2.0 --------------- 

@@ -7,7 +7,7 @@ -- hash: 94a4f8415936756af3ddab6ae4a237db4a3f8ee9d4b93ce27fe211cceca6761b  name:           conduino-version:        0.2.2.0+version:        0.2.3.0 synopsis:       Lightweight composable continuation-based stream processors description:    A lightweight continuation-based stream processing library.                 .@@ -53,4 +53,5 @@     , list-transformer     , mtl     , transformers+    , text   default-language: Haskell2010

@@ -62,7 +62,8 @@   , (.|)   , runPipe, runPipePure   -- * Primitives-  , awaitEither, await, awaitWith, awaitSurely, awaitForever, yield+  , awaitEither, await, awaitWith, awaitSurely, awaitForever+  , yield, yieldLazy   -- * Special chaining   , (&|), (|.)   , fuseBoth, fuseUpstream, fuseBothMaybe@@ -71,6 +72,7 @@   , feedPipe, feedPipeEither   -- * Pipe transformers   , mapInput, mapOutput, mapUpRes, trimapPipe+  , passthrough   , hoistPipe   , feedbackPipe, feedbackPipeEither   -- * Wrappers@@ -88,18 +90,19 @@ import           Control.Applicative import           Control.Monad import           Control.Monad.Trans.Class-import           Control.Monad.Trans.Free        (FreeT(..), FreeF(..))+import           Control.Monad.Trans.Free         (FreeT(..), FreeF(..)) import           Control.Monad.Trans.Free.Church import           Control.Monad.Trans.State import           Data.Bifunctor import           Data.Conduino.Internal import           Data.Functor import           Data.Functor.Identity-import           Data.Sequence                   (Seq(..))+import           Data.Sequence                    (Seq(..)) import           Data.Void-import           List.Transformer                (ListT(..), Step(..))-import qualified Data.Sequence                   as Seq-import qualified List.Transformer                as LT+import           List.Transformer                 (ListT(..), Step(..))+import qualified Control.Monad.Trans.State.Strict as SS+import qualified Data.Sequence                    as Seq+import qualified List.Transformer                 as LT  -- | Await input from upstream.  Will block until upstream 'yield's. --@@ -111,6 +114,7 @@ -- Will always return 'Just' if @u@ is 'Void'. await :: Pipe i o u m (Maybe i) await = either (const Nothing) Just <$> awaitEither+{-# INLINE await #-}  -- | 'await', but directly chaining a continuation if the 'await' was -- succesful.@@ -123,6 +127,7 @@ awaitWith f = awaitEither >>= \case     Left  r -> pure r     Right x -> f x+{-# INLINE awaitWith #-}  -- | Await input from upstream where the upstream pipe is guaranteed to -- never terminate.@@ -136,6 +141,7 @@ -- of 'awaitSurely'. awaitSurely :: Pipe i o Void m i awaitSurely = either absurd id <$> awaitEither+{-# INLINE awaitSurely #-}  -- | A useful utility function over repeated 'await's.  Will repeatedly -- 'await' and then continue with the given pipe whenever the upstream pipe@@ -148,6 +154,7 @@ -- @ awaitForever :: (i -> Pipe i o u m a) -> Pipe i o u m u awaitForever = awaitForeverWith pure+{-# INLINE awaitForever #-}  -- | 'awaitForever', but with a way to handle the result of the -- upstream pipe, which will be called when the upstream pipe stops@@ -161,6 +168,7 @@     go = awaitEither >>= \case       Left x  -> mapInput (const ()) $ f x       Right x -> g x *> go+{-# INLINE awaitForeverWith #-}  -- | Run a pipe that is both a source and a sink (an "effect") into the -- effect that it represents.@@ -210,6 +218,7 @@ -- effects. runPipePure :: Pipe () Void Void Identity a -> a runPipePure = runIdentity . runPipe+{-# INLINE runPipePure #-}  -- | Repeatedly run 'squeezePipe' by giving it items from an input list. -- Returns the outputs observed, and 'Left' if the input list was exhausted@@ -224,6 +233,7 @@     -> m ([o], Either (i -> Pipe i o u m a) ([i], a)) feedPipe xs = (fmap . second . first) (. Right)             . feedPipeEither xs+{-# INLINE feedPipe #-}  -- | Repeatedly run 'squeezePipeEither' by giving it items from an input -- list.  Returns the outputs observed, and 'Left' if the input list was@@ -244,6 +254,7 @@         []   -> pure (zs, Left n)         y:ys -> first (zs ++) <$> feedPipeEither ys (n (Right y))       Right z -> pure (zs, Right (xs, z))+{-# INLINE feedPipeEither #-}  -- | "Squeeze" a pipe by extracting all output that can be extracted -- before any input is requested.  Returns a 'Left' if the pipe eventually@@ -257,6 +268,7 @@     -> m ([o], Either (i -> Pipe i o u m a) a) squeezePipe = (fmap . second . first) (. Right)             . squeezePipeEither+{-# INLINE squeezePipe #-}  -- | "Squeeze" a pipe by extracting all output that can be extracted before -- any input is requested.  Returns a 'Left' if the pipe eventually does@@ -269,7 +281,7 @@     :: Monad m     => Pipe i o u m a     -> m ([o], Either (Either u i -> Pipe i o u m a) a)-squeezePipeEither p = runFT (pipeFree p)+squeezePipeEither (Pipe (FT p)) = p     (pure . ([],) . Right)     (\pNext -> \case         PAwaitF f g -> pure . ([],) . Left $ (unSqueeze =<<) . lift . pNext . either f g@@ -281,6 +293,7 @@       case nxt of         Left f  -> f =<< awaitEither         Right a -> pure a+{-# INLINE squeezePipeEither #-}  -- | The main operator for chaining pipes together.  @pipe1 .| pipe2@ will -- connect the output of @pipe1@ to the input of @pipe2@.@@ -305,8 +318,9 @@     => Pipe a b u m v     -> Pipe b c v m r     -> Pipe a c u m r-Pipe p .| Pipe q = Pipe $ toFT $ compPipe_ (fromFT p) (fromFT q)+p .| q = fromRecPipe $ compPipe_ (toRecPipe p) (toRecPipe q) infixr 2 .|+{-# INLINE (.|) #-}  compPipe_     :: forall a b c u v m r. (Monad m)@@ -338,6 +352,7 @@         go = awaitEither >>= \case           Left  y -> pure (y, x)           Right _ -> go+{-# INLINE fuseBoth #-}  -- | Like 'fuseBoth' and '&|', except does not wait for the upstream pipe -- to terminate.  Return 'Nothing' in the first field if the upstream pipe hasn't terminated,@@ -349,6 +364,7 @@                  .| (q >>= check)   where     check x = (,x) . either Just (const Nothing) <$> awaitEither+{-# INLINE fuseBothMaybe #-}  -- | Useful prefix version of '|.'. --@@ -359,6 +375,7 @@     -> Pipe b c v m r     -> Pipe a c u m v fuseUpstream p q = fst <$> fuseBoth p q+{-# INLINE fuseUpstream #-}  -- | Like @.|@, but get the result of /both/ pipes on termination, instead -- of just the second.  This means that @p &| q@ will only terminate with a result when@@ -368,6 +385,7 @@ -- @since 0.2.1.0 (&|) :: Monad m => Pipe a b u m v -> Pipe b c v m r -> Pipe a c u m (v, r) (&|) = fuseBoth+{-# INLINE (&|) #-}  -- | Like @.|@, but keep the result of the /first/ pipe, instead of the -- second.  This means that @p |. q@ will only terminate with a result when@@ -377,10 +395,30 @@ -- @since 0.2.1.0 (|.) :: Monad m => Pipe a b u m v -> Pipe b c v m r -> Pipe a c u m v (|.) = fuseUpstream+{-# INLINE (|.) #-}  infixr 2 &| infixr 2 |. +-- | Passthrough and pair each output with the /last/ input that triggered+-- it.  'Nothing' will occur initially if the pipe outputs anything without+-- consuming any values, but after the first 'Just', should only output+-- Justs forever.+--+-- @since 0.2.3.0+passthrough+    :: Monad m+    => Pipe i o u m a+    -> Pipe i (Maybe i, o) u m a+passthrough p = fmap fst . runStatePS Nothing $+       awaitForever passOn+    .| hoistPipe lift p+    .| awaitForever tagIn+  where+    passOn i = lift (SS.put (Just i)) *> yield i+    tagIn i = yield . (,i) =<< lift SS.get+{-# INLINE passthrough #-}+ -- | Loop a pipe into itself. -- -- *  Will feed all output back to the input@@ -393,6 +431,7 @@     => Pipe x x u m a     -> Pipe x x u m a feedbackPipe = feedbackPipeEither . mapInput (either id id)+{-# INLINE feedbackPipe #-}  -- | A version of 'feedbackPipe' that distinguishes upstream input from -- downstream output fed back.  Gets 'Left' from upstream, and 'Right' from@@ -407,19 +446,20 @@     :: Monad m     => Pipe (Either i o) o u m a     -> Pipe i o u m a-feedbackPipeEither p = fmap fst . runStateP Seq.empty $+feedbackPipeEither p = fmap fst . runStatePS Seq.empty $        popper     .| hoistPipe lift p-    .| awaitForever (\x -> lift (modify (:|> x)) *> yield x)+    .| awaitForever (\x -> lift (SS.modify (:|> x)) *> yield x)   where-    popper = lift get >>= \case+    popper = lift SS.get >>= \case       Empty -> awaitEither >>= \case         Left r  -> pure r         Right x -> yield (Left x) >> popper       x :<| xs -> do-        lift $ put xs+        lift $ SS.put xs         yield (Right x)         popper+{-# INLINE feedbackPipeEither #-}  -- | A newtype wrapper over a source (@'Pipe' () o 'Void'@) that gives it an -- alternative 'Applicative' and 'Alternative' instance, matching "ListT@@ -472,6 +512,7 @@ zipSource :: Monad m => Pipe () (a -> b) u m () -> Pipe () a v m () -> Pipe () b w m () zipSource (PipeList fs) (PipeList xs) = PipeList . fmap Just $     uncurry ($) <$> LT.zip (concatListT fs) (concatListT xs)+{-# INLINE zipSource #-}  concatListT :: Monad m => ListT m (Maybe a) -> ListT m a concatListT xs = ListT $ next xs >>= \case@@ -495,12 +536,13 @@     :: Applicative m     => Pipe () o u m ()     -> ListT m (Maybe o)-toListT p = ListT $ runFT (pipeFree p)+toListT (Pipe (FT p)) = ListT $ p     (\_ -> pure Nil)     (\pNext -> \case         PAwaitF _ g -> pure $ Cons Nothing  (ListT . pNext $ g ())         PYieldF x y -> pure $ Cons (Just x) (ListT . pNext $ y   )     )+{-# INLINE toListT #-}  -- | A source is essentially 'ListT' producing a 'Maybe' result.  This -- converts a 'ListT' to the source it encodes.@@ -526,6 +568,7 @@ genSource f = Pipe $ FT $ \pDone pFree -> f $ \case     Nothing      -> pDone ()     Just (x, xs) -> pFree id (PYieldF x xs)+{-# INLINE genSource #-}  -- | A source can be "run" by providing a continuation to handle and -- sequence each of its outputs.  Is ths inverse of 'genSource'.@@ -535,7 +578,7 @@     :: Pipe () o u m ()     -> (Maybe (o, m r) -> m r)    -- ^ handler ('Nothing' = done, @'Just' (x, next)@ = yielded value and next action     -> m r-withSource p f = runFT (pipeFree p)+withSource (Pipe (FT p)) f = p     (\_ -> f Nothing)     (\pNext -> \case         PAwaitF _ g -> pNext $ g ()@@ -552,6 +595,7 @@ unconsZipSource (ZipSource (PipeList p)) = next p <&> \case     Cons x xs -> Just (x, ZipSource (PipeList xs))     Nil       -> Nothing+{-# INLINE unconsZipSource #-}  -- | A newtype wrapper over a sink (@'Pipe' i 'Void'@) that gives it an -- alternative 'Applicative' and 'Alternative' instance.@@ -607,7 +651,8 @@     => Pipe i Void u m (a -> b)     -> Pipe i Void u m a     -> Pipe i Void u m b-zipSink (Pipe p) (Pipe q) = Pipe $ toFT $ zipSink_ (fromFT p) (fromFT q)+zipSink p q = fromRecPipe $ zipSink_ (toRecPipe p) (toRecPipe q)+{-# INLINE zipSink #-}  -- | Distribute input to both sinks, and finishes with the result of -- the one that finishes first.@@ -616,7 +661,8 @@     => Pipe i Void u m a     -> Pipe i Void u m a     -> Pipe i Void u m a-altSink (Pipe p) (Pipe q) = Pipe $ toFT $ altSink_ (fromFT p) (fromFT q)+altSink p q = fromRecPipe $ altSink_ (toRecPipe p) (toRecPipe q)+{-# INLINE altSink #-}  -- | '<*>' = distribute input to all, and return result when they finish --

@@ -38,6 +38,7 @@   , repeatEitherM   , replicateM   , sourceHandleLines+  , sourceHandleLinesText   , stdinLines   , sourceHandle   , stdin@@ -87,6 +88,8 @@ import qualified Data.ByteString               as BS import qualified Data.ByteString.Lazy.Internal as BSL import qualified Data.Sequence                 as Seq+import qualified Data.Text                     as T+import qualified Data.Text.IO                  as T import qualified System.IO                     as S  -- | A version of 'unfoldMaybe' that can choose the "result" value by@@ -100,6 +103,7 @@     go z = case f z of       Left  r       -> pure r       Right (x, z') -> yield x *> go z'+{-# INLINE unfoldEither #-}  -- | A version of 'unfold' that can terminate and end by returning -- 'Nothing'.@@ -108,6 +112,7 @@     -> s     -> Pipe i o u m () unfoldMaybe f = unfoldEither (maybe (Left ()) Right . f)+{-# INLINE unfoldMaybe #-}  -- | Repeatedly apply an "unfolding" function to a given initial state, -- yielding the first item in the tuple as output and updating the state as@@ -122,6 +127,7 @@     go z = yield x *> go z'       where         (x, z') = f z+{-# INLINE unfold #-}  -- | A version of 'iterateMaybe' that can specify a result value by -- providing it in the 'Left'.@@ -130,6 +136,7 @@     -> o     -> Pipe i o u m a iterateEither f = unfoldEither (fmap (join (,)) . f)+{-# INLINE iterateEither #-}  -- | A version of 'iterate' that can choose to terminate and stop by -- returning 'Nothing'.@@ -138,6 +145,7 @@     -> o     -> Pipe i o u m () iterateMaybe f = unfoldMaybe (fmap (join (,)) . f)+{-# INLINE iterateMaybe #-}  -- | Repeatedly apply a function to a given starting value and yield each -- result forever.@@ -161,18 +169,22 @@     -> o     -> Pipe i o u m a iterate f = unfold (join (,) . f)+{-# INLINE iterate #-}  -- | Yield every item in a foldable container. sourceList :: Foldable t => t a -> Pipe i a u m () sourceList = traverse_ yield+{-# INLINE sourceList #-}  -- | Repeatedly yield a given item forever. repeat :: o -> Pipe i o u m a repeat = forever . yield+{-# INLINE repeat #-}  -- | Yield a given item a certain number of times. replicate :: Int -> o -> Pipe i o u m () replicate n = replicateM_ n . yield+{-# INLINE replicate #-}  -- | Like 'repeatMaybeM', but allow specification of a final result type. repeatEitherM@@ -184,6 +196,7 @@     go = lift x >>= \case       Left r  -> pure r       Right y -> yield y *> go+{-# INLINE repeatEitherM #-}  -- | Repeat a monadic action, yielding the item in the 'Just' every time. -- As soon as it sees 'Nothing', stop producing forever.@@ -195,6 +208,7 @@     => m (Maybe o)     -> Pipe i o u m () repeatMaybeM = repeatEitherM . fmap (maybe (Left ()) Right)+{-# INLINE repeatMaybeM #-}  -- | Repeat a monadic action a given number of times, yielding each result, -- and then stop producing forever.@@ -206,22 +220,28 @@     => Int     -> m o     -> Pipe i o u m ()-replicateM n x = replicateM_ n $ lift x >>= yield+replicateM n = replicateM_ n . (yield =<<) . lift+{-# INLINE replicateM #-}  -- | Source from each line received from 'stdin'.  This stops as soon as -- end-of-file is reached, or an empty line is seen. stdinLines :: MonadIO m => Pipe i String u m () stdinLines = sourceHandleLines S.stdin+          .| takeWhile (not . null)+{-# INLINE stdinLines #-}  -- | Source from stdin, yielding bytestrings as they are drawn.  If you -- want to retrieve each line as a string, see 'stdinLines'. stdin :: MonadIO m => Pipe i BS.ByteString u m () stdin = sourceHandle S.stdin+{-# INLINE stdin #-}  -- | Source from a given I/O handle, yielding each line drawn as a string. -- To draw raw bytes, use 'sourceHandle'. ----- This stop as soon as end-of-file is reached, or an empty line is seen.+-- This stop as soon as end-of-file is reached.+--+-- Sinve v0.2.3.0, this now continues through empty lines. sourceHandleLines     :: MonadIO m     => S.Handle@@ -232,9 +252,28 @@       then pure Nothing       else liftIO . catchJust                 (guard . isEOFError)-                (mfilter (not . null) . Just <$> S.hGetLine h)+                (Just <$> S.hGetLine h)                 $ \_ -> pure Nothing+{-# INLINE sourceHandleLines #-} +-- | Source from a given I/O handle, yielding each line drawn as a string.+-- To draw raw bytes, use 'sourceHandle'.+--+-- This stop as soon as end-of-file is reached, or an empty line is seen.+sourceHandleLinesText+    :: MonadIO m+    => S.Handle+    -> Pipe i T.Text u m ()+sourceHandleLinesText h = repeatMaybeM $ do+    d <- liftIO $ S.hIsEOF h+    if d+      then pure Nothing+      else liftIO . catchJust+                (guard . isEOFError)+                (Just <$> T.hGetLine h)+                $ \_ -> pure Nothing+{-# INLINE sourceHandleLinesText #-}+ -- | Source from a given I/O handle, yielding bytestrings as they are -- pulled.  If you want to retrieve each line as a string, see -- 'sourceHandleLines'.@@ -246,6 +285,7 @@                . fmap (mfilter (not . BS.null) . Just)                . liftIO                $ BS.hGetSome h BSL.defaultChunkSize+{-# INLINE sourceHandle #-}  -- | Sink into a given I/O handle, writing each input to the handle. sinkHandle@@ -254,14 +294,17 @@     -> Pipe BS.ByteString o u m () sinkHandle h = mapM (liftIO . BS.hPut h)             .| sinkNull+{-# INLINE sinkHandle #-}  -- | A sink into stdout. stdout :: MonadIO m => Pipe BS.ByteString o u m () stdout = sinkHandle S.stdout+{-# INLINE stdout #-}  -- | A sink into stderr. stderr :: MonadIO m => Pipe BS.ByteString o u m () stderr = sinkHandle S.stderr+{-# INLINE stderr #-}  -- | Repeat a monadic action forever, yielding each output. --@@ -274,15 +317,18 @@ repeatM x = go   where     go = (yield =<< lift x) *> go+{-# INLINE repeatM #-}  -- | Process every incoming item with a pure function, and yield its -- output. map :: (i -> o) -> Pipe i o u m u map f = awaitForever (yield . f)+{-# INLINE map #-}  -- | Map a monadic function to process every input, and yield its output. mapM :: Monad m => (i -> m o) -> Pipe i o u m u mapM f = awaitForever ((yield =<<) . lift . f)+{-# INLINE mapM #-}  -- | Execute a monadic function to process every input, passing through the -- original value back downstream.@@ -290,6 +336,7 @@ -- @since 0.2.1.0 iterM :: Monad m => (i -> m ()) -> Pipe i i u m u iterM f = mapM (\x -> x <$ f x)+{-# INLINE iterM #-}  -- | Map a pure "stateful" function over each incoming item.  Give -- a function to update the state and return an output and an initial@@ -303,6 +350,7 @@     go !x = awaitWith $ \y ->         let (!x', !z) = f y x         in  yield z *> go x'+{-# INLINE mapAccum #-}  -- | Like 'foldl', but yields every accumulator value downstream. --@@ -320,6 +368,7 @@     go !x = awaitWith $ \y ->       let x' = f x y       in  yield x' *> go x'+{-# INLINE scan #-}  -- | Yield consecutive pairs of values. --@@ -333,6 +382,7 @@     go x = awaitWith $ \y -> do       yield (x, y)       go y+{-# INLINE pairs #-}  -- | Yield consecutive runs of at most @n@ of values, starting with an -- empty sequence.@@ -357,7 +407,7 @@     go xs = do       yield xs       awaitWith $ \y -> go . Seq.drop (Seq.length xs - n + 1) $ (xs Seq.:|> y)-+{-# INLINE consecutive #-}  -- | Let a given number of items pass through the stream uninhibited, and -- then stop producing forever.@@ -373,6 +423,7 @@ take :: Int -> Pipe i i u m () take n = void . runMaybeT . replicateM_ n $     lift . yield =<< MaybeT await+{-# INLINE take #-}  -- | Let elements pass until an element is received that does not satisfy -- the predicate, then stop producing forever.@@ -386,12 +437,14 @@       Just x         | p x       -> yield x *> go         | otherwise -> pure ()+{-# INLINE takeWhile #-}  -- | Only allow values satisfying a predicate to pass. filter     :: (i -> Bool)     -> Pipe i i u m u filter p = awaitForever $ \x -> when (p x) $ yield x+{-# INLINE filter #-}  -- | Map a function returning a container onto every incoming item, and -- yield all of the outputs from that function.@@ -400,11 +453,13 @@     => (i -> t o)     -> Pipe i o u m u concatMap f = awaitForever (sourceList . f)+{-# INLINE concatMap #-}  -- | Take an input of containers and output each of their elements -- successively. concat :: Foldable t => Pipe (t i) i u m u concat = awaitForever sourceList+{-# INLINE concat #-}  -- | Right-fold every input into an accumulated value. --@@ -416,6 +471,7 @@     go = await >>= \case       Nothing -> pure z       Just x  -> f x <$> go+{-# INLINE foldr #-}  -- | Left-fold every input into an accumulated value. --@@ -427,11 +483,13 @@     go !z = await >>= \case       Nothing -> pure z       Just !x -> go (f z x)+{-# INLINE foldl #-}  -- | Fold every incoming item monoidally, and return the result once -- finished. fold :: Monoid a => Pipe a o u m a fold = foldl (<>) mempty+{-# INLINE fold #-}  -- | Fold every incoming item according to a monoidal projection, and -- return the result once finished.@@ -448,13 +506,16 @@ -- @ foldMap :: Monoid a => (i -> a) -> Pipe i o u m a foldMap f = foldl (\x y -> x <> f y) mempty+{-# INLINE foldMap #-}  -- | Sink every incoming item into a list. -- -- Note that this keeps the entire list in memory until it is all--- eventually read.+-- eventually read, and does no "lazy IO".  It exhauts the whole stream+-- before returning anything. sinkList :: Pipe i o u m [i] sinkList = foldr (:) []+{-# INLINE sinkList #-}  -- | Ignore a certain amount of items from the input stream, and then stop -- producing forever.@@ -466,6 +527,7 @@ -- [4,5,6,7,8] drop :: Int -> Pipe i o u m () drop n = replicateM_ n await+{-# INLINE drop #-}  -- | Ignore items from an input stream as long as they match a predicate. -- Afterwards, stop producing forever.@@ -482,6 +544,7 @@       Just x         | p x       -> go         | otherwise -> pure ()+{-# INLINE dropWhile #-}  -- | Consume an entire input stream and ignore all of its outputs. sinkNull :: Pipe i o u m ()@@ -493,4 +556,9 @@ -- -- To get the first item ("head"), use 'await' or 'awaitSurely'. last :: Pipe i o u m (Maybe i)-last = fmap getLast <$> foldMap (Just . Last)+last = go Nothing+  where+    go x = await >>= \case+      Nothing -> pure x+      Just r  -> go (Just r)+{-# INLINE last #-}

@@ -26,13 +26,13 @@     Pipe(..)   , PipeF(..)   , awaitEither-  , yield-  , trimapPipe, mapInput, mapOutput, mapUpRes+  , yield, yieldLazy+  , trimapPipe, trimapPipeF, mapInput, mapOutput, mapUpRes   , hoistPipe   , RecPipe   , toRecPipe, fromRecPipe   , withRecPipe-  , runStateP+  , runStateP, runStatePS   , pAwaitF, pYieldF   ) where @@ -46,6 +46,7 @@ import           Control.Monad.Trans.Free.Church import           Control.Monad.Trans.State import           Data.Functor+import qualified Control.Monad.Trans.State.Strict  as SS  #if !MIN_VERSION_base(4,13,0) import           Control.Monad.Fail@@ -70,11 +71,15 @@ -- thing but explicitly recursive. data PipeF i o u a =       PAwaitF (u -> a) (i -> a)+      -- todo: strict version?     | PYieldF o a   deriving Functor  makeFree ''PipeF +{-# INLINE pYieldF #-}+{-# INLINE pAwaitF #-}+ -- | Similar to a conduit from the /conduit/ package. -- -- For a @'Pipe' i o u m a@, you have:@@ -176,11 +181,22 @@ -- (expected input type) or a @u@ if the upstream pipe terminates. awaitEither :: Pipe i o u m (Either u i) awaitEither = pAwaitF+{-# INLINE awaitEither #-}  -- | Send output downstream.+--+-- Since v0.2.3.0, is strict.  See 'yieldLazy' for the original behavior. yield :: o -> Pipe i o u m ()-yield = pYieldF+yield x = x `seq` yieldLazy x+{-# INLINE yield #-} +-- | Send output downstream without forcing its argument.+--+-- @since 0.2.3.0+yieldLazy :: o -> Pipe i o u m ()+yieldLazy = pYieldF+{-# INLINE yieldLazy #-}+ -- | Map over the input type, output type, and upstream result type. -- -- If you want to map over the result type, use 'fmap'.@@ -190,12 +206,20 @@     -> (u -> v)     -> Pipe j p v m a     -> Pipe i o u m a-trimapPipe f g h = Pipe . transFT go . pipeFree-  where-    go = \case-      PAwaitF a b -> PAwaitF (a . h) (b . f)-      PYieldF a x -> PYieldF (g a) x+trimapPipe f g h (Pipe p) = Pipe (transFT (trimapPipeF f g h) p)+{-# INLINE trimapPipe #-} +trimapPipeF+    :: (i -> j)+    -> (p -> o)+    -> (u -> v)+    -> PipeF j p v a+    -> PipeF i o u a+trimapPipeF f g h = \case+    PAwaitF a b -> PAwaitF (a . h) (b . f)+    PYieldF a x -> PYieldF (g a) x+{-# INLINE trimapPipeF #-}+ -- | Transform the underlying monad of a pipe. -- -- Note that if you are trying to work with monad transformers, this is@@ -207,20 +231,24 @@     -> Pipe i o u m a     -> Pipe i o u n a hoistPipe f = Pipe . hoistFT f . pipeFree+{-# INLINE hoistPipe #-}  -- | (Contravariantly) map over the expected input type. mapInput :: (i -> j) -> Pipe j o u m a -> Pipe i o u m a mapInput f = trimapPipe f id id+{-# INLINE mapInput #-}  -- | Map over the downstream output type. -- -- If you want to map over the result type, use 'fmap'. mapOutput :: (p -> o) -> Pipe i p u m a -> Pipe i o u m a mapOutput f = trimapPipe id f id+{-# INLINE mapOutput #-}  -- | (Contravariantly) map over the upstream result type. mapUpRes :: (u -> v) -> Pipe i o v m a -> Pipe i o u m a mapUpRes = trimapPipe id id+{-# INLINE mapUpRes #-}  -- | A version of 'Pipe' that uses explicit, concrete recursion instead of -- church-encoding like 'Pipe'.  Some functions --- especially ones that@@ -232,11 +260,18 @@ -- defined in terms of 'Pipe', it can be easier to write certain low-level -- pipe combining functions in terms of 'RecPipe' than 'Pipe'. toRecPipe :: Monad m => Pipe i o u m a -> RecPipe i o u m a-toRecPipe = fromFT . pipeFree+toRecPipe = _fromFT . pipeFree+{-# INLINE toRecPipe #-} +-- | An inlined version of fromFT+_fromFT :: (Monad m, Functor f) => FT f m a -> FreeT f m a+_fromFT (FT k) = FreeT $ k (return . Pure) (\xg -> runFreeT . wrap . fmap (FreeT . xg))+{-# INLINE _fromFT #-}+ -- | Convert a 'RecPipe' back into a 'Pipe'. fromRecPipe :: Monad m => RecPipe i o u m a -> Pipe i o u m a fromRecPipe = Pipe . toFT+{-# INLINE fromRecPipe #-}  -- | Convenint wrapper over 'toRecPipe' and 'fromRecPipe'. --@@ -247,6 +282,7 @@     -> Pipe i o u m a     -> Pipe j p v n b withRecPipe f = fromRecPipe . f . toRecPipe+{-# INLINE withRecPipe #-}  -- | Turn a 'Pipe' that runs over 'StateT' into a "state-modifying 'Pipe'", -- that returns the final state when it terminates.@@ -304,4 +340,21 @@       case q of         Pure x -> Pure (x, s')         Free l -> Free $ go s' <$> l+{-# INLINE runStateP #-}++-- | 'runStateP', but for "Control.Monad.Trans.State.Strict".+--+-- @since 0.2.1.0+runStatePS+    :: Monad m+    => s+    -> Pipe i o u (SS.StateT s m) a+    -> Pipe i o u m (a, s)+runStatePS = withRecPipe . go+  where+    go s (FreeT p) = FreeT $ SS.runStateT p s <&> \(q, s') ->+      case q of+        Pure x -> Pure (x, s')+        Free l -> Free $ go s' <$> l+{-# INLINE runStatePS #-} 

@@ -1,5 +1,6 @@-{-# LANGUAGE LambdaCase   #-}-{-# LANGUAGE ViewPatterns #-}+{-# LANGUAGE LambdaCase    #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE ViewPatterns  #-}  -- | -- Module      : Data.Conduino.Lift@@ -85,10 +86,12 @@   , catchP, runCatchP   ) where +import           Control.Monad import           Control.Monad.Catch.Pure import           Control.Monad.Trans.Class import           Control.Monad.Trans.Except import           Control.Monad.Trans.Free+import           Control.Monad.Trans.Free.Church import           Control.Monad.Trans.RWS           (RWST(..)) import           Control.Monad.Trans.Reader import           Control.Monad.Trans.State@@ -129,6 +132,7 @@     -> Pipe i o u (StateT s m) a     -> Pipe i o u m s execStateP s = fmap snd . runStateP s+{-# INLINE execStateP #-}  -- | Takes a 'Pipe' over 'StateT' and "hides" the state from the outside -- world.  Give an initial state --- the pipe behaves the same way, but to@@ -146,6 +150,7 @@     -> Pipe i o u (StateT s m) a     -> Pipe i o u m a evalStateP s = fmap fst . runStateP s+{-# INLINE evalStateP #-}  -- | 'stateP', but for "Control.Monad.Trans.State.Strict". --@@ -158,21 +163,7 @@     s       <- lift SS.get     (x, s') <- hoistPipe lift (f s)     x <$ lift (SS.put s')---- | 'runStateP', but for "Control.Monad.Trans.State.Strict".------ @since 0.2.1.0-runStatePS-    :: Monad m-    => s-    -> Pipe i o u (SS.StateT s m) a-    -> Pipe i o u m (a, s)-runStatePS = withRecPipe . go-  where-    go s (FreeT p) = FreeT $ SS.runStateT p s <&> \(q, s') ->-      case q of-        Pure x -> Pure (x, s')-        Free l -> Free $ go s' <$> l+{-# INLINE statePS #-}  -- | 'execStateP', but for "Control.Monad.Trans.State.Strict". --@@ -183,6 +174,7 @@     -> Pipe i o u (SS.StateT s m) a     -> Pipe i o u m s execStatePS s = fmap snd . runStatePS s+{-# INLINE execStatePS #-}  -- | 'evalStateP', but for "Control.Monad.Trans.State.Strict". --@@ -193,6 +185,7 @@     -> Pipe i o u (SS.StateT s m) a     -> Pipe i o u m a evalStatePS s = fmap fst . runStatePS s+{-# INLINE evalStatePS #-}  -- | Turn a "failable-result" 'Pipe' into a pipe over 'ExceptT'. --@@ -209,6 +202,7 @@ exceptP p = hoistPipe lift p >>= \case     Left  e -> lift $ throwE e     Right x -> pure x+{-# INLINE exceptP #-}  -- | Turn a 'Pipe' that runs over 'ExceptT' into an "early-terminating -- 'Pipe'" that "succesfully" returns 'Left' or 'Right'.@@ -259,12 +253,11 @@     :: Monad m     => Pipe i o u (ExceptT e m) a     -> Pipe i o u m (Either e a)-runExceptP = withRecPipe go-  where-    go (FreeT p) = FreeT $ runExceptT p <&> \case-      Left  e        -> Pure $ Left e-      Right (Pure x) -> Pure $ Right x-      Right (Free l) -> Free $ go <$> l+runExceptP (Pipe (FT f)) = Pipe $ FT $ \pr bd -> either (pr . Left) pure =<< runExceptT (+      f (lift . pr . Right)+        (\c -> lift . bd (either (pr . Left) pure <=< runExceptT . c))+    )+{-# INLINE runExceptP #-}  -- | A handy version of 'runExceptP' that discards its output, so it can be -- easier to chain using '.|'.  It's useful if you are using 'runExceptP'@@ -276,6 +269,7 @@     => Pipe i o u (ExceptT e m) a     -> Pipe i o u m () runExceptP_ = void . runExceptP+{-# INLINE runExceptP_ #-}  -- | Like 'exceptP', but for 'CatchT'.  See 'exceptP' for usage details and -- caveats.  In general, can be useful for chaining with other 'CatchT'@@ -294,6 +288,7 @@ catchP p = hoistPipe lift p >>= \case     Left  e -> lift $ throwM e     Right x -> pure x+{-# INLINE catchP #-}  -- | Like 'runExceptP', but for 'CatchT'.  See 'runExceptP' for usage -- details.  In general, can be useful for "isolating" a 'CatchT' pipe from@@ -304,12 +299,20 @@     :: Monad m     => Pipe i o u (CatchT m) a     -> Pipe i o u m (Either SomeException a)+-- runCatchP (Pipe (FT f)) = Pipe $ FT $ \pr bd -> either (pr . Left) pure =<< runCatchT (+--       f (lift . pr . Right)+--         (\c -> lift . bd (either (pr . Left) pure <=< runCatchT . c))+--     ) runCatchP = withRecPipe go   where-    go (FreeT p) = FreeT $ runCatchT p <&> \case-      Left  e        -> Pure $ Left e-      Right (Pure x) -> Pure $ Right x-      Right (Free l) -> Free $ go <$> l+    go (FreeT p) = FreeT $ runCatchT p <&>+      either (Pure . Left)+             (\case Pure x -> Pure (Right x); Free l -> Free (go <$> l))+      -- \case+      -- Left  e        -> Pure $ Left e+      -- Right (Pure x) -> Pure $ Right x+      -- Right (Free l) -> Free $ go <$> l+{-# INLINE runCatchP #-}  -- | Turn a "parameterized 'Pipe'" into a 'Pipe' that runs over 'ReaderT', -- so you can chain it with other 'ReaderT' pipes.@@ -414,6 +417,7 @@       case r of         Pure x -> Pure (x, w')         Free l -> Free $ go w' <$> l+{-# INLINE runWriterP #-}  -- | 'runWriterP', but only returning the final log after succesful -- termination.@@ -424,6 +428,7 @@     => Pipe i o u (WriterT w m) a     -> Pipe i o u m w execWriterP = fmap snd . runWriterP+{-# INLINE execWriterP #-}  -- | 'writerP', but for "Control.Monad.Trans.Writer.Strict". --@@ -449,6 +454,7 @@       case r of         Pure x -> Pure (x, w')         Free l -> Free $ go w' <$> l+{-# INLINE runWriterPS #-}  -- | 'execWriterP', but for "Control.Monad.Trans.Writer.Strict". --@@ -458,6 +464,7 @@     => Pipe i o u (WriterT w m) a     -> Pipe i o u m w execWriterPS = fmap snd . runWriterP+{-# INLINE execWriterPS #-}  -- | Turn a parameterized, state-transforming, log-producing 'Pipe' into -- a 'Pipe' over 'RWST', which can be useful for chaining it with other@@ -501,6 +508,7 @@       case q of         Pure x -> Pure (x, s', w')         Free l -> Free $ go w' s' <$> l+{-# INLINE runRWSP #-}  -- | 'runRWSP', but ignoring the final state. --@@ -512,6 +520,7 @@     -> Pipe i o u (RWST r w s m) a     -> Pipe i o u m (a, w) evalRWSP r s = fmap (\(x,_,w) -> (x,w)) . runRWSP r s+{-# INLINE evalRWSP #-}  -- | 'runRWSP', but ignoring the result value. --@@ -523,6 +532,7 @@     -> Pipe i o u (RWST r w s m) a     -> Pipe i o u m (s, w) execRWSP r s = fmap (\(_,s',w) -> (s',w)) . runRWSP r s+{-# INLINE execRWSP #-}  -- | 'rwsP', but for "Control.Monad.Trans.RWS.Strict". --@@ -553,6 +563,7 @@       case q of         Pure x -> Pure (x, s', w')         Free l -> Free $ go w' s' <$> l+{-# INLINE runRWSPS #-}  -- | 'evalRWSPS', but for "Control.Monad.Trans.RWS.Strict". --@@ -564,6 +575,7 @@     -> Pipe i o u (RWSS.RWST r w s m) a     -> Pipe i o u m (a, w) evalRWSPS r s = fmap (\(x,_,w) -> (x,w)) . runRWSPS r s+{-# INLINE evalRWSPS #-}  -- | 'execRWSPS', but for "Control.Monad.Trans.RWS.Strict". --@@ -575,3 +587,4 @@     -> Pipe i o u (RWSS.RWST r w s m) a     -> Pipe i o u m (s, w) execRWSPS r s = fmap (\(_,s',w) -> (s',w)) . runRWSPS r s+{-# INLINE execRWSPS #-}