diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,17 @@ -# Changelog for conduino +Changelog +========= -## Unreleased changes +Version 0.2.0.0 +--------------- + +*October 30, 2019* + + + +* Initial release + +Version 0.1.0.0 +--------------- + +(Accidental incomplete release made by mistake) + diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,1 +1,135 @@ # conduino + +A lightweight continuation-based stream processing library. + +It is similar in nature to pipes and conduit, but useful if you just want +something quick to manage composable stream processing without focus on IO. + +## Why a stream processing library? + +A stream processing library is a way to stream processors in a *composable* way: +instead of defining your entire stream processing function as a single +recursive loop with some global state, instead think about each "stage" of the process, +and isolate each state to its own segment. Each component can contain its own +isolated state: + +```haskell +runPipePure $ sourceList [1..10] + .| scan (+) 0 + .| sinkList +-- [1,3,6,10,15,21,28,36,45,55] +``` + +All of these components have internal "state": + +* `sourceList` keeps track of "which" item in the list to yield next +* `scan` keeps track of the current running sum +* `sinkList` keeps track of all items that have been seen so far, as a list + +They all work together without knowing any other component's internal state, so +you can write your total streaming function without concerning yourself, at +each stage, with the entire part. + +In addition, there are useful functions to "combine" stream processors: + +* `zipSink` combines sinks in an "and" sort of way: combine two sinks in + parallel and finish when all finish. +* `altSink` combines sinks in an "or" sort of way: combine two sinks in + parallel and finish when any of them finish +* `zipSource` combines sources in parallel and collate their outputs. + +Stream processing libraries are also useful for streaming composition of +monadic effects (like IO or State), as well. + +## Details and usage + +API-wise, is closer to *conduit* than *pipes*. Pull-based, where the main +"running" function is: + +```haskell +runPipe :: Pipe () Void u m a -> m a +``` + +That is, the "production" and "consumption" is integrated into one single pipe, +and then run all at once. Contrast this to *pipes*, where consumption is not +integrated into the pipe, but rather your choice of "runner" determines how +your pipe is consumed. + +One extra advantage over *conduit* is that we have the ability to model pipes +that will never stop producing output, so we can have an `await` function that +can reliably fetch items upstream. This matches more *pipes*-style requests. + +For a `Pipe i o u m a`, you have: + +* `i`: Type of input stream (the things you can `await`) +* `o`: Type of output stream (the things you `yield`) +* `u`: Type of the *result* of the upstream pipe (Outputted when upstream + pipe terminates) +* `m`: Underlying monad (the things you can `lift`) +* `a`: Result type when pipe terminates (outputted when finished, with + `pure` or `return`) + +Some specializations: + +* If `i` is `()`, the pipe is a *source* --- it doesn't need anything to + produce items. It will pump out items on its own, for pipes downstream to + receive and process. + +* If `o` is `Void`, the pipe is a *sink* --- it will never `yield` anything + downstream. It will consume items from things upstream, and produce a + result (`a`) if and when it terminates. + +* If `u` is `Void`, then the pipe's upstream is limitless, and never + terminates. This means that you can use `awaitSurely` instead of `await`, + to get await a value that is guaranteed to come. You'll get an `i` instead + of a `Maybe i`. + + ```haskell + await :: Pipe i o u m (Maybe i) + awaitsurely :: Pipe i o Void m i + ``` + +* If `a` is `Void`, then the pipe never terminates --- it will keep on + consuming and/or producing values forever. If this is a sink, it means + that the sink will never terminate, and so `runPipe` will also never + terminate. If it is a source, it means that if you chain something + downstream with `.|`, that downstream pipe can use `awaitSurely` to + guarantee something being passed down. + +Usually you would use it by chaining together pipes with `.|` and then running +the result with `runPipe`. + +```haskell +runPipe $ someSource + .| somePipe + .| someOtherPipe + .| someSink +``` + + +## Why does this package exist? + +This package is taking some code I've used some closed-source projects and +pulling it out as a full library. I wrote it, despite the existence of *pipes* +and *conduit*, because: + +1. I wanted conduit-style semantics for stream composition (source - producer - + sink all in one package). +2. I wanted type-enforced guaranteed "awaits" based on type-enforced + guaranteed infinite producers. +3. I wanted to be able to combine stream processors "in parallel" in + different ways (`zipSink`, for "and", and `altSink`, for "or"). +3. I wanted something lightweight without the dependencies dealing with IO, + since I wasn't really doing resource-sensitive IO. + +*conduino* is a small, lightweight version that is focused not necessarily on +"effects" streaming, but rather on composable bits of logic. It is basically a +lightweight version of conduit-style streaming. It is slightly different from +pipes in terms of API. + +One major difference from *conduit* is the `u` parameter, which allows for +things like `awaitSurely`, to ensure that upstream pipes will never terminate. + +If you need to do some important IO and handle things like managing resources, +or leverage interoperability with existing libraries...switch to a more mature +library like *conduit* or *pipes* immediately :) diff --git a/conduino.cabal b/conduino.cabal --- a/conduino.cabal +++ b/conduino.cabal @@ -4,12 +4,17 @@ -- -- see: https://github.com/sol/hpack -- --- hash: cd71825a7b764b1f7f6f1cd7515dfaecdd7070939ea2baffd66443887143a5bf +-- hash: e4f50c4a474c128ace934290082d6a4eda455370d44bb3efb0192215a4fbcd1b name: conduino -version: 0.1.0.0 +version: 0.2.0.0 synopsis: Lightweight composable continuation-based stream processors -description: Please see the README on GitHub at +description: A lightweight continuation-based stream processing library. + . + It is similar in nature to pipes and conduit, but useful if you just want + something quick to manage composable stream processing without focus on IO. + . + See README for more information. category: Control homepage: https://github.com/mstksg/conduino#readme bug-reports: https://github.com/mstksg/conduino/issues @@ -18,6 +23,7 @@ copyright: (c) Justin Le 2019 license: BSD3 license-file: LICENSE +tested-with: GHC >= 8.4 && < 8.10 build-type: Simple extra-source-files: README.md @@ -30,14 +36,19 @@ library exposed-modules: Data.Conduino - Lib + Data.Conduino.Combinators + Data.Conduino.Internal other-modules: Paths_conduino hs-source-dirs: src ghc-options: -Wall -Wcompat -Wredundant-constraints -Werror=incomplete-patterns build-depends: - base >=4.7 && <5 + base >=4.11 && <5 + , bytestring + , containers , free + , list-transformer + , mtl , transformers default-language: Haskell2010 diff --git a/src/Data/Conduino.hs b/src/Data/Conduino.hs --- a/src/Data/Conduino.hs +++ b/src/Data/Conduino.hs @@ -2,207 +2,432 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE LambdaCase #-} +{-# LANGUAGE PatternSynonyms #-} +{-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeInType #-} +{-# LANGUAGE ViewPatterns #-} +-- | +-- Module : Data.Conduino +-- Copyright : (c) Justin Le 2019 +-- License : BSD3 +-- +-- Maintainer : justin@jle.im +-- Stability : experimental +-- Portability : non-portable +-- +-- Base API for 'Pipe'. See documentation for 'Pipe', '.|', and 'runPipe' +-- for information on usage. +-- +-- A "prelude" of useful pipes can be found in "Data.Conduino.Combinators". +-- +-- == Why a stream processing library? +-- +-- A stream processing library is a way to stream processors in a /composable/ way: +-- instead of defining your entire stream processing function as a single +-- recursive loop with some global state, instead think about each "stage" of the process, +-- and isolate each state to its own segment. Each component can contain its own +-- isolated state: +-- +-- >>> runPipePure $ sourceList [1..10] +-- .| scan (+) 0 +-- .| sinkList +-- [1,3,6,10,15,21,28,36,45,55] +-- +-- All of these components have internal "state": +-- +-- * @sourceList@ keeps track of "which" item in the list to yield next +-- * @scan@ keeps track of the current running sum +-- * @sinkList@ keeps track of all items that have been seen so far, as a list +-- +-- They all work together without knowing any other component's internal state, so +-- you can write your total streaming function without concerning yourself, at +-- each stage, with the entire part. +-- +-- In addition, there are useful functions to "combine" stream processors: +-- +-- * 'zipSink' combines sinks in an "and" sort of way: combine two sinks in +-- parallel and finish when all finish. +-- * 'altSink' combines sinks in an "or" sort of way: combine two sinks in +-- parallel and finish when any of them finish +-- * 'zipSource' combines sources in parallel and collate their outputs. +-- +-- Stream processing libraries are also useful for streaming composition of +-- monadic effects (like IO or State), as well. +-- module Data.Conduino ( Pipe , (.|) - , runPipe - , awaitEither, await, awaitSurely - , repeatM, unfoldP, unfoldPForever, iterateP, sourceList - , awaitForever, mapP, mapMP - , dropP - , foldrP, sinkList + , runPipe, runPipePure + , awaitEither, await, awaitWith, awaitSurely, awaitForever, yield + , mapInput, mapOutput, mapUpRes, trimapPipe + -- * Wrappers + , ZipSource(..) + , unconsZipSource , ZipSink(..) + , zipSink, altSink + -- * Generators + , toListT, fromListT + , pattern PipeList + , withSource, genSource ) where import Control.Applicative import Control.Monad -import Control.Monad.Free.Class -import Control.Monad.Free.TH import Control.Monad.Trans.Class import Control.Monad.Trans.Free (FreeT(..), FreeF(..)) import Control.Monad.Trans.Free.Church -import Data.Foldable +import Data.Conduino.Internal +import Data.Functor +import Data.Functor.Identity import Data.Void - -data PipeF i o u a = - PAwaitF (i -> a) (u -> a) - | PYieldF o a - deriving Functor - -makeFree ''PipeF +import List.Transformer (ListT(..), Step(..)) +import qualified List.Transformer as LT --- | Similar to Conduit +-- | Await input from upstream. Will block until upstream 'yield's. -- --- * @i@: Type of input stream --- * @o@: Type of output stream --- * @u@: Type of the /result/ of the upstream pipe (Outputted when --- upstream pipe finishes) --- * @m@: Underlying monad --- * @a@: Result type (Outputted when finished) +-- Will return 'Nothing' if the upstream pipe finishes and terminates. -- --- Some specializations: +-- If the upstream pipe never terminates, then you can use 'awaitSurely' to +-- guarantee a result. -- --- * A pipe is a /source/ if @i@ is '()': it doesn't need anything to go --- pump out items. +-- Will always return 'Just' if @u@ is 'Void'. +await :: Pipe i o u m (Maybe i) +await = either (const Nothing) Just <$> awaitEither + +-- | 'await', but directly chaining a continuation if the 'await' was +-- succesful. -- --- If a pipe is source and @a@ is 'Void', it means that it will --- produce forever. +-- The await will always be succesful if @u@ is 'Void'. -- --- * A pipe is a /sink/ if @o@ is 'Void': it will never yield anything --- else downstream. +-- This is a way of writing code in a way that is agnostic to how the +-- upstream pipe terminates. +awaitWith :: (i -> Pipe i o u m u) -> Pipe i o u m u +awaitWith f = awaitEither >>= \case + Left r -> pure r + Right x -> f x + +-- | Await input from upstream where the upstream pipe is guaranteed to +-- never terminate. -- --- * If a pipe is both a source and a sink, it is an /effect/. +-- A common type error will occur if @u@ (upstream pipe result type) is not +-- 'Void' -- it might be @()@ or some non-'Void' type. This means that the +-- upstream pipe terminates, so awaiting cannot be assured. -- --- * Normally you can ask for input upstream with 'await', which returns --- 'Nothing' if the pipe upstream stops producing. However, if @u@ is --- 'Void', it means that the pipe upstream will never stop, so you can --- use 'awaitSurely' to get a guaranteed answer. -newtype Pipe i o u m a = Pipe { pipeFree :: FT (PipeF i o u) m a } - deriving (Functor, Applicative, Monad, MonadTrans, MonadFree (PipeF i o u)) - -type Source o = Pipe () o -type Sink i = Pipe i Void -type Effect = Pipe () Void -type Forever p = p Void - -awaitEither :: Pipe i o u m (Either i u) -awaitEither = pAwaitF - -yield :: o -> Pipe i o u m () -yield = pYieldF +-- In that case, either change your upstream pipe to be one that never +-- terminates (which is most likely not possible), or use 'await' instead +-- of 'awaitSurely'. +awaitSurely :: Pipe i o Void m i +awaitSurely = either absurd id <$> awaitEither -await :: Pipe i o u m (Maybe i) -await = either Just (const Nothing) <$> awaitEither +-- | A useful utility function over repeated 'await's. Will repeatedly +-- 'await' and then continue with the given pipe whenever the upstream pipe +-- yields. +-- +-- Can be used to implement many pipe combinators: +-- +-- @ +-- 'Data.Conduino.Combinators.map' f = 'awaitForever' $ \x -> 'yield' (f x) +-- @ +awaitForever :: (i -> Pipe i o u m a) -> Pipe i o u m u +awaitForever = awaitForeverWith pure -awaitSurely :: Pipe i o Void m i -awaitSurely = either id absurd <$> awaitEither +-- | 'awaitForever', but with a way to handle the result of the +-- upstream pipe, which will be called when the upstream pipe stops +-- producing. +awaitForeverWith + :: (u -> Pipe () o u m b) -- ^ how to handle upstream ending, transitioning to a source + -> (i -> Pipe i o u m a) -- ^ how to handle upstream output + -> Pipe i o u m b +awaitForeverWith f g = go + where + go = awaitEither >>= \case + Left x -> mapInput (const ()) $ f x + Right x -> g x *> go +-- | Run a pipe that is both a source and a sink (an "effect") into the +-- effect that it represents. +-- +-- Usually you wouild construct this using something like: +-- +-- @ +-- 'runPipe' $ someSource +-- '.|' somePipe +-- .| someOtherPipe +-- .| someSink +-- @ +-- +-- 'runPipe' will produce the result of that final sink. +-- +-- Some common errors you might receive: +-- +-- * @i@ is not @()@: If you give a pipe where the first parameter +-- ("input") is not @()@, it means that your pipe is not a producer. +-- Pre-compose it (using '.|') with a producer of the type you need. +-- +-- For example, if you have a @myPipe :: 'Pipe' 'Int' o u m a@, this is +-- a pipe that is awaiting 'Int's from upstream. Pre-compose with +-- a producer of 'Int's, like @'Data.Conduino.Combinators.sourceList' +-- [1,2,3] '.|' myPipe@, in order to be able to run it. +-- +-- * @o@ is not 'Void': If you give a pipe where the second parameter +-- ("output") is not 'Void', it means that your pipe is not a consumer. +-- Post-compose it (using '.|') with a consumer of the type you need. +-- +-- For example, if you have @myPipe :: 'Pipe' i 'Int' u m a@, this is +-- a pipe that is yielding 'Int's downstream that are going unhandled. +-- Post-compose it a consumer of 'Int's, like @myPipe '.|' +-- 'Data.Conduino.foldl' (+) 0@, in order to be able to run it. +-- +-- If you just want to ignore all downstream yields, post-compose with +-- 'Data.Conduino.Combinators.sinkNull'. +-- runPipe :: Monad m => Pipe () Void u m a -> m a runPipe = iterT go . pipeFree where go = \case - PAwaitF f _ -> f () + PAwaitF _ f -> f () PYieldF o _ -> absurd o --- can this be done without going through FreeT? +-- | 'runPipe' when the underlying monad is 'Identity', and so has no +-- effects. +runPipePure :: Pipe () Void Void Identity a -> a +runPipePure = runIdentity . runPipe + +-- | The main operator for chaining pipes together. @pipe1 .| pipe2@ will +-- connect the output of @pipe1@ to the input of @pipe2@. +-- +-- "Running" a pipe will draw from @pipe2@, and if @pipe2@ ever asks for +-- input (with 'await' or something similar), it will block until @pipe1@ +-- outputs something (or signals termination). +-- +-- The structure of a full pipeline usually looks like: +-- +-- @ +-- 'runPipe' $ someSource +-- '.|' somePipe +-- .| someOtherPipe +-- .| someSink +-- @ +-- +-- Where you route a source into a series of pipes, which eventually ends +-- up at a sink. 'runPipe' will then produce the result of that sink. (.|) :: Monad m => 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) +infixr 2 .| compPipe_ :: forall a b c u v m r. (Monad m) - => FreeT (PipeF a b u) m v - -> FreeT (PipeF b c v) m r - -> FreeT (PipeF a c u) m r -compPipe_ p q = FreeT $ runFreeT q >>= \case + => RecPipe a b u m v + -> RecPipe b c v m r + -> RecPipe a c u m r +compPipe_ p q = FreeT $ runFreeT q >>= \qq -> case qq of Pure x -> pure . Pure $ x - Free (PAwaitF f g) -> runFreeT p >>= \case - Pure x' -> runFreeT $ compPipe_ p (g x') - Free (PAwaitF f' g') -> pure . Free $ PAwaitF ((`compPipe_` q) . f') - ((`compPipe_` q) . g') - Free (PYieldF x' y') -> runFreeT $ compPipe_ y' (f x') + Free (PAwaitF f g) -> runFreeT p >>= \pp -> case pp of + Pure x' -> runFreeT $ compPipe_ (FreeT (pure pp)) (f x') + Free (PAwaitF f' g') -> pure . Free $ PAwaitF ((`compPipe_` FreeT (pure qq)) . f') + ((`compPipe_` FreeT (pure qq)) . g') + Free (PYieldF x' y') -> runFreeT $ compPipe_ y' (g x') Free (PYieldF x y) -> pure . Free $ PYieldF x (compPipe_ p y) -infixr 2 .| -unfoldP :: (b -> Maybe (a, b)) -> b -> Pipe i a u m () -unfoldP f = go - where - go z = case f z of - Nothing -> pure () - Just (x, z') -> yield x *> go z' +-- | A newtype wrapper over a source (@'Pipe' () o 'Void'@) that gives it an +-- alternative 'Applicative' and 'Alternative' instance, matching "ListT +-- done right". +-- +-- '<*>' will pair up each output that the sources produce: if you 'await' +-- a value from downstream, it will wait until both paired sources yield +-- before passing them on together. +-- +-- '<|>' will completely exhaust the first source before moving on to the +-- next source. +-- +-- 'ZipSource' is effectively equivalent to "ListT done right", the true +-- List Monad transformer. '<|>' is concatentation. You can use this type +-- with 'lift' to lift a yielding action and '<|>' to sequence yields to +-- implement the pattern described in +-- , +-- where you can write streaming producers in a polymorphic way, and have +-- it run with pipes, conduit, etc. +-- +-- The main difference is that its 'Applicative' instance ("zipping") is +-- different from the traditional 'Applicative' instance for 'ListT' +-- ("all combinations"). Effectively this becomes like a "zipping" +-- 'Applicative' instance for 'ListT'. +-- +-- If you want a 'Monad' (or 'Control.Monad.IO.Class.MonadIO') instance, +-- use 'ListT' instead, and convert using 'toListT'/'fromListT' or the +-- 'PipeList' pattern/constructor. +newtype ZipSource m a = ZipSource { getZipSource :: Pipe () a Void m () } -unfoldPForever :: (b -> (a, b)) -> b -> Pipe i a u m r -unfoldPForever f = go +-- | A source is equivalent to a 'ListT' producing a 'Maybe'; this pattern +-- synonym lets you treat it as such. It essentialyl wraps over 'toListT' +-- and 'fromListT'. +pattern PipeList :: Monad m => ListT m (Maybe a) -> Pipe () a u m () +pattern PipeList xs <- (toListT->xs) where - go z = yield x *> go z' - where - (x, z') = f z + PipeList xs = fromListT xs +{-# COMPLETE PipeList #-} -iterateP :: (a -> a) -> a -> Pipe i a u m r -iterateP f = unfoldPForever (join (,) . f) +instance Functor (ZipSource m) where + fmap f = ZipSource . mapOutput f . getZipSource -sourceList :: Foldable t => t a -> Pipe i a u m () -sourceList = traverse_ yield +instance Monad m => Applicative (ZipSource m) where + pure = ZipSource . yield + ZipSource (PipeList fs) <*> ZipSource (PipeList xs) = ZipSource . PipeList . fmap Just $ + uncurry ($) + <$> LT.zip (concatListT fs) (concatListT xs) -repeatM :: Monad m => m o -> Pipe i o u m u -repeatM x = go - where - go = (yield =<< lift x) *> go +concatListT :: Monad m => ListT m (Maybe a) -> ListT m a +concatListT xs = ListT $ next xs >>= \case + Nil -> pure Nil + Cons Nothing ys -> next (concatListT ys) + Cons (Just y) ys -> pure $ Cons y (concatListT ys) -awaitForever :: (i -> Pipe i o u m a) -> Pipe i o u m u -awaitForever f = go - where - go = awaitEither >>= \case - Left x -> f x *> go - Right x -> pure x +instance Monad m => Alternative (ZipSource m) where + empty = ZipSource $ pure () + ZipSource p <|> ZipSource q = ZipSource (p *> q) --- finishPipe --- :: u --- -> Pipe i o u m a --- -> Pipe i o Void m a +instance MonadTrans ZipSource where + lift = ZipSource . (yield =<<) . lift -mapP :: (a -> b) -> Pipe a b u m u -mapP f = awaitForever (yield . f) +-- | A source is essentially equivalent to 'ListT' producing a 'Maybe' +-- result. This converts it to the 'ListT' it encodes. +-- +-- See 'ZipSource' for a wrapper over 'Pipe' that gives the right 'Functor' +-- and 'Alternative' instances. +toListT + :: Applicative m + => Pipe () o u m () + -> ListT m (Maybe o) +toListT p = ListT $ runFT (pipeFree p) + (\_ -> pure Nil) + (\pNext -> \case + PAwaitF _ g -> pure $ Cons Nothing (ListT . pNext $ g ()) + PYieldF x y -> pure $ Cons (Just x) (ListT . pNext $ y ) + ) -mapMP :: Monad m => (a -> m b) -> Pipe a b u m u -mapMP f = awaitForever ((yield =<<) . lift . f) +-- | A source is essentially 'ListT' producing a 'Maybe' result. This +-- converts a 'ListT' to the source it encodes. +-- +-- See 'ZipSource' for a wrapper over 'Pipe' that gives the right 'Functor' +-- and 'Alternative' instances. +fromListT + :: Monad m + => ListT m (Maybe o) + -> Pipe i o u m () +fromListT = fromRecPipe . go + where + go xs = FreeT $ next xs >>= \case + Nil -> pure . Pure $ () + Cons Nothing ys -> pure . Free $ PAwaitF (\_ -> pure ()) $ \_ -> go ys + Cons (Just y) ys -> pure . Free $ PYieldF y (go ys) -dropP :: Int -> Pipe i o u m () -dropP n = replicateM_ n await +---- | A source is essentially equiavlent to 'ListT'. This converts +---- a 'ListT' to the source it encodes. +---- +---- See 'ZipSource' for a wrapper over 'Pipe' that gives the right 'Functor' +---- and 'Alternative' instances. +--fromListT +-- :: Monad m +-- => ListT m o +-- -> Pipe i o u m () +--fromListT = fromRecPipe . go +-- where +-- go xs = FreeT $ next xs >>= \case +-- Nil -> pure . Pure $ () +-- Cons y ys -> pure . Free $ PYieldF y (go ys) -foldrP :: (a -> b -> b) -> b -> Pipe a Void u m b -foldrP f z = go - where - go = await >>= \case - Nothing -> pure z - Just x -> f x <$> go +-- | Given a "generator" of @o@ in @m@, return a /source/ that that +-- generator encodes. Is the inverse of 'withSource'. +-- +-- The generator is essentially a church-encoded 'ListT'. +genSource + :: (forall r. (Maybe (o, m r) -> m r) -> m r) + -> Pipe i o u m () +genSource f = Pipe $ FT $ \pDone pFree -> f $ \case + Nothing -> pDone () + Just (x, xs) -> pFree id (PYieldF x xs) -sinkList :: Pipe i Void u m [i] -sinkList = foldrP (:) [] +-- | A source can be "run" by providing a continuation to handle and +-- sequence each of its outputs. Is ths inverse of 'genSource'. +-- +-- This essentially turns a pipe into a church-encoded 'ListT'. +withSource + :: 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) + (\_ -> f Nothing) + (\pNext -> \case + PAwaitF _ g -> pNext $ g () + PYieldF x y -> f (Just (x, pNext y)) + ) +-- | 'ZipSource' is effectively 'ListT' returning a 'Maybe'. As such, you +-- can use 'unconsZipSource' to "peel off" the first yielded item, if it +-- exists, and return the "rest of the list". +unconsZipSource + :: Monad m + => ZipSource m a + -> m (Maybe (Maybe a, ZipSource m a)) +unconsZipSource (ZipSource (PipeList p)) = next p <&> \case + Cons x xs -> Just (x, ZipSource (PipeList xs)) + Nil -> Nothing + +-- | A newtype wrapper over a sink (@'Pipe' i 'Void'@) that gives it an +-- alternative 'Applicative' and 'Alternative' instance. +-- +-- '<*>' will distribute input over both sinks, and output a final result +-- once both sinks finish. +-- +-- '<|>' will distribute input over both sinks, and output a final result +-- as soon as one or the other finishes. newtype ZipSink i u m a = ZipSink { getZipSink :: Pipe i Void u m a } deriving Functor zipSink_ :: Monad m - => FreeT (PipeF i Void u) m (a -> b) - -> FreeT (PipeF i Void u) m a - -> FreeT (PipeF i Void u) m b -zipSink_ p q = FreeT $ go <$> runFreeT p <*> runFreeT q - where - go = \case - Pure x -> \case - Pure x' -> Pure $ x x' - Free (PAwaitF f' g') -> Free $ PAwaitF (zipSink_ p . f') (zipSink_ p . g') - Free (PYieldF x' _ ) -> absurd x' - Free (PAwaitF f g) -> \case - Pure _ -> Free $ PAwaitF ((`zipSink_` q) . f) ((`zipSink_` q) . g) - Free (PAwaitF f' g') -> Free $ PAwaitF (zipSink_ <$> f <*> f') (zipSink_ <$> g <*> g') - Free (PYieldF x' _ ) -> absurd x' - Free (PYieldF x _) -> absurd x + => RecPipe i Void u m (a -> b) + -> RecPipe i Void u m a + -> RecPipe i Void u m b +zipSink_ p q = FreeT $ runFreeT p >>= \pp -> case pp of + Pure x -> runFreeT q >>= \case + Pure x' -> pure . Pure $ x x' + Free (PAwaitF f' g') -> pure . Free $ + PAwaitF (zipSink_ (FreeT (pure pp)) . f') + (zipSink_ (FreeT (pure pp)) . g') + Free (PYieldF x' _ ) -> absurd x' + Free (PAwaitF f g) -> runFreeT q >>= \qq -> case qq of + Pure _ -> pure . Free $ + PAwaitF ((`zipSink_` FreeT (pure qq)) . f) + ((`zipSink_` FreeT (pure qq)) . g) + Free (PAwaitF f' g') -> pure . Free $ + PAwaitF (zipSink_ <$> f <*> f') (zipSink_ <$> g <*> g') + Free (PYieldF x' _ ) -> absurd x' + Free (PYieldF x _) -> absurd x altSink_ :: Monad m - => FreeT (PipeF i Void u) m a - -> FreeT (PipeF i Void u) m a - -> FreeT (PipeF i Void u) m a -altSink_ p q = FreeT $ go <$> runFreeT p <*> runFreeT q - where - go = \case - Pure x -> \_ -> Pure x - Free (PAwaitF f g) -> \case - Pure x' -> Pure x' - Free (PAwaitF f' g') -> Free $ PAwaitF (altSink_ <$> f <*> f') (altSink_ <$> g <*> g') - Free (PYieldF x' _ ) -> absurd x' - Free (PYieldF x _) -> absurd x + => RecPipe i Void u m a + -> RecPipe i Void u m a + -> RecPipe i Void u m a +altSink_ p q = FreeT $ runFreeT p >>= \case + Pure x -> pure . Pure $ x + Free (PAwaitF f g) -> runFreeT q <&> \case + Pure x' -> Pure x' + Free (PAwaitF f' g') -> Free $ PAwaitF (altSink_ <$> f <*> f') (altSink_ <$> g <*> g') + Free (PYieldF x' _ ) -> absurd x' + Free (PYieldF x _) -> absurd x +-- | Distribute input to both sinks, and finishes with the final result +-- once both finish. +-- +-- Forms an identity with 'pure'. zipSink :: Monad m => Pipe i Void u m (a -> b) @@ -210,6 +435,8 @@ -> Pipe i Void u m b zipSink (Pipe p) (Pipe q) = Pipe $ toFT $ zipSink_ (fromFT p) (fromFT q) +-- | Distribute input to both sinks, and finishes with the result of +-- the one that finishes first. altSink :: Monad m => Pipe i Void u m a @@ -233,3 +460,6 @@ where go = forever await ZipSink p <|> ZipSink q = ZipSink $ altSink p q + +instance MonadTrans (ZipSink i u) where + lift = ZipSink . lift diff --git a/src/Data/Conduino/Combinators.hs b/src/Data/Conduino/Combinators.hs new file mode 100644 --- /dev/null +++ b/src/Data/Conduino/Combinators.hs @@ -0,0 +1,488 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} + +-- | +-- Module : Data.Conduino.Combinators +-- Copyright : (c) Justin Le 2019 +-- License : BSD3 +-- +-- Maintainer : justin@jle.im +-- Stability : experimental +-- Portability : non-portable +-- +-- A basic collection of base 'Pipe's that serve as a "prelude" for the +-- package. This module is meant to be imported qualified. +-- +-- > import qualified Data.Conduino.Combinators as C +-- +module Data.Conduino.Combinators ( + -- * Sources + -- ** Pure + -- *** Infinite + unfold + , iterate + , repeat + -- *** Finite + , unfoldMaybe + , unfoldEither + , iterateMaybe + , iterateEither + , sourceList + , replicate + -- ** Monadic + -- *** Infinite + , repeatM + -- *** Finite + , repeatMaybeM + , repeatEitherM + , replicateM + , sourceHandleLines + , stdinLines + , sourceHandle + , stdin + -- * Pipes + , map + , mapM + , scan + , mapAccum + , take + , takeWhile + , filter + , concatMap + , concat + , pairs + , consecutive + -- * Sinks + -- ** Pure + , drop + , dropWhile + , foldr + , foldl + , foldMap + , fold + , sinkNull + , sinkList + , last + -- ** Monadic + , sinkHandle + , stdout + , stderr + ) where + +import Control.Applicative +import Control.Exception +import Control.Monad hiding (mapM, replicateM) +import Control.Monad.IO.Class +import Control.Monad.Trans.Class +import Control.Monad.Trans.Maybe +import Data.Conduino +import Data.Either +import Data.Foldable hiding (foldr, foldl, fold, concat, concatMap, foldMap) +import Data.Maybe +import Data.Semigroup +import Prelude hiding (map, iterate, mapM, replicate, repeat, foldr, drop, foldl, last, take, concatMap, filter, concat, takeWhile, dropWhile, foldMap) +import System.IO.Error +import qualified Data.ByteString as BS +import qualified Data.ByteString.Lazy.Internal as BSL +import qualified Data.Sequence as Seq +import qualified System.IO as S + +-- | A version of 'unfoldMaybe' that can choose the "result" value by +-- passing it in as 'Left'. +unfoldEither + :: (s -> Either a (o, s)) + -> s + -> Pipe i o u m a +unfoldEither f = go + where + go z = case f z of + Left r -> pure r + Right (x, z') -> yield x *> go z' + +-- | A version of 'unfold' that can terminate and end by returning +-- 'Nothing'. +unfoldMaybe + :: (s -> Maybe (o, s)) + -> s + -> Pipe i o u m () +unfoldMaybe f = unfoldEither (maybe (Left ()) Right . f) + +-- | Repeatedly apply an "unfolding" function to a given initial state, +-- yielding the first item in the tuple as output and updating the state as +-- the second item in the tuple. Goes on forever. See 'unfoldMaybe' for +-- a version that stops. +unfold + :: (s -> (o, s)) + -> s + -> Pipe i o u m a +unfold f = go + where + go z = yield x *> go z' + where + (x, z') = f z + +-- | A version of 'iterateMaybe' that can specify a result value by +-- providing it in the 'Left'. +iterateEither + :: (o -> Either a o) + -> o + -> Pipe i o u m a +iterateEither f = unfoldEither (fmap (join (,)) . f) + +-- | A version of 'iterate' that can choose to terminate and stop by +-- returning 'Nothing'. +iterateMaybe + :: (o -> Maybe o) + -> o + -> Pipe i o u m () +iterateMaybe f = unfoldMaybe (fmap (join (,)) . f) + +-- | Repeatedly apply a function to a given starting value and yield each +-- result forever. +-- +-- >>> runPipePure $ iterate succ 0 +-- .| take 5 +-- .| sinkList +-- +-- [1,2,3,4,5] +-- +-- This doesn't yield the original starting value. However, you can yield +-- it iterate after: +-- +-- >>> runPipePure $ (yield 0 >> iterate succ 0) +-- .| take 5 +-- .| sinkList +-- +-- [0,1,2,3,4,5] +iterate + :: (o -> o) + -> o + -> Pipe i o u m a +iterate f = unfold (join (,) . f) + +-- | Yield every item in a foldable container. +sourceList :: Foldable t => t a -> Pipe i a u m () +sourceList = traverse_ yield + +-- | Repeatedly yield a given item forever. +repeat :: o -> Pipe i o u m a +repeat = forever . yield + +-- | Yield a given item a certain number of times. +replicate :: Int -> o -> Pipe i o u m () +replicate n = replicateM_ n . yield + +-- | Like 'repeatMaybeM', but allow specification of a final result type. +repeatEitherM + :: Monad m + => m (Either a o) + -> Pipe i o u m a +repeatEitherM x = go + where + go = lift x >>= \case + Left r -> pure r + Right y -> yield y *> go + +-- | Repeat a monadic action, yielding the item in the 'Just' every time. +-- As soon as it sees 'Nothing', stop producing forever. +-- +-- Remember that each item will only be "executed" when something +-- downstream requests output. +repeatMaybeM + :: Monad m + => m (Maybe o) + -> Pipe i o u m () +repeatMaybeM = repeatEitherM . fmap (maybe (Left ()) Right) + +-- | Repeat a monadic action a given number of times, yielding each result, +-- and then stop producing forever. +-- +-- Remember that each item will only be "executed" when something +-- downstream requests output. +replicateM + :: Monad m + => Int + -> m o + -> Pipe i o u m () +replicateM n x = replicateM_ n $ lift x >>= yield + +-- | 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 + +-- | 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 + +-- | 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. +sourceHandleLines + :: MonadIO m + => S.Handle + -> Pipe i String u m () +sourceHandleLines h = repeatMaybeM $ do + d <- liftIO $ S.hIsEOF h + if d + then pure Nothing + else liftIO . catchJust + (guard . isEOFError) + (mfilter (not . null) . Just <$> S.hGetLine h) + $ \_ -> pure Nothing + +-- | 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'. +sourceHandle + :: MonadIO m + => S.Handle + -> Pipe i BS.ByteString u m () +sourceHandle h = repeatMaybeM + . fmap (mfilter (not . BS.null) . Just) + . liftIO + $ BS.hGetSome h BSL.defaultChunkSize + +-- | Sink into a given I/O handle, writing each input to the handle. +sinkHandle + :: MonadIO m + => S.Handle + -> Pipe BS.ByteString o u m () +sinkHandle h = mapM (liftIO . BS.hPut h) + .| sinkNull + +-- | A sink into stdout. +stdout :: MonadIO m => Pipe BS.ByteString o u m () +stdout = sinkHandle S.stdout + +-- | A sink into stderr. +stderr :: MonadIO m => Pipe BS.ByteString o u m () +stderr = sinkHandle S.stderr + +-- | Repeat a monadic action forever, yielding each output. +-- +-- Remember that each item will only be "executed" when something +-- downstream requests output. +repeatM + :: Monad m + => m o + -> Pipe i o u m a +repeatM x = go + where + go = (yield =<< lift x) *> go + +-- | 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) + +-- | 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) + +-- | Map a pure "stateful" function over each incoming item. Give +-- a function to update the state and return an output and an initial +-- state. +mapAccum + :: (i -> s -> (s, o)) -- ^ update state and output + -> s -- ^ initial state + -> Pipe i o u m u +mapAccum f = go + where + go !x = awaitWith $ \y -> + let (!x', !z) = f y x + in yield z *> go x' + +-- | Like 'foldl', but yields every accumulator value downstream. +-- +-- >>> runPipePure $ sourceList [1..10] +-- .| scan (+) 0 +-- .| sinkList +-- [1,3,6,10,15,21,28,36,45,55] +-- @ +scan + :: (o -> i -> o) + -> o + -> Pipe i o u m u +scan f = go + where + go !x = awaitWith $ \y -> + let x' = f x y + in yield x' *> go x' + +-- | Yield consecutive pairs of values. +-- +-- >>> runPipePure $ sourceList [1..5] +-- .| pairs +-- .| sinkList +-- [(1,2),(2,3),(3,4),(4,5)] +pairs :: Pipe i (i, i) u m u +pairs = awaitWith go + where + go x = awaitWith $ \y -> do + yield (x, y) + go y + +-- | Yield consecutive runs of at most @n@ of values, starting with an +-- empty sequence. +-- +-- To get only "full" sequences, pipe with 'filter'. +-- +-- >>> runPipePure $ sourceList [1..6] +-- .| consecutive 3 +-- .| map toList +-- .| sinkList +-- [[],[1],[1,2],[1,2,3],[2,3,4],[3,4,5],[4,5,6]] +-- +-- >>> runPipePure $ sourceList [1..6] +-- .| consecutive 3 +-- .| filter ((== 3) . Seq.length) +-- .| map toList +-- .| sinkList +-- [[1,2,3],[2,3,4],[3,4,5],[4,5,6]] +consecutive :: Int -> Pipe i (Seq.Seq i) u m u +consecutive n = go Seq.empty + where + go xs = do + yield xs + awaitWith $ \y -> go . Seq.drop (Seq.length xs - n + 1) $ (xs Seq.:|> y) + + +-- | Let a given number of items pass through the stream uninhibited, and +-- then stop producing forever. +-- +-- This is most useful if you sequence a second conduit after it. +-- +-- >>> runPipePure $ sourceList [1..8] +-- .| (do take 3 .| map (*2) -- double the first 3 items +-- map negate -- negate the rest +-- ) +-- .| sinkList +-- [2,4,6,-4,-5,-6,-7,-8] +take :: Int -> Pipe i i u m () +take n = void . runMaybeT . replicateM_ n $ + lift . yield =<< MaybeT await + +-- | Let elements pass until an element is received that does not satisfy +-- the predicate, then stop producing forever. +-- +-- Like 'take', is most useful if you sequence a second conduit after it. +takeWhile :: (i -> Bool) -> Pipe i i u m () +takeWhile p = go + where + go = await >>= \case + Nothing -> pure () + Just x + | p x -> yield x *> go + | otherwise -> pure () + +-- | 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 + +-- | Map a function returning a container onto every incoming item, and +-- yield all of the outputs from that function. +concatMap + :: Foldable t + => (i -> t o) + -> Pipe i o u m u +concatMap f = awaitForever (sourceList . f) + +-- | 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 + +-- | Right-fold every input into an accumulated value. +-- +-- Essentially this builds up a giant continuation that will be run all at +-- once on the final result. +foldr :: (a -> b -> b) -> b -> Pipe a o u m b +foldr f z = go + where + go = await >>= \case + Nothing -> pure z + Just x -> f x <$> go + +-- | Left-fold every input into an accumulated value. +-- +-- Essentially this maintains a state and modifies that state with every +-- input, using the given accumulating function. +foldl :: (b -> a -> b) -> b -> Pipe a o u m b +foldl f = go + where + go !z = await >>= \case + Nothing -> pure z + Just !x -> go (f z x) + +-- | Fold every incoming item monoidally, and return the result once +-- finished. +fold :: Monoid a => Pipe a o u m a +fold = foldl (<>) mempty + +-- | Fold every incoming item according to a monoidal projection, and +-- return the result once finished. +-- +-- This can be used to implement many useful consumers, like ones that find +-- the sum or the maximum item: +-- +-- @ +-- sum :: Num i => Pipe i o u m i +-- sum = getSum <$> foldMap Sum +-- +-- maximum :: Ord i => Pipe i o u m (Maybe i) +-- maximum = fmap getMax <$> foldMap (Just . Max) +-- @ +foldMap :: Monoid a => (i -> a) -> Pipe i o u m a +foldMap f = foldl (\x y -> x <> f y) mempty + +-- | Sink every incoming item into a list. +-- +-- Note that this keeps the entire list in memory until it is all +-- eventually read. +sinkList :: Pipe i o u m [i] +sinkList = foldr (:) [] + +-- | Ignore a certain amount of items from the input stream, and then stop +-- producing forever. +-- +-- This is most useful if you sequence a second consumer after it: +-- +-- >>> runPipePure $ sourceList [1..8] +-- .| (drop 3 >> 'sinkList') +-- [4,5,6,7,8] +drop :: Int -> Pipe i o u m () +drop n = replicateM_ n await + +-- | Ignore items from an input stream as long as they match a predicate. +-- Afterwards, stop producing forever. +-- +-- Like for 'drop', is most useful of you sequence a second consumer after +-- it. +dropWhile + :: (i -> Bool) + -> Pipe i o u m () +dropWhile p = go + where + go = await >>= \case + Nothing -> pure () + Just x + | p x -> go + | otherwise -> pure () + +-- | Consume an entire input stream and ignore all of its outputs. +sinkNull :: Pipe i o u m () +sinkNull = await >>= \case + Nothing -> pure () + Just _ -> sinkNull + +-- | Get the last item emitted by a stream. +-- +-- To get the first item ("head"), use 'await' or 'awaitSurely'. +last :: Pipe i o u m (Maybe i) +last = fmap getLast <$> foldMap (Just . Last) diff --git a/src/Data/Conduino/Internal.hs b/src/Data/Conduino/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Data/Conduino/Internal.hs @@ -0,0 +1,223 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DeriveFunctor #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeInType #-} +{-# LANGUAGE TypeOperators #-} +{-# OPTIONS_HADDOCK not-home #-} + +-- | +-- Module : Data.Conduino.Internal +-- Copyright : (c) Justin Le 2019 +-- License : BSD3 +-- +-- Maintainer : justin@jle.im +-- Stability : experimental +-- Portability : non-portable +-- +-- Internal module exposing the internals of 'Pipe', including its +-- underlying representation and base functor. +-- +module Data.Conduino.Internal ( + Pipe(..) + , PipeF(..) + , awaitEither + , yield + , trimapPipe, mapInput, mapOutput, mapUpRes + , hoistPipe + , RecPipe + , toRecPipe, fromRecPipe + ) where + +import Control.Monad.Except +import Control.Monad.Free.Class +import Control.Monad.Free.TH +import Control.Monad.RWS +import Control.Monad.Trans.Free (FreeT(..)) +import Control.Monad.Trans.Free.Church + +#if !MIN_VERSION_base(4,13,0) +import Control.Monad.Fail +#endif + +-- | Base functor of 'Pipe'. +-- +-- A pipe fundamentally has the ability to await and the ability to yield. +-- The other functionality are implemented. +-- +-- * Lifting effects is implemented by the 'MonadTrans' and 'MonadIO' +-- instances that 'FT' gives. +-- * /Ending/ with a result is implemented by the 'Applicative' instance's +-- 'pure' that 'FT' gives. +-- * Applicative and monadic sequenceing "after a pipe is done" is +-- implemented by the 'Applicative' and 'Monad' instances that 'FT' +-- gives. +-- +-- On top of these we implement 'Data.Conduino..|' and other combinators +-- based on the structure that 'FT' gives. For some functions, it can be +-- easier to use an alternative encoding, 'RecPipe', which is the same +-- thing but explicitly recursive. +data PipeF i o u a = + PAwaitF (u -> a) (i -> a) + | PYieldF o a + deriving Functor + +makeFree ''PipeF + +-- | Similar to a conduit from the /conduit/ package. +-- +-- For a @'Pipe' i o u m a@, you have: +-- +-- * @i@: Type of input stream (the things you can 'Data.Conduino.await') +-- * @o@: Type of output stream (the things you 'yield') +-- * @u@: Type of the /result/ of the upstream pipe (Outputted when +-- upstream pipe terminates) +-- * @m@: Underlying monad (the things you can 'lift') +-- * @a@: Result type when pipe terminates (outputted when finished, with +-- 'pure' or 'return') +-- +-- Some specializations: +-- +-- * If @i@ is @()@, the pipe is a /source/ --- it doesn't need anything +-- to produce items. It will pump out items on its own, for pipes +-- downstream to receive and process. +-- +-- * If @o@ is 'Void', the pipe is a /sink/ --- it will never 'yield' +-- anything downstream. It will consume items from things upstream, and +-- produce a result (@a@) if and when it terminates. +-- +-- * If @u@ is 'Void', then the pipe's upstream is limitless, and never +-- terminates. This means that you can use 'Data.Condunio.awaitSurely' +-- instead of 'Data.Conduino.await', to get await a value that is +-- guaranteed to come. You'll get an @i@ instead of a @'Maybe' i@. +-- +-- * If @a@ is 'Void', then the pipe never terminates --- it will keep on +-- consuming and/or producing values forever. If this is a sink, it +-- means that the sink will never terminate, and so +-- 'Data.Condunio.runPipe' will also never terminate. If it is +-- a source, it means that if you chain something downstream with +-- 'Data.Condunio..|', that downstream pipe can use 'awaitSurely' to +-- guarantee something being passed down. +-- +-- Applicative and Monadic sequencing of pipes chains by exhaustion. +-- +-- @ +-- do pipeX +-- pipeY +-- pipeZ +-- @ +-- +-- is a pipe itself, that behaves like @pipeX@ until it terminates, then +-- @pipeY@ until it terminates, then @pipeZ@ until it terminates. The +-- 'Monad' instance allows you to choose "which pipe to behave like next" +-- based on the terminating result of a previous pipe. +-- +-- @ +-- do x <- pipeX +-- pipeBasedOn x +-- @ +-- +-- Usually you would use it by chaining together pipes with +-- 'Data.Condunio..|' and then running the result with +-- 'Data.Condunio.runPipe'. +-- +-- @ +-- 'Data.Conduino.runPipe' $ someSource +-- 'Data.Conduino..|' somePipe +-- .| someOtherPipe +-- .| someSink +-- @ +-- +-- See 'Data.Condunio..|' and 'Data.Condunio.runPipe' for more information +-- on usage. +-- +-- For a "prelude" of commonly used 'Pipe's, see +-- "Data.Condunio.Combinators". +-- +newtype Pipe i o u m a = Pipe { pipeFree :: FT (PipeF i o u) m a } + deriving + ( Functor + , Applicative + , Monad + , MonadTrans + , MonadFree (PipeF i o u) + , MonadIO + , MonadState s + , MonadReader r + , MonadWriter w + , MonadError e + , MonadRWS r w s + ) + +instance MonadFail m => MonadFail (Pipe i o u m) where +#if MIN_VERSION_base(4,13,0) + fail = lift . fail +#else + fail = lift . Control.Monad.Fail.fail +#endif + +-- | Await on upstream output. Will block until it receives an @i@ +-- (expected input type) or a @u@ if the upstream pipe terminates. +awaitEither :: Pipe i o u m (Either u i) +awaitEither = pAwaitF + +-- | Send output downstream. +yield :: o -> Pipe i o u m () +yield = pYieldF + +-- | Map over the input type, output type, and upstream result type. +-- +-- If you want to map over the result type, use 'fmap'. +trimapPipe + :: (i -> j) + -> (p -> o) + -> (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 + +-- | Transform the underlying monad of a pipe. +hoistPipe + :: (Monad m, Monad n) + => (forall x. m x -> n x) + -> Pipe i o u m a + -> Pipe i o u n a +hoistPipe f = Pipe . hoistFT f . pipeFree + +-- | (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 + +-- | 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 + +-- | (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 + +-- | A version of 'Pipe' that uses explicit, concrete recursion instead of +-- church-encoding like 'Pipe'. Some functions --- especially ones that +-- combine multiple pipes into one --- are easier to implement in this +-- form. +type RecPipe i o u = FreeT (PipeF i o u) + +-- | Convert from a 'Pipe' to a 'RecPipe'. While most of this library is +-- 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 + +-- | 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 diff --git a/src/Lib.hs b/src/Lib.hs deleted file mode 100644 --- a/src/Lib.hs +++ /dev/null @@ -1,6 +0,0 @@ -module Lib - ( someFunc - ) where - -someFunc :: IO () -someFunc = putStrLn "someFunc"