bluefin-internal 0.0.7.0 → 0.0.8.0

raw patch · 5 files changed

+274/−10 lines, 5 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• bluefin-internal.cabal +1/−1
• src/Bluefin/Internal.hs +54/−7
• src/Bluefin/Internal/Examples.hs +213/−1
• src/Bluefin/Internal/Pipes.hs +1/−1

@@ -1,3 +1,8 @@+## 0.0.8.0++Add `Consume`, `await`, `consumeStream`, `zipCoroutines`,+`consumeEach`, and associated examples.+ ## 0.0.7.0  Add `withEffToIO'`, `race`, `connectCoroutines`, `receiveStream`,

@@ -1,6 +1,6 @@ cabal-version:      3.0 name:               bluefin-internal-version:            0.0.7.0+version:            0.0.8.0 license:            MIT license-file:       LICENSE author:             Tom Ellis

@@ -13,6 +13,7 @@ import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Exception (throwIO, tryJust) import qualified Control.Exception+import Control.Monad (forever) import Control.Monad.Base (MonadBase (liftBase)) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.IO.Unlift (MonadUnliftIO, withRunInIO)@@ -112,7 +113,7 @@  -- | Connect two coroutines.  Their execution is interleaved by -- exchanging @a@s and @b@s. When the first yields its first @a@ it--- starts the second (which is waiting to receive an @a@).+-- starts the second (which is awaiting an @a@). connectCoroutines ::   forall es a b r.   (forall e. Coroutine a b e -> Eff (e :& es) r) ->@@ -137,12 +138,40 @@    race (useImplWithin t1) (useImplWithin t2) io +-- | Old name for 'consumeStream'.  @receiveStream@ will be deprecated+-- in a future version. receiveStream ::-  (forall e. Coroutine () a e -> Eff (e :& es) r) ->+  (forall e. Consume a e -> Eff (e :& es) r) ->   (forall e. Stream a e -> Eff (e :& es) r) ->   Eff es r-receiveStream r s = connectCoroutines r (\() -> s)+receiveStream = consumeStream +consumeStream ::+  -- | Each 'await' from the @Consume@ ...+  (forall e. Consume a e -> Eff (e :& es) r) ->+  -- | ... receives the value 'yield'ed from the @Stream@+  (forall e. Stream a e -> Eff (e :& es) r) ->+  Eff es r+consumeStream r s = connectCoroutines r (\() -> s)++zipCoroutines ::+  (e1 :> es) =>+  Coroutine (a1, a2) b e1 ->+  (forall e. Coroutine a1 b e -> Eff (e :& es) r) ->+  (forall e. Coroutine a2 b e -> Eff (e :& es) r) ->+  -- | ͘+  Eff es r+zipCoroutines c m1 m2 = do+  connectCoroutines m1 $ \a1 c1 -> do+    connectCoroutines (useImplWithin m2) $ \a2 c2 -> do+      evalState (a1, a2) $ \ass -> do+        forever $ do+          as <- get ass+          b' <- yieldCoroutine c as+          a1' <- yieldCoroutine c1 b'+          a2' <- yieldCoroutine c2 b'+          put ass (a1', a2')+ instance (e :> es) => MonadBase IO (EffReader (IOE e) es) where   liftBase = liftIO @@ -271,6 +300,8 @@ -- @a@ and then expects values of type @()@. type Stream a = Coroutine a () +type Consume a = Coroutine () a+ -- | You can define a @Handle@ instance for your compound handles.  As -- an example, an "application" handle with a dynamic effect for -- database queries, a concrete effect for application state and a@@ -400,10 +431,10 @@ -- Right "No exception thrown" -- @ throw ::-  (ex :> es) =>-  Exception e ex ->+  (e :> es) =>+  Exception ex e ->   -- | Value to throw-  e ->+  ex ->   Eff es a throw (UnsafeMkException throw_) e = UnsafeMkEff (throw_ e) @@ -488,7 +519,6 @@   (a -> Eff es ()) ->   -- | Run the body   (a -> Eff es b) ->-  -- |   Eff es b bracket before after body =   UnsafeMkEff $@@ -722,6 +752,23 @@     ii <- get i     yield st (ii, s)     put i (ii + 1)++-- | A version of 'forEach' specialized to @Consume@.  Every time the+-- @Consume@ is used to 'await' a @b@, feed it the one created by the+-- handler.+consumeEach ::+  -- | Body+  ( forall e.+    Consume b e ->+    Eff (e :& es) r+  ) ->+  -- | Value to send to each @await@ in the body.+  Eff es b ->+  Eff es r+consumeEach k e = forEach k (\() -> e)++await :: (e :> es) => Consume a e -> Eff es a+await r = yieldCoroutine r ()  type EarlyReturn = Exception 

@@ -15,7 +15,7 @@ import qualified Bluefin.Internal.Pipes as P import Control.Exception (IOException) import qualified Control.Exception-import Control.Monad (forever, unless, when)+import Control.Monad (forever, replicateM_, unless, when) import Control.Monad.IO.Class (liftIO) import Data.Foldable (for_) import Data.Monoid (Any (Any, getAny))@@ -169,6 +169,189 @@    forEach formattedLines $ \line -> effIO io (putStrLn line) +awaitList ::+  (e :> es) =>+  [a] ->+  IOE e ->+  (forall e1. Consume a e1 -> Eff (e1 :& es) ()) ->+  Eff es ()+awaitList l io k = evalState l $ \s -> do+  withJump $ \done ->+    bracket+      (pure ())+      (\() -> effIO io (putStrLn "Released"))+      $ \() -> do+        consumeEach (useImplWithin k) $ do+          (x, xs) <-+            get s >>= \case+              [] -> jumpTo done+              x : xs -> pure (x, xs)+          put s xs+          pure x++takeRec ::+  (e3 :> es) =>+  Int ->+  (forall e. Consume a e -> Eff (e :& es) ()) ->+  Consume a e3 ->+  Eff es ()+takeRec n k rec =+  withJump $ \done -> evalState n $ \s -> consumeEach (useImplWithin k) $ do+    s' <- get s+    if s' <= 0+      then jumpTo done+      else do+        modify s (subtract 1)+        await rec++mapRec ::+  (e :> es) =>+  (a -> b) ->+  (forall e1. Consume b e1 -> Eff (e1 :& es) ()) ->+  Consume a e ->+  Eff es ()+mapRec f = traverseRec (pure . f)++traverseRec ::+  (e :> es) =>+  (a -> Eff es b) ->+  (forall e1. Consume b e1 -> Eff (e1 :& es) ()) ->+  Consume a e ->+  Eff es ()+traverseRec f k rec = forEach k $ \() -> do+  r <- await rec+  f r++awaitUsage ::+  (e1 :> es, e2 :> es) =>+  IOE e1 ->+  (forall e. Consume () e -> Eff (e :& es) ()) ->+  Consume Int e2 ->+  Eff es ()+awaitUsage io x = do+  mapRec (* 11) $+    mapRec (subtract 1) $+      takeRec 3 $+        traverseRec (effIO io . print) $+          useImplWithin x++awaitExample :: IO ()+awaitExample = runEff $ \io -> do+  awaitList [1 :: Int ..] io $ awaitUsage io $ \rec -> do+    replicateM_ 5 (await rec)++consumeStreamExample :: IO (Either String String)+consumeStreamExample = runEff $ \io -> do+  try $ \ex -> do+    consumeStream+      ( \r ->+          bracket+            (effIO io (putStrLn "Starting 2"))+            (\_ -> effIO io (putStrLn "Leaving 2"))+            $ \_ -> do+              for_ [1 :: Int .. 100] $ \n -> do+                b' <- await r+                effIO+                  io+                  ( putStrLn+                      ("Consumed body " ++ show b' ++ " at time " ++ show n)+                  )+              pure "Consumer finished first"+      )+      ( \y -> bracket+          (effIO io (putStrLn "Starting 1"))+          (\_ -> effIO io (putStrLn "Leaving 1"))+          $ \_ -> do+            for_ [1 :: Int .. 10] $ \n -> do+              effIO io (putStrLn ("Sending " ++ show n))+              yield y n+              when (n > 5) $ do+                effIO io (putStrLn "Aborting...")+                throw ex "Aborted"++            pure "Yielder finished first"+      )++consumeStreamExample2 :: IO ()+consumeStreamExample2 = runEff $ \io -> do+  let counter yeven yodd = for_ [0 :: Int .. 10] $ \i -> do+        if even i+          then yield yeven i+          else yield yodd i++  let foo yeven =+        consumeStream+          ( \r -> forever $ do+              i <- await r+              effIO io (putStrLn ("Odd: " ++ show i))+          )+          (counter yeven)++  let bar =+        consumeStream+          ( \r -> forever $ do+              i <- await r+              effIO io (putStrLn ("Even: " ++ show i))+          )+          foo++  bar++connectExample :: IO (Either String String)+connectExample = runEff $ \io -> do+  try $ \ex -> do+    connectCoroutines+      ( \y -> bracket+          (effIO io (putStrLn "Starting 1"))+          (\_ -> effIO io (putStrLn "Leaving 1"))+          $ \_ -> do+            for_ [1 :: Int .. 10] $ \n -> do+              effIO io (putStrLn ("Sending " ++ show n))+              yield y n+              when (n > 5) $ do+                effIO io (putStrLn "Aborting...")+                throw ex "Aborted"++            pure "Yielder finished first"+      )+      ( \binit r ->+          bracket+            (effIO io (putStrLn "Starting 2"))+            (\_ -> effIO io (putStrLn "Leaving 2"))+            $ \_ -> do+              effIO io (putStrLn ("Consumed intial " ++ show binit))+              for_ [1 :: Int .. 100] $ \n -> do+                b' <- await r+                effIO+                  io+                  ( putStrLn+                      ("Consumed body " ++ show b' ++ " at time " ++ show n)+                  )+              pure "Consumer finished first"+      )++zipCoroutinesExample :: IO ()+zipCoroutinesExample = runEff $ \io -> do+  let m1 y = do+        r <- yieldCoroutine y 1+        evalState r $ \rs -> do+          for_ [1 .. 10 :: Int] $ \i -> do+            r' <- get rs+            r'' <- yieldCoroutine y (r' + i)+            put rs r''++  let m2 y = do+        r <- yieldCoroutine y 1+        evalState r $ \rs -> do+          for_ [1 .. 5 :: Int] $ \i -> do+            r' <- get rs+            r'' <- yieldCoroutine y (r' - i)+            put rs r''++  forEach (\c -> zipCoroutines c m1 m2) $ \i@(i1, i2) -> do+    effIO io (print i)+    pure (i1 + i2)+ -- Count the number of (strictly) positives and (strictly) negatives -- in a list, unless we see a zero, in which case we bail with an -- error message.@@ -665,3 +848,32 @@ pipesExample2 :: IO String pipesExample2 = runEff $ \io -> runEffect $ do   stdinLn io >-> takeWhile' (/= "quit") >-> stdoutLn io++-- Acquiring resource+-- 1+-- 2+-- 3+-- 4+-- 5+-- Releasing resource+-- Finishing+promptCoroutine :: IO ()+promptCoroutine = runEff $ \io -> do+  -- consumeStream connects a consumer to a producer+  consumeStream+    -- Like a pipes Consumer.  Prints the first five elements it+    -- awaits.+    ( \r -> for_ [1 :: Int .. 5] $ \_ -> do+        v <- await r+        effIO io (print v)+    )+    -- Like a pipes Producer. Yields successive integers indefinitely.+    -- Unlike in pipes, we can simply use Bluefin's standard bracket+    -- for prompt release of a resource+    ( \y ->+        bracket+          (effIO io (putStrLn "Acquiring resource"))+          (\_ -> effIO io (putStrLn "Releasing resource"))+          (\_ -> for_ [1 :: Int ..] $ \i -> yield y i)+    )+  effIO io (putStrLn "Finishing")

@@ -1,6 +1,6 @@ module Bluefin.Internal.Pipes where -import Bluefin.Internal hiding (yield)+import Bluefin.Internal hiding (yield, await) import qualified Bluefin.Internal import Control.Monad (forever) import Data.Foldable (for_)