diff --git a/quiver.cabal b/quiver.cabal --- a/quiver.cabal +++ b/quiver.cabal @@ -1,5 +1,5 @@ name: quiver -version: 0.0.0.1 +version: 0.0.0.2 synopsis: Quiver finite stream processing library homepage: https://github.com/zadarnowski/quiver category: Control @@ -13,9 +13,16 @@ description: /Quiver/ is a powerful stream processing library for - combinatorial and monadic transformations of both inductive - and coinductive data structures. + combinatorial and monadic representation of computations + over both inductive and coinductive data streams. + It is similar to Gabriel Gonzalez's /pipes/ and + Michael Snoyman's /conduit/, but generalises both + with support for functor-based computations and + a clean support for finite (i.e., inductive) data + streams, both upstream and downstream of the computation + being defined. + cabal-version: >= 1.18 build-type: Simple license: BSD3 @@ -28,7 +35,7 @@ source-repository this type: git location: https://github.com/zadarnowski/quiver.git - tag: 0.0.0.1 + tag: 0.0.0.2 library hs-source-dirs: src diff --git a/src/Control/Quiver.lhs b/src/Control/Quiver.lhs --- a/src/Control/Quiver.lhs +++ b/src/Control/Quiver.lhs @@ -1,12 +1,25 @@ +> -- | Module: Control.Quiver +> -- Description: Core Quiver definitions +> -- Copyright: © 2015 Patryk Zadarnowski +> -- License: BSD3 +> -- Maintainer: pat@jantar.org +> -- Stability: experimental +> -- Portability: portable +> -- +> -- This module provides the core types and combinators +> -- of the Quiver stream processing library. + > {-# LANGUAGE RankNTypes, TupleSections #-} > module Control.Quiver ( -> Q, +> -- Imported from @Control.Quiver.Internal@: +> P, Consumer, Producer, Effect, > consume, produce, enclose, deliver, -> decouple, evacuate, +> decouple, deplete, +> -- Defined below: > fetch, fetch', > emit, emit', emit_, -> liftQ, +> liftP, > (>>->), (>->>), > ) where @@ -14,50 +27,91 @@ > infixl 0 >>->, >->> -> fetch :: a' -> Q a' a b' b f (Maybe a) -> fetch x = Consume x (Deliver . Just) (Deliver Nothing) +> -- | @fetch x@ represents a singleton stream processor that +> -- sends the request value @x@ upstream and delivers the +> -- next input value received, or @Nothing@ if the upstream +> -- processor has been depleted. -> fetch' :: a' -> (forall x' x . Q x' x b' b f a) -> Q a' a b' b f a -> fetch' x r = Consume x Deliver r +> fetch :: a' -> P a' a b' b f (Maybe a) +> fetch x = consume x (deliver . Just) (deliver Nothing) -> emit :: b -> Q a' a b' b f (Maybe b') -> emit y = Produce y (Deliver . Just) (Deliver Nothing) +> -- | @fetch' x q@ represents a singleton stream processor that +> -- sends the request value @x@ upstream and delivers the next +> -- input value received, or, if the upstream processor has +> -- been depleted, continues with the decoupled processor @q@. -> emit' :: b -> (forall x' x . Q a' a x' x f b') -> Q a' a b' b f b' -> emit' y r = Produce y Deliver r +> fetch' :: a' -> Producer b' b f a -> P a' a b' b f a +> fetch' x q = consume x deliver q -> emit_ :: b -> Q a' a b' b f () -> emit_ y = Produce y (Deliver . const ()) (Deliver ()) +> -- | @emit y@ represents a singleton stream processor that +> -- produces a single output value @y@ and delivers the +> -- response received from the downstream processor, or +> -- @Nothing@ if the downstream processor has been decoupled. -> liftQ :: Functor f => f c -> Q a' a b' b f c -> liftQ = Enclose . fmap Deliver +> emit :: b -> P a' a b' b f (Maybe b') +> emit y = produce y (deliver . Just) (deliver Nothing) -> (>>->) :: Functor f => Q a' a t' t f c1 -> Q t' t b' b f c2 -> Q a' a b' b f (c1, c2) -> (Consume x1 k1 r1) >>-> q2 = Consume x1 ((>>-> q2) . k1) (r1 >>-> q2) -> (Produce y1 k1 r1) >>-> q2 = loop q2 +> -- | @emit' y q@ represents a singleton stream processor that +> -- produces a single output value @y@ and delivers the +> -- response received from the downstream processor, or, +> -- if the downstream processor has been decoupled, continues +> -- with the depleted processor @q@. + +> emit' :: b -> Consumer a' a f b' -> P a' a b' b f b' +> emit' y q = produce y deliver q + +> -- | @emit' y q@ represents a singleton stream processor that +> -- produces a single output value @y@, ignoring any response +> -- received from the downstream processor. + +> emit_ :: b -> P a' a b' b f () +> emit_ y = produce y (deliver . const ()) (deliver ()) + +> -- | @liftP@ lifts the value of a base functor into a stream processor. + +> liftP :: Functor f => f r -> P a' a b' b f r +> liftP = enclose . fmap deliver + +> -- | The @>>->@ represents a push-based composition of stream processor. +> -- @p1 >>-> p2@ represents a stream processor that forwards the output +> -- of @p1@ to @p2@, delivering the result of both processors. +> -- The new processor is /driven/ by @p2@, so, if the base functor +> -- represents a non-commutative monad, any effects of @p2@ will be +> -- observed before those of @p1@. + +> (>>->) :: Functor f => P a' a b' b f r1 -> P b' b c' c f r2 -> P a' a c' c f (r1, r2) +> (Consume x1 k1 q1) >>-> p2 = consume x1 ((>>-> p2) . k1) (q1 >>-> p2) +> (Produce y1 k1 q1) >>-> p2 = loop p2 > where > loop (Consume x2 k2 _) = k1 x2 >>-> k2 y1 -> loop (Produce y2 k2 r2) = Produce y2 (loop . k2) (loop' r2) -> loop (Enclose f2) = Enclose (fmap loop f2) -> loop (Deliver z2) = fmap (, z2) r1 +> loop (Produce y2 k2 q2) = produce y2 (loop . k2) (loop' q2) +> loop (Enclose f2) = enclose (fmap loop f2) +> loop (Deliver r2) = fmap (, r2) q1 > loop' (Consume x2 k2 _) = k1 x2 >>-> k2 y1 -> loop' (Produce _ _ r2) = loop' r2 -> loop' (Enclose f2) = Enclose (fmap loop' f2) -> loop' (Deliver z2) = fmap (, z2) r1 -> (Enclose f1) >>-> q2 = Enclose (fmap (>>-> q2) f1) -> (Deliver z1) >>-> q2 = fmap (z1 ,) (decouple q2) +> loop' (Produce _ _ q2) = loop' q2 +> loop' (Enclose f2) = enclose (fmap loop' f2) +> loop' (Deliver r2) = fmap (, r2) q1 +> (Enclose f1) >>-> p2 = enclose (fmap (>>-> p2) f1) +> (Deliver r1) >>-> p2 = fmap (r1 ,) (decouple p2) -> (>->>) :: Functor f => Q a' a t' t f c1 -> Q t' t b' b f c2 -> Q a' a b' b f (c1, c2) -> q1 >->> (Consume x2 k2 r2) = loop q1 +> -- | The @>->>@ represents a pull-based composition of stream processor. +> -- @p1 >->> p2@ represents a stream processor that forwards the output +> -- of @p1@ to @p2@, delivering the result of both processors. +> -- The new processor is /driven/ by @p1@, so, if the base functor +> -- represents a non-commutative monad, any effects of @p1@ will be +> -- observed before those of @p2@. + +> (>->>) :: Functor f => P a' a b' b f r1 -> P b' b c' c f r2 -> P a' a c' c f (r1, r2) +> p1 >->> (Consume x2 k2 q2) = loop p1 > where -> loop (Consume x1 k1 r1) = Consume x1 (loop . k1) (loop' r1) +> loop (Consume x1 k1 q1) = consume x1 (loop . k1) (loop' q1) > loop (Produce y1 k1 _) = k1 x2 >->> k2 y1 -> loop (Enclose f1) = Enclose (fmap loop f1) -> loop (Deliver z1) = fmap (z1 ,) r2 -> loop' (Consume _ _ t1) = loop' t1 +> loop (Enclose f1) = enclose (fmap loop f1) +> loop (Deliver r1) = fmap (r1 ,) q2 +> loop' (Consume _ _ q1) = loop' q1 > loop' (Produce y1 k1 _) = k1 x2 >->> k2 y1 -> loop' (Enclose f1) = Enclose (fmap loop' f1) -> loop' (Deliver z1) = fmap (z1 ,) r2 -> q1 >->> (Produce y2 k2 r2) = Produce y2 ((q1 >->>) . k2) (q1 >->> r2) -> q1 >->> (Enclose f2) = Enclose (fmap (q1 >->>) f2) -> q1 >->> (Deliver z2) = fmap (, z2) (evacuate q1) +> loop' (Enclose f1) = enclose (fmap loop' f1) +> loop' (Deliver r1) = fmap (r1 ,) q2 +> p1 >->> (Produce y2 k2 q2) = produce y2 ((p1 >->>) . k2) (p1 >->> q2) +> p1 >->> (Enclose f2) = enclose (fmap (p1 >->>) f2) +> p1 >->> (Deliver r2) = fmap (, r2) (deplete p1) diff --git a/src/Control/Quiver/Internal.lhs b/src/Control/Quiver/Internal.lhs --- a/src/Control/Quiver/Internal.lhs +++ b/src/Control/Quiver/Internal.lhs @@ -1,56 +1,201 @@ +> -- | Module: Control.Quiver.Internal +> -- Description: Common definitions +> -- Copyright: © 2015 Patryk Zadarnowski +> -- License: BSD3 +> -- Maintainer: pat@jantar.org +> -- Stability: experimental +> -- Portability: portable +> -- +> -- This module provides a host of common definitions, +> -- including the main Quiver /processor/ type @P@, +> -- that are reexported by other Quiver modules as +> -- required. +> -- +> -- This is the only module in the Quiver library that +> -- exposes the actual four constructors of the stream +> -- processor type @P@, allowing for definition of low +> -- level stream processor transformations, such as +> -- conversions between @P@ and other stream processing +> -- libraries. +> -- +> -- As a matter of style, Quiver users should strive to +> -- avoid explicit pattern matching on the @P@ type and +> -- rely instead on the various high level combinators +> -- exported elsewhere, in order to improve chances of +> -- successful deforestation by the various Quiver +> -- rewrite rules. + > {-# LANGUAGE RankNTypes, TupleSections #-} > module Control.Quiver.Internal ( -> Q (..), +> P (..), Producer, Consumer, Effect, > consume, produce, enclose, deliver, -> decouple, evacuate +> decouple, deplete, > ) where -> data Q a' a b' b f c = -> Consume a' (a -> Q a' a b' b f c) (forall x' x . Q x' x b' b f c) -> | Produce b (b' -> Q a' a b' b f c) (forall x' x . Q a' a x' x f c) -> | Enclose (f (Q a' a b' b f c)) -> | Deliver c -> consume :: a' -> (a -> Q a' a b' b f c) -> (forall x' x . Q x' x b' b f c) -> Q a' a b' b f c + Data Types + ========== + +> -- | The main Quiver /stream processor/ type @P a' a b' b f r@, +> -- representing a producer/consumer structure with /bidirectional/, +> -- /bounded/ communication on both the upstream (consumer) and +> -- downstream (producer) channel. The six type parameters have +> -- the following intuitive meaning: +> -- +> -- * @a'@ is the type of a /request/ values sent by the stream +> -- processor to its upstream partner in order to receive the +> -- next element of the input stream. +> -- +> -- * @a@ is the type of the actual information being consumed +> -- by this stream processor (i.e., its input stream.) +> -- +> -- * @b'@ is the type of the /response/ values received from +> -- the downstream partner for each elemnet of the output +> -- stream produced by this stream processor. +> -- +> -- * @f@ is the type of the stream processor's /base functor/; +> -- usually this is a monad used for stateful stream processing, +> -- exception handling and/or real-world interaction. +> -- +> -- * @r@ is the stream processor's /delivery type/, used for +> -- monadic stream processor definition. +> -- +> -- Every stream processor is a functor over its delivery type. +> -- However, if the base functor @f@ meets the additional requirements +> -- of 'Applicative' or 'Monad', so will the stream processor itself. +> -- Note that, unlike most other stream processing libraries, @f@ +> -- is not required to be a monad in most applications, although +> -- only time will tell whether this generalisation has useful +> -- applications in the real world. + +> data P a' a b' b f r = + +> -- | @Consume x k q@ represents a /consumer step/, in which +> -- the request @x@ is sent upstream and the returned input +> -- value is supplied to the /continuation processor/ @k@, +> -- or, if the upstream partner has been /depleted/ (i.e., +> -- delivered its ultimate result, hence reaching the end +> -- of processing), to the /decoupled continuation/ @q@. + +> Consume a' (a -> P a' a b' b f r) (Producer b' b f r) | + +> -- | @Produce y k q@ represent a /producer step/, in which +> -- the output value @y@ is sent downstream, and the returned +> -- acknowledgement is supplied to the /continuation processor/ +> -- @k@, or, if the downstream partner has been /decoupled/ +> -- (i.e., delivered its ultimate result, hence reaching the end +> -- of processing), to the /depleted continuation/ @q@. + +> Produce b (b' -> P a' a b' b f r) (Consumer a' a f r) | + +> -- | @Enclose@ allows for selective application of the base +> -- functor @f@ the the remainder of the computation. + +> Enclose (f (P a' a b' b f r)) | + +> -- | @Deliver r@ completes processing of information, delivering +> -- its ultimate result @r@. + +> Deliver r + +> -- | A Quiver /producer/, represented by a stream processor +> -- with unspecified input types. + +> type Producer b' b f r = forall a' a . P a' a b' b f r + +> -- | A Quiver /consumer/, represented by a stream processor +> -- with unspecified output types. + +> type Consumer a' a f r = forall b' b . P a' a b' b f r + +> -- | A Quiver /effect/, represented by a stream processor +> -- with unspecified input and output types. + +> type Effect f r = forall a' a b' b . P a' a b' b f r + + + Instances + ========= + +> instance Functor f => Functor (P a' a b' b f) where +> fmap ff (Consume x k q) = Consume x (fmap ff . k) (fmap ff q) +> fmap ff (Produce y k q) = Produce y (fmap ff . k) (fmap ff q) +> fmap ff (Enclose f) = Enclose (fmap (fmap ff) f) +> fmap ff (Deliver r) = Deliver (ff r) + +> instance Applicative f => Applicative (P a' a b' b f) where +> pure = Deliver +> (Consume x k q) <*> p = Consume x ((<*> p) . k) (q <*> decouple p) +> (Produce y k q) <*> p = Produce y ((<*> p) . k) (q <*> deplete p) +> (Enclose f) <*> p = Enclose (fmap (<*> p) f) +> (Deliver r) <*> p = fmap r p + +> instance Monad f => Monad (P a' a b' b f) where +> (Consume x k q) >>= kk = Consume x ((>>= kk) . k) (q >>= decouple . kk) +> (Produce y k q) >>= kk = Produce y ((>>= kk) . k) (q >>= deplete . kk) +> (Enclose f) >>= kk = Enclose (fmap (>>= kk) f) +> (Deliver r) >>= kk = kk r + + + Primitive Combinators + ===================== + +> -- | @consume x k q@ represents a /consumer step/, in which +> -- the request @x@ is sent upstream and the returned input +> -- value is supplied to the /continuation processor/ @k@, +> -- or, if the upstream partner has been /depleted/ (i.e., +> -- delivered its ultimate result, hence reaching the end +> -- of processing), to the /decoupled continuation/ @q@. + +> consume :: a' -> (a -> P a' a b' b f r) -> Producer b' b f r -> P a' a b' b f r > consume = Consume -> produce :: b -> (b' -> Q a' a b' b f c) -> (forall x' x . Q a' a x' x f c) -> Q a' a b' b f c +> -- | @produce y k q@ represent a /producer step/, in which +> -- the output value @y@ is sent downstream, and the returned +> -- acknowledgement is supplied to the /continuation processor/ +> -- @k@, or, if the downstream partner has been /decoupled/ +> -- (i.e., delivered its ultimate result, hence reaching the end +> -- of processing), to the /depleted continuation/ @q@. + +> produce :: b -> (b' -> P a' a b' b f r) -> Consumer a' a f r -> P a' a b' b f r > produce = Produce -> enclose :: f (Q a' a b' b f c) -> Q a' a b' b f c +> -- | @enclose@ allows for selective application of the base +> -- functor @f@ the the remainder of the computation. + +> enclose :: f (P a' a b' b f r) -> P a' a b' b f r > enclose = Enclose -> deliver :: c -> Q a' a b' b f c +> -- | @deliver r@ completes processing of information, delivering +> -- its ultimate result @r@. + +> deliver :: r -> P a' a b' b f r > deliver = Deliver -> instance Functor f => Functor (Q a' a b' b f) where -> fmap ff (Consume x k r) = Consume x (fmap ff . k) (fmap ff r) -> fmap ff (Produce y k r) = Produce y (fmap ff . k) (fmap ff r) -> fmap ff (Enclose f) = Enclose (fmap (fmap ff) f) -> fmap ff (Deliver z) = Deliver (ff z) -> instance Applicative f => Applicative (Q a' a b' b f) where -> pure = Deliver -> (Consume x k r) <*> q = Consume x ((<*> q) . k) (r <*> decouple q) -> (Produce y k r) <*> q = Produce y ((<*> q) . k) (r <*> evacuate q) -> (Enclose f) <*> q = Enclose (fmap (<*> q) f) -> (Deliver z) <*> q = fmap z q + Utilities + ========= -> instance Monad f => Monad (Q a' a b' b f) where -> (Consume x k r) >>= kk = Consume x ((>>= kk) . k) (r >>= decouple . kk) -> (Produce y k r) >>= kk = Produce y ((>>= kk) . k) (r >>= evacuate . kk) -> (Enclose f) >>= kk = Enclose (fmap (>>= kk) f) -> (Deliver z) >>= kk = kk z +> -- | @decouple p@ /decouples/ the stream processor @p@, by replacing +> -- the first consumer step in @p@ with that step's decoupled contination, +> -- effectively converting @p@ into a producer processor that no longer +> -- expects to receive any input. -> decouple :: Functor f => Q a' a b' b f c -> forall x' x . Q x' x b' b f c -> decouple (Consume _ _ r) = r -> decouple (Produce y k r) = Produce y (decouple . k) (decouple r) +> decouple :: Functor f => P a' a b' b f r -> Producer b' b f r +> decouple (Consume _ _ q) = q +> decouple (Produce y k q) = Produce y (decouple . k) (decouple q) > decouple (Enclose f) = Enclose (fmap decouple f) -> decouple (Deliver z) = Deliver z +> decouple (Deliver r) = Deliver r -> evacuate :: Functor f => Q a' a b' b f c -> forall x' x . Q a' a x' x f c -> evacuate (Consume x k r) = Consume x (evacuate . k) (evacuate r) -> evacuate (Produce _ _ r) = r -> evacuate (Enclose f) = Enclose (fmap evacuate f) -> evacuate (Deliver z) = Deliver z +> -- | @deplete p@ /depletes/ the stream processor @p@, by replacing +> -- the first producer step in @p@ with that step's depleted contination, +> -- effectively converting @p@ into a consumer processor that will never +> -- produce any more output. + +> deplete :: Functor f => P a' a b' b f r -> Consumer a' a f r +> deplete (Consume x k q) = Consume x (deplete . k) (deplete q) +> deplete (Produce _ _ q) = q +> deplete (Enclose f) = Enclose (fmap deplete f) +> deplete (Deliver r) = Deliver r +