churros 0.1.5.0 → 0.1.6.0

raw patch · 7 files changed

+120/−11 lines, 7 filesdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• churros.cabal +6/−3
• src/Control/Churro/Transport/MVar.hs +34/−0
• src/Control/Churro/Transport/MVar/Latest.hs +43/−0
• src/Control/Churro/Transport/Unagi/Bounded.hs +4/−2
• src/Control/Churro/Types.hs +6/−5
• test/Churro/Test/Examples.hs +22/−1

@@ -1,5 +1,10 @@ # Revision history for Churros +## 0.1.6.0 -- 2022-10-16++* Bump base dependency bounds to allow use of GHC 9.x+* Fix warnings+ ## 0.1.5.1 -- 2020-10-20  * Generalising processes to allow for different transport for process group.

@@ -1,6 +1,6 @@ cabal-version:       2.2 name:                churros-version:             0.1.5.0+version:             0.1.6.0 license-file:        LICENSE author:              Lyndon Maydwell maintainer:          lyndon@sordina.net@@ -24,7 +24,7 @@ common churros-dependencies   default-language: Haskell2010   build-depends:-    base >= 4.10 && < 4.15,+    base >= 4.10 && < 4.17,     time,     async,     stm,@@ -42,7 +42,8 @@     Control.Churro.Transport,     Control.Churro.Transport.Chan     Control.Churro.Transport.Unagi-    Control.Churro.Transport.Unagi.Bounded+    Control.Churro.Transport.MVar+    Control.Churro.Transport.MVar.Latest  test-suite churros-test   import:            churros-dependencies@@ -59,4 +60,6 @@     Control.Churro.Transport.Chan     Control.Churro.Transport.Unagi     Control.Churro.Transport.Unagi.Bounded+    Control.Churro.Transport.MVar+    Control.Churro.Transport.MVar.Latest     Churro.Test.Examples

@@ -0,0 +1,34 @@+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wno-orphans #-}+{-# LANGUAGE BlockArguments #-}++-- | MVar Transport Instance. Blocks when item hasn't been consumed downstream.++module Control.Churro.Transport.MVar where+    +import Control.Churro.Types+import Control.Churro.Prelude++import Control.Concurrent+import Data.Void++instance Transport MVar where+    data In  MVar a = ChanIn  (MVar a)+    data Out MVar a = ChanOut (MVar a)+    yank (ChanOut c) = takeMVar c+    yeet (ChanIn  c) = putMVar  c+    flex = do +        c <- newEmptyMVar+        return (ChanIn c, ChanOut c)++type ChurroMVar a = Churro a MVar++-- | Convenience function for running a Churro with an MVar Transport.+-- +runWaitMVar :: ChurroMVar a Void Void -> IO a+runWaitMVar = runWait++-- | Convenience function for running a Churro into a List with an MVar Transport.+-- +runWaitListMVar :: ChurroMVar () Void o -> IO [o]+runWaitListMVar = runWaitList

@@ -0,0 +1,43 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE TypeFamilies #-}+{-# OPTIONS_GHC -Wno-orphans #-}+{-# LANGUAGE BlockArguments #-}++-- | MVar Transport Instance. New items overwrite the current queued item.+-- +-- Useful in subscription-like contexts where you don't care about outdated values.+-- +-- This is surprisingly useful since it doesn't block sources, but also doesn't accumulate items.+-- +-- WARNING: Don't use if you want to ensure that all produced items are consumed!+-- +module Control.Churro.Transport.MVar.Latest where+    +import Control.Churro.Types+import Control.Churro.Prelude++import Control.Concurrent+import Data.Void++data Latest a where++instance Transport Latest where+    data In  Latest a    = ChanIn  (MVar a)+    data Out Latest a    = ChanOut (MVar a)+    yank (ChanOut c)   = takeMVar c+    yeet (ChanIn  c) v = tryTakeMVar c >> putMVar  c v+    flex = do +        c <- newEmptyMVar+        return (ChanIn c, ChanOut c)++type ChurroLatest a = Churro a Latest++-- | Convenience function for running a Churro with a MVar backed Latest Transport.+-- +runWaitLatest :: ChurroLatest a Void Void -> IO a+runWaitLatest = runWait++-- | Convenience function for running a Churro into a List with a MVar backed Latest Transport.+-- +runWaitListLatest :: ChurroLatest () Void o -> IO [o]+runWaitListLatest = runWaitList

@@ -50,13 +50,15 @@ runWaitListUnagi :: KnownNat n => ChurroUnagiBounded () n Void o -> IO [o] runWaitListUnagi = runWaitList --- A version of processes that mandates the use of a single item buffer for the internal processes.+-- | A version of processes that mandates the use of a single item buffer for the internal processes.+--  -- This is useful in order to prevent allocating to processes that are not yet idle. --  processesUnagi :: (Traversable f, Transport t, Monoid a) => f (ChurroUnagiBounded a 1 i o) -> Churro a t i o processesUnagi = processes --- A version of thief that mandates the use of a single item buffer for the internal processes.+-- | A version of thief that mandates the use of a single item buffer for the internal processes.+--  -- This is useful in order to prevent allocating to processes that are not yet idle. --  thiefUnagi :: (Transport t, Monoid a) => Int -> ChurroUnagiBounded a 1 i o -> Churro a t i o

@@ -15,6 +15,7 @@ import Control.Category import Control.Concurrent.Async (cancel, wait, Async, async) import Data.Void+import Data.Kind (Type) import Control.Exception (finally)  -- $setup@@ -64,9 +65,9 @@ -- Channels like Chan that have a single channel act as in/out simply reuse the -- same channel in the pair returned. -- -class Transport (t :: * -> *) where-    data In  t :: * -> *-    data Out t :: * -> *+class Transport (t :: Type -> Type) where+    data In  t :: Type -> Type+    data Out t :: Type -> Type     flex :: IO (In t a, Out t a)  -- ^ Create a new pair of Transports.     yank :: Out t a -> IO a       -- ^ Yank an item off the Transport     yeet :: In t a -> a -> IO ()  -- ^ Yeet an item onto the Transport@@ -143,13 +144,13 @@                 fx <- yank fo                 gx <- yank go                 case (fx, gx) of-                    (Just f', Just g') -> (yeet o $ Just (f' g')) >> prog+                    (Just f', Just g') -> yeet o (Just (f' g')) >> prog                     _                  -> return ()          -- TODO: Should we cancel asyncs here in finally block?         prog         yeet o Nothing-        wait fa+        _ <- wait fa         wait ga  -- | More general variant of `pure` with Monoid constraint.

@@ -16,7 +16,9 @@  import Control.Churro import Control.Concurrent.Async (wait)-import Control.Concurrent (Chan)+import Control.Concurrent (MVar, Chan)+import Control.Churro.Transport.MVar ()+import Control.Churro.Transport.MVar.Latest (Latest)  -- $setup -- @@ -63,6 +65,25 @@     where     maps    = map fromList $ zipWith zip updates updates     updates = map (take 2) (tails [0 :: Int ..])++-- | Combination of different transports:++mvarTest :: Churro () MVar Void Void+mvarTest = sourceIO @MVar (\cb -> cb (1 :: Int) >> cb 2 >> cb 3) >>> delay 1 >>> sinkPrint++latestTest :: Churro () Latest Void Void+latestTest = sourceIO @Latest (\cb -> cb (1 :: Int) >> cb 2 >> cb 3) >>> delay 1 >>> sinkPrint++-- | Should only output the latest values after sink is free to consume+-- +-- >>> runWait mvarTest+-- 1+-- 2+-- 3++-- >>> runWait latestTest+-- 1+-- 3  -- | Consumers terminaiting should kill sources from producing. --