churros 0.1.2.0 → 0.1.3.0

raw patch · 10 files changed

+158/−117 lines, 10 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• README.md +8/−8
• churros.cabal +1/−1
• src/Control/Churro/Prelude.hs +78/−50
• src/Control/Churro/Transport/Chan.hs +3/−3
• src/Control/Churro/Transport/Unagi.hs +3/−3
• src/Control/Churro/Transport/Unagi/Bounded.hs +3/−3
• src/Control/Churro/Types.hs +52/−40
• test/Churro/Test/Examples.hs +4/−4
• test/Churro/Test/Main.hs +2/−5

@@ -1,5 +1,9 @@ # Revision history for Churros +## 0.1.3.0 -- 2020-10-17++* Generalised functions to have Monoid for Async action result.+ ## 0.1.2.0 -- 2020-10-15  * Added type families to transport class to allow for different in/out types

@@ -13,13 +13,13 @@ Signup Example:  ```-New-Users {-> ID} ---> Validate {ID -> (ID,Email} ?--> Deploy {Email -> Port} --?-> Log {Port|Receipt ->}------------------      -------------------------- |    ----------------------   |   ----------------------    Subscribe                 REST Request        |          Run Server         |        Write File-                                                  |                             |-                                                  +--> Warn {Email -> Receipt} -+-                                                       ------------------------                                                             Send Email+New-Users -->  Validate -------?----->  Deploy  -----?--------------> Log+Subscribe    REST Request      |      Run Server     |         Write File+{-> ID}    {ID -> (ID,Email}   |    {Email -> Port}  |  {Port|Receipt ->}+                               |                     |+                               +-------> Warn -------++                                      Send Email+                                  {Email -> Receipt} ```  Developed from a history of attempting to use co-routines libraries for setting up complicated asynchronous processes@@ -37,7 +37,7 @@ Disadvantages:  * No pure interface!-* No expressive return type for the async action representing the background process is encoded in the Churro datatype (could this be addressed?)+* Type for the async action restricted to Monoid for most operations. * Limited ability to perform lock-step computation (although this is by design)  See [Hackage](https://hackage.haskell.org/package/churros-0.1.0.0/candidate) for more info!

@@ -1,6 +1,6 @@ cabal-version:       2.2 name:                churros-version:             0.1.2.0+version:             0.1.3.0 license-file:        LICENSE author:              Lyndon Maydwell maintainer:          lyndon@sordina.net

@@ -18,7 +18,6 @@ import           Control.Exception        (Exception, SomeException, try) import           Control.Monad            (replicateM_, when) import           Data.Foldable            (for_)-import           Data.IORef               (newIORef, readIORef, writeIORef) import           Data.Maybe               (isJust) import           Data.Time                (NominalDiffTime) import           Data.Void                (Void)@@ -29,6 +28,7 @@ --  -- The examples in this module require the following imports: -- +-- >>> :set -XBlockArguments -- >>> import Control.Churro.Transport -- >>> import Data.Time.Clock -- @@ -37,7 +37,7 @@  -- | Automatically wait for a churro to complete. -- -runWait :: Transport t => Churro t Void Void -> IO ()+runWait :: Transport t => Churro a t Void Void -> IO a runWait x = wait =<< run x  -- | Read the output of a Churro into a list.@@ -50,32 +50,21 @@ -- >>> runWaitListChan $ sourceList [0..4] >>> arr succ -- [1,2,3,4,5] -- -runWaitList :: Transport t => Churro t Void o -> IO [o]-runWaitList x = do-    t <- newIORef []--    let-        c = buildChurro \i _o -> do-            l <- yankList i-            writeIORef t l--    runWait $ x >>> c--    readIORef t+runWaitList :: (Transport t, Monoid a) => Churro a t Void b -> IO [b]+runWaitList c = runWait $ (c >>> arr' (:[])) >>>> sink   -- | Run a sourced and sinked (double-dipped) churro and return an async action representing the in-flight processes.--- -run :: Transport t => Churro t Void Void -> IO (Async ())+--+run :: Transport t => Churro a t Void Void -> IO (Async a) run = run'  -- | Run any churro, there is no check that this was spawned with a source, or terminated with a sink. --   This is unsafe, since the pipeline may not generate or consume in a predictable way. --   Use `run` instead unless you are confident you know what you're doing. -- -run' :: Transport t => Churro t i o -> IO (Async ())+run' :: Transport t => Churro a t i o -> IO (Async a) run' c = do-    -- Compose an empty sourceList to ensure termination-    (_i,_o,a) <- runChurro (sourceList [] >>> c)+    (_i,_o,a) <- runChurro c     return a  -- * Library@@ -91,7 +80,7 @@ --  -- >>> runWaitChan $ pure 23 >>> sinkPrint -- 23-sourceSingleton :: Transport t => o -> Churro t Void o+sourceSingleton :: Transport t => o -> Churro () t Void o sourceSingleton x = sourceList [x]  -- | Create a source from a list of items, sending each down the churro independently.@@ -99,7 +88,7 @@ -- >>> runWaitChan $ sourceList [4,2] >>> sinkPrint -- 4 -- 2-sourceList :: (Transport t, Foldable f) => f o -> Churro t Void o+sourceList :: (Transport t, Foldable f) => f o -> Churro () t Void o sourceList = sourceIO . for_  -- | Create a source from an IO action that is passed a function to yield new items.@@ -107,50 +96,89 @@ -- >>> runWaitChan $ sourceIO (\cb -> cb 4 >> cb 2) >>> sinkPrint -- 4 -- 2-sourceIO :: Transport t => ((o -> IO ()) -> IO ()) -> Churro t Void o+sourceIO :: Transport t => ((o -> IO ()) -> IO a) -> Churro a t Void o sourceIO cb =     buildChurro \_i o -> do-        cb (yeet o . Just)+        r <- cb (yeet o . Just)         yeet o Nothing+        return r  -- | Combine a list of sources into a single source. --  -- Sends individual items downstream without attempting to combine them. -- --- >>> runWaitChan $ sources [pure 1, pure 1] >>> sinkPrint+-- Warning: Passing an empty list is unspecified.+-- +-- TODO: Use NonEmptyList instead of []+-- +-- >>> runWaitChan $ sources_ [pure 1, pure 1] >>> sinkPrint -- 1 -- 1-sources :: Transport t => [Source t o] -> Source t o+sources :: (Transport t, Monoid a) => [Source a t o] -> Source () t o sources ss = sourceIO \cb -> do-    asyncs <- mapM (\s -> run $ s >>> sinkIO cb) ss-    (a, _) <- waitAny asyncs-    wait a+    asyncs <- mapM (\s -> run $ s >>>> sinkIO cb) ss+    (_, r) <- waitAny asyncs+    return r +-- | This is `sources` specialised to `()`.+sources_ :: Transport t => [Source () t o] -> Source () t o+sources_ = sources+ -- ** Sinks  -- | Consume all items with no additional effects. --  -- TODO: Decide if we should use some kind of `nf` evaluation here to force items. -- --- >>> runWaitChan $ pure 1 >>> process print >>> sink+-- >>> runWaitChan $ pure 1 >>> process print >>> sink_ -- 1 -- -sink :: Transport t => Churro t b Void-sink = sinkIO (const (return ()))+sink_ :: Transport t => Churro () t i Void+sink_ = sinkIO (const (return ())) +-- | Consume all items and combines them into a result via their monoid.+-- +-- >>> :set -XFlexibleContexts+-- >>> r <- runWaitChan $ pure' [1 :: Int] >>> sink+-- >>> print r+-- [1]+sink :: (Transport t, Monoid a) => Churro a t a Void+sink = sinkIO return+ -- | Consume a churro with an IO process. --  -- >>> runWaitChan $ pure 1 >>> sinkIO (\x -> print "hello" >> print (succ x)) -- "hello" -- 2-sinkIO :: Transport t => (o -> IO ()) -> Churro t o Void+sinkIO :: (Transport t, Monoid a) => (o -> IO a) -> Churro a t o Void sinkIO cb = buildChurro \i _o -> yankAll i cb +-- | Create a "sink" with more flexibility about when items are demanded using a higher-order "HO" callback.+-- +-- This also allows a non-unit async action that can be recovered when run.+-- +-- WARNING: You should use the provided callback if you want to acually create a sink.+-- +-- TODO: Use hidden callback return type in order to ensure that the callback is called.+-- +-- >>> import System.Timeout (timeout)+-- >>> :{+-- do+--   r <- timeout 100000 $ runWaitChan $ sourceSingleton 1 >>>> sinkHO \ya -> do+--     ya (print . show)+--     return 25+--   print r+-- :}+-- "1"+-- Just 25+sinkHO :: Transport t => (((i -> IO ()) -> IO ()) -> IO a) -> Churro a t i o+sinkHO cb = buildChurro \i _o -> cb (yankAll i)+ -- | Consume and print each item. Used in many examples, but not much use outside debugging! --  -- >>> runWaitChan $ pure "hi" >>> sinkPrint -- "hi"-sinkPrint :: (Transport t, Show a) => Churro t a Void+sinkPrint :: (Transport t, Show a) => Churro () t a Void sinkPrint = sinkIO print      @@ -162,15 +190,15 @@ -- >>> runWaitChan $ pure "hi" >>> process (\x -> print x >> return (reverse x)) >>> sinkPrint -- "hi" -- "ih"-process :: Transport t => (a -> IO b) -> Churro t a b+process :: Transport t => (a -> IO b) -> Churro () t a b process f = processN (fmap pure . f)  -- | Print each item then pass it on.-processPrint :: (Transport t, Show b) => Churro t b b+processPrint :: (Transport t, Show b) => Churro () t b b processPrint = process \x -> do print x >> return x  -- | Print each item with an additional debugging label.-processDebug :: (Transport t, Show b) => String -> Churro t b b+processDebug :: (Transport t, Show b) => String -> Churro () t b b processDebug d = process \x -> putStrLn ("Debugging [" <> d <> "]: " <> show x) >> return x  -- | Process each item with an IO action and potentially yield many items as a result.@@ -180,7 +208,7 @@ -- "1" -- 1 -- 2-processN :: Transport t => (a -> IO [b]) -> Churro t a b+processN :: Transport t => (i -> IO [o]) -> Churro () t i o processN f =     buildChurro \i o -> do         yankAll i \x -> do mapM_ (yeet o . Just) =<< f x@@ -191,7 +219,7 @@ -- >>> runWaitChan $ sourceList [Just 1, Nothing, Just 3] >>> justs >>> sinkPrint -- 1 -- 3-justs :: Transport t => Churro t (Maybe a) a+justs :: Transport t => Churro () t (Maybe a) a justs = mapN (maybe [] pure)  -- | Extract ls from (Left l)s.@@ -199,14 +227,14 @@ -- >>> runWaitChan $ sourceList [Left 1, Right 2, Left 3] >>> lefts >>> sinkPrint -- 1 -- 3-lefts :: Transport t => Churro t (Either a b) a+lefts :: Transport t => Churro () t (Either a b) a lefts = mapN (either pure (const []))  -- | Extract rs from (Right r)s. --  -- >>> runWaitChan $ sourceList [Left 1, Right 2, Left 3] >>> rights >>> sinkPrint -- 2-rights :: Transport t => Churro t (Either a b) b+rights :: Transport t => Churro () t (Either a b) b rights = mapN (either (const []) pure)  -- | Take and yield the first n items.@@ -221,7 +249,7 @@ -- This implementation explicitly stops propagating when the Churro completes, -- although this could be handled by downstream consumer composition terminating -- the producer and just using replicateM.-takeC :: (Transport t, Integral n) => n -> Churro t a a+takeC :: (Transport t, Integral n) => n -> Churro () t a a takeC n = buildChurro \i o -> go n i o     where     go t i o@@ -236,7 +264,7 @@ -- >>> runWaitChan $ sourceList [1..4] >>> dropC 2 >>> sinkPrint -- 3 -- 4-dropC :: (Transport t, Integral n) => n -> Churro t a a+dropC :: (Transport t, Integral n) => n -> Churro () t a a dropC n = buildChurro \i o -> do     replicateM_ (fromIntegral n) (yank i) -- TODO: Check the async behaviour of this...     c2c id i o@@ -246,7 +274,7 @@ -- >>> runWaitChan $ sourceList [1..5] >>> filterC (> 3) >>> sinkPrint -- 4 -- 5-filterC :: Transport t => (a -> Bool) -> Churro t a a+filterC :: Transport t => (a -> Bool) -> Churro () t a a filterC p = mapN (filter p . pure)  -- | Run a pure function over items, producing multiple outputs.@@ -254,7 +282,7 @@ -- >>> runWaitChan $ pure 9 >>> mapN (\x -> [x,x*10]) >>> sinkPrint -- 9 -- 90-mapN :: Transport t => (a -> [b]) -> Churro t a b+mapN :: Transport t => (a -> [b]) -> Churro () t a b mapN f = processN (return . f)  -- | Delay items from being sent downstream.@@ -268,11 +296,11 @@ --  -- >>> runWaitChan $ sourceList [1..2] >>> delay 0.1 >>> sinkTimeCheck -- True-delay :: Transport t => NominalDiffTime -> Churro t a a+delay :: Transport t => NominalDiffTime -> Churro () t a a delay = delayMicro . ceiling @Double . fromRational . (*1000000) . toRational  -- | Delay items in microseconds. Works the same way as `delay`.-delayMicro :: Transport t => Int -> Churro t a a+delayMicro :: Transport t => Int -> Churro () t a a delayMicro d = process \x -> do     threadDelay d     return x@@ -282,7 +310,7 @@ -- >>> runWaitChan $ sourceList [1,2,3] >>> withPrevious >>> sinkPrint -- (1,2) -- (2,3)-withPrevious :: Transport t => Churro t a (a,a)+withPrevious :: Transport t => Churro () t a (a,a) withPrevious = buildChurro \i o -> do     prog Nothing i o      yeet o Nothing@@ -323,12 +351,12 @@ -- "GT" -- 2 -- -processRetry :: Transport t => Natural -> (i -> IO o) -> Churro t i o+processRetry :: Transport t => Natural -> (i -> IO o) -> Churro () t i o processRetry retries f = processRetry' @SomeException retries f >>> rights  -- | Raw version of `processRetry`. -- Polymorphic over exception type and forwards errors. --   -processRetry' :: (Exception e, Transport t) => Natural -> (i -> IO o) -> Churro t i (Either e o)+processRetry' :: (Exception e, Transport t) => Natural -> (i -> IO o) -> Churro () t i (Either e o) processRetry' retries f = arr (0,) >>> processRetry'' retries f  -- | Rawest version of `processRetry`.@@ -336,7 +364,7 @@ --  --   Also polymorphic over exception type. And forwards errors. --   -processRetry'' :: (Transport t, Exception e, Ord n, Enum n) => n -> (a -> IO b) -> Churro t (n, a) (Either e b)+processRetry'' :: (Transport t, Exception e, Ord n, Enum n) => n -> (a -> IO b) -> Churro () t (n, a) (Either e b) processRetry'' retries f =     buildChurro' \i' o i -> do         yankAll o \(n, y) -> do

@@ -21,14 +21,14 @@         c <- newChan         return (ChanIn c, ChanOut c) -type ChurroChan = Churro Chan+type ChurroChan a = Churro a Chan  -- | Convenience function for running a Churro with a Chan Transport. -- -runWaitChan :: ChurroChan Void Void -> IO ()+runWaitChan :: ChurroChan a Void Void -> IO a runWaitChan = runWait  -- | Convenience function for running a Churro into a List with a Chan Transport. -- -runWaitListChan :: ChurroChan Void o -> IO [o]+runWaitListChan :: ChurroChan () Void o -> IO [o] runWaitListChan = runWaitList

@@ -22,14 +22,14 @@         (i, o) <- newChan         return (ChanIn i, ChanOut o) -type ChurroUnagi = Churro Unagi+type ChurroUnagi a = Churro a Unagi  -- | Convenience function for running a Churro with an Unagi Transport. -- -runWaitUnagi :: ChurroUnagi Void Void -> IO ()+runWaitUnagi :: ChurroUnagi a Void Void -> IO a runWaitUnagi = runWait  -- | Convenience function for running a Churro into a List with an Unagi Transport. -- -runWaitListUnagi :: ChurroUnagi Void o -> IO [o]+runWaitListUnagi :: ChurroUnagi () Void o -> IO [o] runWaitListUnagi = runWaitList

@@ -36,16 +36,16 @@         (i, o) <- newChan (fromIntegral (natVal (Proxy :: Proxy n)))         return (ChanIn i, ChanOut o) -type ChurroUnagiBounded n = Churro (UnagiBounded n)+type ChurroUnagiBounded a n = Churro a (UnagiBounded n)  -- | Convenience function for running a Churro with a Bounded Unagi-Chan Transport. -- -runWaitUnagi :: KnownNat n => ChurroUnagiBounded n Void Void -> IO ()+runWaitUnagi :: KnownNat n => ChurroUnagiBounded a n Void Void -> IO a runWaitUnagi = runWait  -- | Convenience function for running a Churro into a List with a Bounded Unagi-Chan Transport. --  -- >>> runWaitListUnagi @10 $ sourceList [1,2,3] >>> arr succ -- [2,3,4]-runWaitListUnagi :: KnownNat n => ChurroUnagiBounded n Void o -> IO [o]+runWaitListUnagi :: KnownNat n => ChurroUnagiBounded () n Void o -> IO [o] runWaitListUnagi = runWaitList

@@ -2,6 +2,7 @@ {-# LANGUAGE InstanceSigs #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE BlockArguments #-}+{-# LANGUAGE NoMonomorphismRestriction #-}  -- | Datatypes and definitions used by Churro library. -- @@ -39,10 +40,10 @@ -- Convenience types of `Source`, `Sink`, and `DoubleDipped` are also defined, -- although use is not required. -- -data Churro t i o   = Churro { runChurro :: IO (In t (Maybe i), Out t (Maybe o), Async ()) }-type Source t   o   = Churro t Void o-type Sink   t i     = Churro t i Void-type DoubleDipped t = Churro t Void Void+data Churro a t i o   = Churro { runChurro :: IO (In t (Maybe i), Out t (Maybe o), Async a) }+type Source a t   o   = Churro a t Void o+type Sink   a t i     = Churro a t i Void+type DoubleDipped a t = Churro a t Void Void  -- | The transport method is abstracted via the Transport class -- @@ -80,7 +81,7 @@ -- >>> runWaitChan $ fmap succ s >>> sinkPrint -- 2 -- 3-instance Transport t => Functor (Churro t i) where+instance Transport t => Functor (Churro a t i) where     fmap f c = Churro do         (i,o,a) <- runChurro c         (i',o') <- flex@@ -99,30 +100,38 @@ --  -- >>> runWaitChan $ pure 1 >>> id >>> id >>> id >>> sinkPrint -- 1-instance Transport t => Category (Churro t) where+instance (Transport t, Monoid a) => Category (Churro a t) where     id = Churro do         (i,o) <- flex-        a     <- async (return ())+        a     <- async mempty         return (i,o,a) -    g . f = Churro do-        (fi, fo, fa) <- runChurro f-        (gi, go, ga) <- runChurro g-        a <- async do c2c id fo gi-        b <- async do-            finally' (cancel a >> cancel fa >> cancel ga) do-                wait ga-                cancel fa-                cancel a-        return (fi, go, b)+    g . f = f >>>> g +-- | Category style composition that allows for return type to change downstream.+-- +(>>>>) :: (Transport t, fo ~ gi) => Churro a1 t fi fo -> Churro a2 t gi go -> Churro a2 t fi go+f >>>> g = Churro do+    (fi, fo, fa) <- runChurro f+    (gi, go, ga) <- runChurro g+    a <- async do c2c id fo gi+    b <- async do+        finally' (cancel a >> cancel fa >> cancel ga) do+            r <- wait ga+            cancel fa+            cancel a+            return r+    return (fi, go, b)+ -- | The Applicative instance allows for pairwise composition of Churro pipelines. --   Once again this is covariat and the composition occurs on the output transports of the Churros. --  --  The `pure` method allows for the creation of a Churro yielding a single item. -- -instance Transport t => Applicative (Churro t Void) where-    pure x = buildChurro \_i o -> yeet o (Just x) >> yeet o Nothing+-- TODO: Generalise () to a Monoid constraint.+-- +instance Transport t => Applicative (Churro () t Void) where+    pure = pure'      f <*> g = buildChurro \_i o -> do         (_fi, fo, fa) <- runChurro f@@ -143,10 +152,14 @@         wait fa         wait ga +-- | More general variant of `pure` with Monoid constraint.+pure' :: (Transport t, Monoid a) => o -> Churro a t i o+pure' x = buildChurro \_i o -> yeet o (Just x) >> yeet o Nothing >> return mempty+ -- | The Arrow instance allows for building non-cyclic directed graphs of churros. --  --  The `arr` method allows for the creation of a that maps items with a pure function.---  This is equivalent to `fmap f id`.+--  This is equivalent to `fmap f id`. This is more general and exposed via arr`. --  -- >>> :set -XArrows -- >>> :{@@ -175,8 +188,8 @@ --  -- >>> runWaitChan $ pure 1 >>> (arr show &&& arr succ) >>> sinkPrint -- ("1",2)-instance Transport t => Arrow (Churro t) where-    arr = flip fmap id+instance Transport t => Arrow (Churro () t) where+    arr = arr'      first c = Churro do         (i,o,a)   <- runChurro c@@ -201,24 +214,26 @@          return (ai',bo',a') +-- | More general version of `arr`.+-- +-- Useful when building pipelines that need to work with return types.+arr' :: (Functor (cat a), Category cat) => (a -> b) -> cat a b+arr' f = fmap f id+ -- ** Helpers  -- | A helper to facilitate constructing a Churro that makes new input and output transports available for manipulation. --  -- The manipulations performed are carried out in the async action associated with the Churro -- -buildChurro :: Transport t => (Out t (Maybe i) -> In t (Maybe o) -> IO ()) -> Churro t i o-buildChurro cb = Churro do-    (ai,ao) <- flex-    (bi,bo) <- flex-    a       <- async do cb ao bi-    return (ai,bo,a)+buildChurro :: Transport t => (Out t (Maybe i) -> In t (Maybe o) -> IO a) -> Churro a t i o+buildChurro cb = buildChurro' \_o' i o -> cb i o  -- | A version of `buildChurro` that also passes the original input to the callback so that you can reschedule items. --  -- Used by "retry" style functions. -- -buildChurro' :: Transport t => (In t (Maybe i) -> Out t (Maybe i) -> In t (Maybe o) -> IO ()) -> Churro t i o+buildChurro' :: Transport t => (In t (Maybe i) -> Out t (Maybe i) -> In t (Maybe o) -> IO a) -> Churro a t i o buildChurro' cb = Churro do     (ai,ao) <- flex     (bi,bo) <- flex@@ -235,29 +250,26 @@ --   Won't terminate until the transport has been consumed. --  yankList :: Transport t => Out t (Maybe a) -> IO [a]-yankList t = do-    x <- yank t-    case x of -        Nothing -> return []-        Just y  -> (y :) <$> yankList t+yankList = flip yankAll (pure . pure)  -- | Yank each item from a transport into a callback. -- -yankAll :: Transport t => Out t (Maybe i) -> (i -> IO a) -> IO ()+yankAll :: (Transport t, Monoid a) => Out t (Maybe i) -> (i -> IO a) -> IO a yankAll c f = do     x <- yank c     case x of-        Nothing -> return ()-        Just y  -> f y >> yankAll c f+        Nothing -> mempty+        Just y  -> f y <> yankAll c f  -- | Yank each raw item from a transport into a callback. --  -- The items are wrapped in Maybes and when all items are yanked, Nothing is fed to the callback. -- -yankAll' :: Transport t => Out t (Maybe a) -> (Maybe a -> IO b) -> IO b+yankAll' :: (Transport t, Monoid b) => Out t (Maybe a) -> (Maybe a -> IO b) -> IO b yankAll' c f = do-    yankAll c (f . Just)-    f Nothing+    x <- yankAll c (f . Just)+    y <- f Nothing+    return (x <> y)  -- | Yank then Yeet each item from one Transport into another. -- 

@@ -35,7 +35,7 @@ -- Debugging [r2]: 1 -- 1  ---linear :: Transport t => Churro t Void Void+linear :: Transport t => Churro () t Void Void linear = sourceList [1::Int]     >>> ((processDebug "l1" >>> processDebug "l2") >>> processDebug "r1" >>> processDebug "r2")     >>> sinkPrint@@ -45,7 +45,7 @@ -- >>> runWaitChan pipeline -- (fromList [(0,0),(1,1)],fromList [(1,1),(2,2)]) -- (fromList [(1,1),(2,2)],fromList [(2,2),(3,3)])-pipeline :: ChurroChan Void Void+pipeline :: ChurroChan () Void Void pipeline = sourceList (take 3 maps)         >>> withPrevious         >>> takeC (10 :: Int)@@ -58,7 +58,7 @@ --  -- This can fail in the following scenarios if cancellation isn't implemented correctly: -- --- >>> timeout 1500000 $ runWaitChan $ sourceList [1..5] >>> delay 1 >>> takeC 1 >>> sinkPrint+-- >>> timeout 150000 $ runWaitChan $ sourceList [1..5] >>> delay 0.1 >>> takeC 1 >>> sinkPrint -- 1 -- Just () -- @@ -70,7 +70,7 @@ --  -- Cancells upstream infinite producer with no inbuilt delay: -- --- >>> timeout 2500000 $ runWaitChan $ sourceList [1..] >>> delay 1 >>> takeC 1 >>> sinkPrint+-- >>> timeout 2500000 $ runWaitChan $ sourceList [1..] >>> delay 0.1 >>> takeC 1 >>> sinkPrint -- 1 -- Just () -- 

@@ -1,12 +1,9 @@  module Main where -import Test.DocTest---- import qualified Churro.Test.Examples as Ex--- import qualified Build_doctests       as DT+import Test.DocTest ( doctest ) --- TODO: Find out how to use a different test manager to allow these to be run independently+-- TODO: Pass additional paths in automatically or via arguments. main :: IO () main = do     doctest