packages feed

churros (empty) → 0.1.0.0

raw patch · 11 files changed

+881/−0 lines, 11 filesdep +asyncdep +basedep +containerssetup-changed

Dependencies added: async, base, containers, doctest, random, stm, time

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for churros++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,21 @@+Copyright 2020 Lyndon Maydwell++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in+all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.++
+ Setup.hs view
@@ -0,0 +1,6 @@++import Distribution.Simple+main = defaultMain++-- import Distribution.Extra.Doctest (defaultMainWithDoctests)+-- main = defaultMainWithDoctests "generic-data-doctest"
+ churros.cabal view
@@ -0,0 +1,57 @@+cabal-version:       2.2+name:                churros+version:             0.1.0.0+license-file:        LICENSE+author:              Lyndon Maydwell+maintainer:          lyndon@sordina.net+build-type:          Simple+extra-source-files:  CHANGELOG.md+license:             MIT+synopsis:            Churros: Channel/Arrow based streaming computation library.+category:            Data, Control+description:         The Churro library takes an opinionated approach to streaming+                     by focusing on IO processes and allowing different transport+                     options.++source-repository head+  type:     git+  location: git@github.com:sordina/churros.git++-- custom-setup+--    setup-depends:+--      base, Cabal, cabal-doctest++common churros-dependencies+  default-language: Haskell2010+  build-depends:+    base >= 4.10 && < 4.15,+    time,+    async,+    stm,+    containers,+    random++library churros-lib+  import: churros-dependencies+  hs-source-dirs:   src+  exposed-modules:+    Control.Churro,+    Control.Churro.Types,+    Control.Churro.Prelude,+    Control.Churro.Transport,+    Control.Churro.Transport.Chan++test-suite churros-test+  import:            churros-dependencies+  type:              exitcode-stdio-1.0+  hs-source-dirs:    src,test+  main-is:           Churro/Test/Main.hs+  build-depends:     doctest+  ghc-options:       -threaded -rtsopts+  other-modules:+    Control.Churro,+    Control.Churro.Types,+    Control.Churro.Prelude,+    Control.Churro.Transport,+    Control.Churro.Transport.Chan,+    Churro.Test.Examples
+ src/Control/Churro.hs view
@@ -0,0 +1,57 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE Arrows #-}++-- | Recommended entrypoint for the Churro library.+--+--   Contains enough functionality to use the package as intended.+--+--   See README.md for more information.+-- +module Control.Churro++    ( module Control.Churro++    -- ** Churro Packages++    , module Control.Churro.Types+    , module Control.Churro.Prelude+    , module Control.Churro.Transport++    -- ** Other Packages++    -- *** Re-exported from Control.Category+    , Control.Category.id++    -- *** Re-exported from Control.Arrow+    , Control.Arrow.arr+    , Control.Arrow.first+    , Control.Arrow.second+    , (Control.Arrow.>>>)+    , (Control.Arrow.<<<)+    , (Control.Arrow.&&&)+    , (Control.Arrow.***)++    -- *** Re-exported from Data.Void+    , Void()++    -- *** Re-exported from GHC.Natural+    , Natural()+    )++where++import Control.Churro.Types+import Control.Churro.Prelude+import Control.Churro.Transport++import Data.Void+import GHC.Natural+import Control.Arrow+import Control.Category++million :: Fractional p => p+million = 1e6
+ src/Control/Churro/Prelude.hs view
@@ -0,0 +1,323 @@+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE BlockArguments #-}++-- | Common transport-agnostic functions for using Churro.+-- +module Control.Churro.Prelude where++import Control.Churro.Types++import Prelude hiding (id, (.))++import           Control.Arrow            (arr)+import           Control.Category         (id, (.), (>>>))+import           Control.Concurrent       (threadDelay)+import           Control.Concurrent.Async (Async, wait)+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)+import           GHC.Natural              (Natural)+++-- $setup+-- +-- The examples in this module require the following imports:+-- +-- >>> import Control.Churro.Transport+-- >>> import Data.Time.Clock+-- ++-- * Runners++-- | Automatically wait for a churro to complete.+-- +runWait :: Transport t => Churro t Void Void -> IO ()+runWait x = wait =<< run x++-- | Read the output of a Churro into a list.+-- +-- Warning: This will block until the Churro terminates,+--          Accumulating items in memory.+--          Only use when you expect a finite amount of output.+--          Otherwise consider composing with a Sink and using `runWait`.+-- +-- >>> 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++-- | 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 = 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' c = do+    -- Compose an empty sourceList to ensure termination+    (_i,_o,a) <- runChurro (sourceList [] >>> c)+    return a++-- * Library++-- ** Sources++-- | A single items source.+--+-- >>> runWaitChan $ sourceSingleton 13 >>> sinkPrint+-- 13+--+-- Equivalent to `pure` from `Applicative`. Redefined here in case you're looking for a source!+-- +-- >>> runWaitChan $ pure 23 >>> sinkPrint+-- 23+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.+--+-- >>> runWaitChan $ sourceList [4,2] >>> sinkPrint+-- 4+-- 2+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.+--+-- >>> runWaitChan $ sourceIO (\cb -> cb 4 >> cb 2) >>> sinkPrint+-- 4+-- 2+sourceIO :: Transport t => ((o -> IO ()) -> IO a2) -> Churro t Void o+sourceIO cb =+    buildChurro \_i o -> do+        cb (yeet o . Just)+        yeet o Nothing++-- ** Sinks++-- | 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 cb = buildChurro \i _o -> yankAll i cb++-- | 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 = sinkIO print++-- ** Churros++-- | Process each item with an IO action.+--   Acts as a one-to-one process.+-- +-- >>> runWaitChan $ pure "hi" >>> process (\x -> print x >> return (reverse x)) >>> sinkPrint+-- "hi"+-- "ih"+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 = 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 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.+--   Acts as a one-to-many process.+-- +-- >>> runWaitChan $ pure 1 >>> processN (\x -> print (show x) >> return [x, succ x]) >>> sinkPrint+-- "1"+-- 1+-- 2+processN :: Transport t => (a -> IO [b]) -> Churro t a b+processN f =+    buildChurro \i o -> do+        yankAll i \x -> do mapM_ (yeet o . Just) =<< f x+        yeet o Nothing++-- | Extract xs from (Just x)s. Similar to `catMaybes`.+-- +-- >>> runWaitChan $ sourceList [Just 1, Nothing, Just 3] >>> justs >>> sinkPrint+-- 1+-- 3+justs :: Transport t => Churro t (Maybe a) a+justs = mapN (maybe [] pure)++-- | Extract ls from (Left l)s.+-- +-- >>> runWaitChan $ sourceList [Left 1, Right 2, Left 3] >>> lefts >>> sinkPrint+-- 1+-- 3+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 = mapN (either (const []) pure)++-- | Take and yield the first n items.+-- +-- WARNING: This is intended to terminate upstream once the items have been consumed+--          downstream, but there is a bug preventing this from working at present!+-- +-- >>> runWaitChan $ sourceList [1..100] >>> takeC 2 >>> sinkPrint+-- 1+-- 2+-- +-- 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 n = buildChurro \i o -> go n i o+    where+    go t i o+        | t <= 0 = yeet o Nothing+        | otherwise = do+            x <- yank i+            yeet o x+            when (isJust x) do go (pred t) i o++-- | Drop the first n items.+-- +-- >>> runWaitChan $ sourceList [1..4] >>> dropC 2 >>> sinkPrint+-- 3+-- 4+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++-- | Filter items according to a predicate.+--+-- >>> runWaitChan $ sourceList [1..5] >>> filterC (> 3) >>> sinkPrint+-- 4+-- 5+filterC :: Transport t => (a -> Bool) -> Churro t a a+filterC p = mapN (filter p . pure)++-- | Run a pure function over items, producing multiple outputs.+-- +-- >>> runWaitChan $ pure 9 >>> mapN (\x -> [x,x*10]) >>> sinkPrint+-- 9+-- 90+mapN :: Transport t => (a -> [b]) -> Churro t a b+mapN f = processN (return . f)++-- | Delay items from being sent downstream.+-- +--   Note: NominalDiffTime's Num instance interprets literals as seconds.+-- +-- >>> let sinkTimeCheck = process (const getCurrentTime) >>> withPrevious >>> arr (\(x,y) -> diffUTCTime y x > 0.01) >>> sinkPrint+-- +-- >>> runWaitChan $ sourceList [1..2] >>> sinkTimeCheck+-- False+-- +-- >>> runWaitChan $ sourceList [1..2] >>> delay 0.1 >>> sinkTimeCheck+-- True+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 d = process \x -> do+    threadDelay d+    return x++-- | Passes consecutive pairs of items downstream.+-- +-- >>> runWaitChan $ sourceList [1,2,3] >>> withPrevious >>> sinkPrint+-- (1,2)+-- (2,3)+withPrevious :: Transport t => Churro t a (a,a)+withPrevious = buildChurro \i o -> do+    prog Nothing i o +    yeet o Nothing+    where+    prog x i o = do+        y <- yank i+        case (x,y) of+            (Just x', Just y') -> yeet o (Just (x',y')) >> prog y i o+            (Nothing, Just y') -> prog (Just y') i o+            _                  -> return ()++-- | Requeue an item if it fails. Swallows exceptions and gives up after retries.+-- +--  Note: Process will always try once so if retries = 1 then a failing process will execute twice.+-- +--  The item is requeues on the input side of the churro, so if other items have+--  been passed in they will appear first!+-- +--  Catches all `SomeException`s. If you wish to narrow the execption type, consider+--  using the processRetry' variant composed with `rights`.+-- +--  Note: There is an edgecase with Chan transport where a queued retry may not execute+--        if a source completes and finalises before the item is requeued.+--        A different transport type may allow a modified retry function that requeues differently.+-- +-- >>> :{+-- let+--   prog = processRetry 1 flakeyThing+--   flakeyThing x = do+--     if x > 1+--       then print "GT"  >> return x+--       else print "LTE" >> error ("oops! " <> show x)+-- in+--   runWaitChan $ sourceList [1,2] >>> delay 0.1 >>> prog >>> sinkPrint+-- :}+-- "LTE"+-- "LTE"+-- "GT"+-- 2+-- +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' retries f = arr (0,) >>> processRetry'' retries f++-- | Rawest version of `processRetry`.+--   Expects the incoming items to contain number of retries.+-- +--   Also polymorphic over exception type. And forwards errors.+--   +processRetry'' :: (Exception e, Transport t) => Natural -> (a -> IO o) -> Churro t (Natural, a) (Either e o)+processRetry'' retries f =+    buildChurro \i o -> do+        yankAll i \(n, y) -> do+            r <- try do f y+            yeet o (Just r)+            case r of+                Right _ -> return ()+                Left  _ -> when (n < retries) do yeet i (Just (succ n, y))+        yeet o Nothing
+ src/Control/Churro/Transport.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE BlockArguments #-}++-- | Re-exporting Transport instances.+-- +-- Also includes convenience functions for working directly with transports.+-- +module Control.Churro.Transport++    ( module Control.Churro.Transport.Chan+    , module Control.Churro.Transport+    )++where++import Control.Churro.Transport.Chan+import Control.Churro.Types++-- | Write a list to a raw Transport.+-- +-- >>> import Control.Concurrent.Chan+-- >>> :{+-- do+--   c <- flex :: IO (Chan (Maybe (Maybe Int)))+--   l2c c (map Just [1,2] ++ [Nothing])+--   yankAll' c print+-- :}+-- Just (Just 1)+-- Just (Just 2)+-- Just Nothing+-- Nothing+-- +l2c :: Transport t => t (Maybe a) -> [a] -> IO ()+l2c c l = mapM_ (yeet c . Just) l >> yeet c Nothing
+ src/Control/Churro/Transport/Chan.hs view
@@ -0,0 +1,29 @@+{-# OPTIONS_GHC -Wno-orphans #-}+{-# LANGUAGE BlockArguments #-}++-- | Chan Transport Instance.++module Control.Churro.Transport.Chan where+    +import Control.Churro.Types+import Control.Churro.Prelude++import Control.Concurrent+import Data.Void++instance Transport Chan where+    flex = newChan+    yank = readChan+    yeet = writeChan++type ChurroChan = Churro Chan++-- | Convenience function for running a Churro with a Chan Transport.+-- +runWaitChan :: ChurroChan Void Void -> IO ()+runWaitChan = runWait++-- | Convenience function for running a Churro into a List with a Chan Transport.+-- +runWaitListChan :: ChurroChan Void o -> IO [o]+runWaitListChan = runWaitList
+ src/Control/Churro/Types.hs view
@@ -0,0 +1,236 @@+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE BlockArguments #-}++-- | Datatypes and definitions used by Churro library.+-- +-- Expand instances for additional documentation!++module Control.Churro.Types where++import Prelude hiding (id, (.))+import Control.Arrow+import Control.Category+import Control.Concurrent.Async (cancel, wait, Async, async)+import Data.Void+import Control.Exception (finally)++-- $setup+-- +-- We import the library for testing, although this would be a circular import in the module itself.+-- +-- >>> import Control.Churro++-- ** Data, Classes and Instances++-- | The core datatype for the library.+--   Parameters`t`, `i` and `o` represent the transport,+--   input and output types respectively.+-- +-- When building a program by composing Churros, the output Transport of one Churro is fed into the input Transports of other Churros.+-- +data Churro t i o = Churro { runChurro :: IO (t (Maybe i), t (Maybe o), Async ()) }++-- | The transport method is abstracted via the Transport class+-- +-- This allows use of pure or impure channels, such as:+-- +-- * Chan (Included in `Control.Churro.Transport.Chan`)+-- * TChan+-- * Seq+-- * Various buffered options+-- +-- Transports used in conjunction with Churros wrap items in Maybe so that once a source has been depleted it can signal completion with+-- a Nothing item.+-- +class Transport t where+    flex :: IO (t a)           -- ^ Create a new Transport+    yank :: t a -> IO a        -- ^ Yank an item of the Transport+    yeet :: t a -> a -> IO ()  -- ^ Yeet an item onto the Transport++-- | Covariant functor instance for Churro - Maps over the output.+-- +-- >>> let s = sourceList [1,2]+-- >>> runWaitChan $ s >>> sinkPrint+-- 1+-- 2+-- +-- >>> runWaitChan $ fmap succ s >>> sinkPrint+-- 2+-- 3+instance Transport t => Functor (Churro t i) where+    fmap f c = Churro do+        (i,o,a) <- runChurro c+        o'  <- flex+        a'  <- async do+            finally' (cancel a) do+                c2c f o o'+                wait a+        return (i,o',a')++-- | The Category instance allows for the creation of Churro pipelines.+-- +-- All other examples of the form `a >>> b` use this instance.+-- +-- The `id` method creates a passthrough arrow.+-- There isn't usually a reason to use `id` directly as it has no effect:+-- +-- >>> runWaitChan $ pure 1 >>> id >>> id >>> id >>> sinkPrint+-- 1+instance Transport t => Category (Churro t) where+    id = Churro do+        a <- async (return ())+        c <- flex+        return (c,c,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)++-- | 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++    f <*> g = buildChurro \_i o -> do+        (_fi, fo, fa) <- runChurro f+        (_gi, go, ga) <- runChurro g++        let+            prog :: IO ()+            prog = do+                fx <- yank fo+                gx <- yank go+                case (fx, gx) of+                    (Just f', Just g') -> (yeet o $ Just (f' g')) >> prog+                    _                  -> return ()++        prog+        yeet o Nothing+        wait fa+        wait ga++-- | 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`.+-- +-- >>> :set -XArrows+-- >>> :{+-- let sect  = process $ \x@(_x,_y,z) -> print x >> return z+--     graph =+--       proc i -> do+--         j <- arr succ  -< i+--         k <- arr show  -< j+--         l <- arr succ  -< j+--         m <- arr (> 5) -< j+--         n <- sect      -< (k,l,m)+--         o <- arr not   -< n+--         sinkPrint      -< o+-- in+-- runWaitChan $ sourceList [1,5,30] >>> graph+-- :}+-- ("2",3,False)+-- ("6",7,True)+-- ("31",32,True)+-- True+-- False+-- False+-- +-- The other Arrow methods are also usable:+-- +-- >>> runWaitChan $ pure 1 >>> (arr show &&& arr succ) >>> sinkPrint+-- ("1",2)+instance Transport t => Arrow (Churro t) where+    arr = flip fmap id++    first c = Churro do+        (i,o,a) <- runChurro c+        i'      <- flex+        o'      <- flex++        let go = do+                is <- yank i'+                yeet i (fmap fst is)++                os <- yank o+                yeet o' $ (,) <$> os <*> fmap snd is++                case (is, os) of+                    (Just _, Just _) -> go+                    _ -> return ()++        a' <- async do+            go+            yeet o' Nothing+            wait a++        return (i',o',a')++-- ** 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 => (t (Maybe i) -> t (Maybe o) -> IO ()) -> Churro t i o+buildChurro cb = Churro do+    i <- flex+    o <- flex+    a <- async do cb i o+    return (i,o,a)++-- | Yeet all items from a list into a transport.+-- +yeetList :: (Foldable t1, Transport t2) => t2 a -> t1 a -> IO ()+yeetList t = mapM_ (yeet t)++-- | Yank all items from a Raw transport into a list.+-- +--   Won't terminate until the transport has been consumed.+-- +yankList :: Transport t => t (Maybe a) -> IO [a]+yankList t = do+    x <- yank t+    case x of +        Nothing -> return []+        Just y  -> (y :) <$> yankList t++-- | Yank each item from a transport into a callback.+-- +yankAll :: Transport t => t (Maybe i) -> (i -> IO a) -> IO ()+yankAll c f = do+    x <- yank c+    case x of+        Nothing -> return ()+        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 => t (Maybe a) -> (Maybe a -> IO b) -> IO b+yankAll' c f = do+    yankAll c (f . Just)+    f Nothing++-- | Yank then Yeet each item from one Transport into another.+-- +-- Raw items are used so `Nothing` should be Yeeted once the transport is depleted.+-- +c2c :: Transport t => (a1 -> a2) -> t (Maybe a1) -> t (Maybe a2) -> IO ()+c2c f i o = yankAll' i (yeet o . fmap f)++-- | Flipped `finally`.+finally' :: IO b -> IO a -> IO a+finally' = flip finally
+ test/Churro/Test/Examples.hs view
@@ -0,0 +1,100 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE NoMonomorphismRestriction #-}+{-# LANGUAGE Arrows #-}++-- | A set of examples of more complicated and problematic Churros.+-- +module Churro.Test.Examples where++import Data.Map (fromList)+import Data.List++import Prelude hiding (id, (.))++import Control.Churro++-- $setup+-- +-- >>> import System.Timeout     (timeout)+-- >>> import Control.Monad      (forever)+-- >>> import Control.Concurrent (threadDelay)++-- ** Tests++-- | Checks that the IO nature of the churros doesn't duplicate operations.+--   Actions within a pipeline should only occur once no matter how the+--   pipeline is composed.+--+-- >>> runWaitChan linear+-- Debugging [l1]: 1+-- Debugging [l2]: 1+-- Debugging [r1]: 1+-- Debugging [r2]: 1+-- 1 +--+linear :: Transport t => Churro t Void Void+linear = sourceList [1::Int]+    >>> ((processDebug "l1" >>> processDebug "l2") >>> processDebug "r1" >>> processDebug "r2")+    >>> sinkPrint++-- | A more complicated pipeline exampe involving maps.+-- +-- >>> 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 = sourceList (take 3 maps)+        >>> withPrevious+        >>> takeC (10 :: Int)+        >>> sinkPrint+    where+    maps    = map fromList $ zipWith zip updates updates+    updates = map (take 2) (tails [0 :: Int ..])++-- | Consumers terminaiting should kill sources from producing.+-- +-- This can fail in the following scenarios if cancellation isn't implemented correctly:+-- +-- >>> timeout 1500000 $ runWaitChan $ sourceList [1..5] >>> delay 1 >>> takeC 1 >>> sinkPrint+-- 1+-- Just ()+-- +-- Cancells infinite producer with inbuilt delays:+-- +-- >>> timeout 1500000 $ runWaitChan $ sourceIO (\cb -> forever (cb 1 >> threadDelay 100000)) >>> takeC 1 >>> sinkPrint+-- 1+-- Just ()+-- +-- Cancells upstream infinite producer with no inbuilt delay:+-- +-- >>> timeout 2500000 $ runWaitChan $ sourceList [1..] >>> delay 1 >>> takeC 1 >>> sinkPrint+-- 1+-- Just ()+-- +-- Note that the timeout here has to be sufficient for a thread switch to occor and the+-- action be cancelled! See what happens if the timeout is only 1.5s:+-- +-- >>> timeout 1500000 $ runWaitChan $ sourceList [1..] >>> delay 1 >>> takeC 1 >>> sinkPrint+-- 1+-- Nothing+-- +-- What should happen is that the Category instance composition of:+-- +--  ...  delay 1 >>> takeC 1  ...+--  ^^^ PRE ^^^^     ^^^ POST ^^^+-- +-- When POST terminates it should cancel the computation in PRE.+-- +-- Failures may be caused by one of the following:+-- +-- * Nested Async actions don't cascade on cancellation (incorrect finally clauses)+-- * The associativity laws of Category are broken, meaning that cancellation doesn't behave as it should.+-- * Producer blocks cancellation from being requested+-- * Chan is blocked preventing indicating termination to consumers+-- * Infinite source is causing issues (ruled out with this test example)+-- +-- This is currently working, but tests here should check that the implementation+-- isn't broken.
+ test/Churro/Test/Main.hs view
@@ -0,0 +1,14 @@++module Main where++import Test.DocTest++-- import qualified Churro.Test.Examples as Ex+-- import qualified Build_doctests       as DT++-- TODO: Find out how to use a different test manager to allow these to be run independently+main :: IO ()+main = do+    doctest ["-isrc", "test/Churro/Test/Examples.hs"]+    -- Ex.main+