pipes-fluid 0.3.1.0 → 0.4.0.0

raw patch · 7 files changed

+321/−321 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• pipes-fluid.cabal +3/−3
• src/Pipes/Fluid.hs +4/−4
• src/Pipes/Fluid/Impulse.hs +123/−0
• src/Pipes/Fluid/ImpulseIO.hs +161/−0
• src/Pipes/Fluid/React.hs +0/−123
• src/Pipes/Fluid/ReactIO.hs +0/−161
• test/Spec.hs +30/−30

@@ -1,5 +1,5 @@ name:                pipes-fluid-version:             0.3.1.0+version:             0.4.0.0 synopsis:            Reactively combines Producers so that a value is yielded as soon as possible. description:         Please see README.md homepage:            https://github.com/louispan/pipes-fluid#readme@@ -17,8 +17,8 @@   hs-source-dirs:      src   exposed-modules:     Pipes.Fluid                        Pipes.Fluid.Merge-                       Pipes.Fluid.React-                       Pipes.Fluid.ReactIO+                       Pipes.Fluid.Impulse+                       Pipes.Fluid.ImpulseIO                        Pipes.Fluid.Sync   build-depends:       base >= 4.7 && < 5                      , constraints >= 0.4 && < 1

@@ -1,11 +1,11 @@ module Pipes.Fluid     ( module Pipes.Fluid.Merge-    , module Pipes.Fluid.React-    , module Pipes.Fluid.ReactIO+    , module Pipes.Fluid.Impulse+    , module Pipes.Fluid.ImpulseIO     , module Pipes.Fluid.Sync     ) where  import Pipes.Fluid.Merge-import Pipes.Fluid.React-import Pipes.Fluid.ReactIO+import Pipes.Fluid.Impulse+import Pipes.Fluid.ImpulseIO import Pipes.Fluid.Sync

@@ -0,0 +1,123 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ImpredicativeTypes #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}++module Pipes.Fluid.Impulse+    ( Impulse(..)+    , module Pipes.Fluid.Merge+    ) where++import Control.Applicative+import Control.Lens+import Control.Monad.Trans.Class+import Data.These+import qualified Pipes as P+import Pipes.Fluid.Merge+import qualified Pipes.Prelude as PP++-- | The applicative instance of this combines multiple Producers reactively+-- ie, yields a value as soon as either or both of the input producers yields a value.+newtype Impulse m a = Impulse+    { impulsively :: P.Producer a m ()+    }++makeWrapped ''Impulse++instance Monad m =>+         Functor (Impulse m) where+    fmap f (Impulse as) = Impulse $ as P.>-> PP.map f+    {-# INLINABLE fmap #-}++-- | Impulseively combines two producers, given initial values to use when the producer is blocked/failed.+-- This only works for Alternative m where failure means there was no effects, eg. 'Control.Concurrent.STM', or @MonadTrans t => t STM@.+-- Be careful of monad transformers like ExceptT that hides the STM Alternative instance.+instance (Alternative m, Monad m) =>+         Applicative (Impulse m) where+    pure = Impulse . P.yield+    {-# INLINABLE pure #-}++    fs <*> as =+        Impulse $+        P.for (impulsively $ merge fs as) $ \r ->+            case r of+                Coupled _ f a -> P.yield $ f a+                -- never got anything from one of the signals, can't do anything yet.+                -- fail/retry/block until we get something from the other signal+                LeftOnly _ _ -> lift empty+                RightOnly _ _-> lift empty+    {-# INLINABLE (<*>) #-}++-- | Impulseively combines two producers, given initial values to use when the produce hasn't produced anything yet+-- Combine two signals, and returns a signal that emits+-- @Either bothfired (Either (leftFired, previousRight) (previousLeft, rightFired))@.+-- This only works for Alternative m where failure means there was no effects, eg. 'Control.Concurrent.STM', or @MonadTrans t => t STM@.+-- Be careful of monad transformers ExceptT that hides the STM Alternative instance.+instance (Alternative m, Monad m) => Merge (Impulse m) where+    merge' px_ py_ (Impulse xs_) (Impulse ys_) = Impulse $ go px_ py_ xs_ ys_+      where+        go px py xs ys = do+            -- use the Alternative of m, not P.Proxy+            r <- lift $ bothOrEither (P.next xs) (P.next ys)+            case r+                -- both fs and as have ended+                  of+                These (Left _) (Left _) -> pure ()+                -- @xs@ ended,                @ys@ failed/retry/blocked+                This (Left _) -> case px of+                    Nothing -> ys P.>-> PP.map (RightOnly OtherDead)+                    Just x -> ys P.>-> PP.map (Coupled (FromRight OtherDead) x)+                -- @xs@ failed/retry/blocked, @ys@ ended+                That (Left _) -> case py of+                    Nothing -> xs P.>-> PP.map (LeftOnly OtherDead)+                    Just y -> xs P.>-> PP.map (\x -> Coupled (FromLeft OtherDead) x y)+                -- @xs@ produced something,   @ys@ failed/retry/blocked+                This (Right (x, xs')) -> do+                    case py of+                        Nothing -> P.yield $ LeftOnly OtherLive x+                        Just y -> P.yield $ Coupled (FromLeft OtherLive) x y+                    go (Just x) py xs' ys+                -- @xs@ failed/retry/blocked, @ys@ produced something+                That (Right (y, ys')) -> do+                    case px of+                        Nothing -> P.yield $ RightOnly OtherLive y+                        Just x -> P.yield $ Coupled (FromRight OtherLive) x y+                    go px (Just y) xs ys'+                -- @xs@ produced something,   @ys@ ended+                These (Right (x, xs')) (Left _) ->+                    case py of+                        Nothing -> do+                            P.yield $ LeftOnly OtherDead x+                            xs' P.>-> PP.map (LeftOnly OtherDead)+                        Just y -> do+                            P.yield $ Coupled (FromLeft OtherDead) x y+                            xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)+                -- @fs@ ended,                @as@ produced something+                These (Left _) (Right (y, ys')) ->+                    case px of+                        Nothing -> do+                            P.yield $ RightOnly OtherDead y+                            ys' P.>-> PP.map (RightOnly OtherDead)+                        Just x -> do+                            P.yield $ Coupled (FromRight OtherDead) x y+                            ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)+                -- both @fs@ and @as@ produced something+                These (Right (x, xs')) (Right (y, ys')) -> do+                    P.yield $ Coupled FromBoth x y+                    go (Just x) (Just y) xs' ys'+    {-# INLINABLE merge' #-}++-- | Used internally by Impulse and ImpulseIO identifying which side (or both) returned values+bothOrEither :: Alternative f => f a -> f b -> f (These a b)+bothOrEither left right =+  (These <$> left <*> right)+  <|>+  (This <$> left)+  <|>+  (That <$> right)+{-# INLINABLE bothOrEither #-}

@@ -0,0 +1,161 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ImpredicativeTypes #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++module Pipes.Fluid.ImpulseIO+    ( ImpulseIO(..)+    , module Pipes.Fluid.Merge+    ) where++import Control.Applicative+import qualified Control.Concurrent.Async.Lifted.Safe as A+import qualified Control.Concurrent.STM as S+import Control.Lens+import Control.Monad.Base+import Control.Monad.Trans.Class+import Control.Monad.Trans.Control+import Data.Constraint.Forall (Forall)+import Data.These+import qualified Pipes as P+import Pipes.Fluid.Merge+import qualified Pipes.Prelude as PP++-- | The applicative instance of this combines multiple Producers reactively+-- ie, yields a value as soon as either or both of the input producers yields a value.+-- This creates two threads each time this combinator is used.+-- Warning: This means that the monadic effects are run in isolation from each other+-- so if the monad is something like (StateT s IO), then the state will alternate+-- between the two input producers, which is most likely not what you want.+newtype ImpulseIO m a = ImpulseIO+    { impulsivelyIO :: P.Producer a m ()+    }++makeWrapped ''ImpulseIO++instance Monad m => Functor (ImpulseIO m) where+  fmap f (ImpulseIO as) = ImpulseIO $ as P.>-> PP.map f+  {-# INLINABLE fmap #-}++instance (MonadBaseControl IO m, Forall (A.Pure m)) => Applicative (ImpulseIO m) where+    pure = ImpulseIO . P.yield+    {-# INLINABLE pure #-}++    -- 'ap' doesn't know about initial values+    fs <*> as = ImpulseIO $ P.for (impulsivelyIO $ merge fs as) $ \r ->+        case r of+            Coupled _ f a -> P.yield $ f a+            -- never got anything from one of the signals, can't do anything yet.+            -- drop the event+            LeftOnly _ _ -> pure ()+            RightOnly _ _-> pure ()+    {-# INLINABLE (<*>) #-}++-- | Reactively combines two producers, given initial values to use when the produce hasn't produced anything yet+-- Combine two signals, and returns a signal that emits+-- @Either bothfired (Either (leftFired, previousRight) (previousLeft, rightFired))@.+-- This creates two threads each time this combinator is used.+-- Warning: This means that the monadic effects are run in isolation from each other+-- so if the monad is something like (StateT s IO), then the state will alternate+-- between the two input producers, which is most likely not what you want.+-- This will be detect as a compile error due to use of Control.Concurrent.Async.Lifted.Safe+instance (MonadBaseControl IO m, Forall (A.Pure m)) => Merge (ImpulseIO m) where+    merge' px_ py_ (ImpulseIO xs_) (ImpulseIO ys_) = ImpulseIO $ do+        ax <- lift $ A.async $ P.next xs_+        ay <- lift $ A.async $ P.next ys_+        doMergeIO px_ py_ ax ay+      where+        doMergeIO :: (MonadBaseControl IO m, Forall (A.Pure m)) =>+             Maybe x+          -> Maybe y+          -> A.Async (Either () (x, P.Producer x m ()))+          -> A.Async (Either () (y, P.Producer y m ()))+          -> P.Producer (Merged x y) m ()+        doMergeIO px py ax ay = do+            r <-+                lift $+                liftBase . S.atomically $ bothOrEither (A.waitSTM ax) (A.waitSTM ay)+            case r of+                -- both @ax@ and @ay@ have ended+                These (Left _) (Left _) -> pure ()+                -- @ax@ ended,                @ay@ still waiting+                This (Left _) -> do+                    ry <- lift $ A.wait ay -- wait for @ay@ to return and then+                                           -- only use @ys@+                    case ry of+                        Left _ -> pure ()+                        Right (y, ys') -> case px of+                            Nothing -> do+                                P.yield $ RightOnly OtherDead y+                                ys' P.>-> PP.map (RightOnly OtherDead)+                            Just x -> do+                                P.yield $ Coupled (FromRight OtherDead) x y+                                ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)+                -- @ax@ still waiting,        @ay@ ended+                That (Left _) -> do+                    rx <- lift $ A.wait ax -- wait for @ax@ to retrun and then+                                           -- only use @xs@+                    case rx of+                        Left _ -> pure ()+                        Right (x, xs') -> case py of+                            Nothing -> do+                                P.yield $ LeftOnly OtherDead x+                                xs' P.>-> PP.map (LeftOnly OtherDead)+                            Just y -> do+                                P.yield $ Coupled (FromLeft OtherDead) x y+                                xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)+                -- @ax@ produced something,   @ay@ still waiting+                This (Right (x, xs')) -> do+                    case py of+                        Nothing -> P.yield $ LeftOnly OtherLive x+                        Just y -> P.yield $ Coupled (FromLeft OtherLive) x y+                    ax' <- lift $ A.async $ P.next xs'+                    doMergeIO (Just x) py ax' ay+                -- @ax@ still waiting,        @ay@ produced something+                That (Right (y, ys')) -> do+                    case px of+                        Nothing -> P.yield $ RightOnly OtherLive y+                        Just x -> P.yield $ Coupled (FromRight OtherLive) x y+                    ay' <- lift $ A.async $ P.next ys'+                    doMergeIO px (Just y) ax ay'+                -- @ax@ produced something,   @ay@ ended+                These (Right (x, xs')) (Left _) ->+                    case py of+                        Nothing -> do+                            P.yield $ LeftOnly OtherDead x+                            xs' P.>-> PP.map (LeftOnly OtherDead)+                        Just y -> do+                            P.yield $ Coupled (FromLeft OtherDead) x y+                            xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)+                -- @af@ ended,                @aa@ produced something+                These (Left _) (Right (y, ys')) ->+                    case px of+                        Nothing -> do+                            P.yield $ RightOnly OtherDead y+                            ys' P.>-> PP.map (RightOnly OtherDead)+                        Just x -> do+                            P.yield $ Coupled (FromRight OtherDead) x y+                            ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)+                -- both @fs@ and @as@ produced something+                These (Right (x, xs')) (Right (y, ys')) -> do+                    P.yield $ Coupled FromBoth x y+                    ax' <- lift $ A.async $ P.next xs'+                    ay' <- lift $ A.async $ P.next ys'+                    doMergeIO (Just x) (Just y) ax' ay'+    {-# INLINABLE merge' #-}++-- | Used internally by Impulse and ImpulseIO identifying which side (or both) returned values+bothOrEither :: Alternative f => f a -> f b -> f (These a b)+bothOrEither left right =+  (These <$> left <*> right)+  <|>+  (This <$> left)+  <|>+  (That <$> right)+{-# INLINABLE bothOrEither #-}

@@ -1,123 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE FunctionalDependencies #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE ImpredicativeTypes #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TypeFamilies #-}--module Pipes.Fluid.React-    ( React(..)-    , module Pipes.Fluid.Merge-    ) where--import Control.Applicative-import Control.Lens-import Control.Monad.Trans.Class-import Data.These-import qualified Pipes as P-import Pipes.Fluid.Merge-import qualified Pipes.Prelude as PP---- | The applicative instance of this combines multiple Producers reactively--- ie, yields a value as soon as either or both of the input producers yields a value.-newtype React m a = React-    { reactively :: P.Producer a m ()-    }--makeWrapped ''React--instance Monad m =>-         Functor (React m) where-    fmap f (React as) = React $ as P.>-> PP.map f-    {-# INLINABLE fmap #-}---- | Reactively combines two producers, given initial values to use when the producer is blocked/failed.--- This only works for Alternative m where failure means there was no effects, eg. 'Control.Concurrent.STM', or @MonadTrans t => t STM@.--- Be careful of monad transformers like ExceptT that hides the STM Alternative instance.-instance (Alternative m, Monad m) =>-         Applicative (React m) where-    pure = React . P.yield-    {-# INLINABLE pure #-}--    fs <*> as =-        React $-        P.for (reactively $ merge fs as) $ \r ->-            case r of-                Coupled _ f a -> P.yield $ f a-                -- never got anything from one of the signals, can't do anything yet.-                -- fail/retry/block until we get something from the other signal-                LeftOnly _ _ -> lift empty-                RightOnly _ _-> lift empty-    {-# INLINABLE (<*>) #-}---- | Reactively combines two producers, given initial values to use when the produce hasn't produced anything yet--- Combine two signals, and returns a signal that emits--- @Either bothfired (Either (leftFired, previousRight) (previousLeft, rightFired))@.--- This only works for Alternative m where failure means there was no effects, eg. 'Control.Concurrent.STM', or @MonadTrans t => t STM@.--- Be careful of monad transformers ExceptT that hides the STM Alternative instance.-instance (Alternative m, Monad m) => Merge (React m) where-    merge' px_ py_ (React xs_) (React ys_) = React $ go px_ py_ xs_ ys_-      where-        go px py xs ys = do-            -- use the Alternative of m, not P.Proxy-            r <- lift $ bothOrEither (P.next xs) (P.next ys)-            case r-                -- both fs and as have ended-                  of-                These (Left _) (Left _) -> pure ()-                -- @xs@ ended,                @ys@ failed/retry/blocked-                This (Left _) -> case px of-                    Nothing -> ys P.>-> PP.map (RightOnly OtherDead)-                    Just x -> ys P.>-> PP.map (Coupled (FromRight OtherDead) x)-                -- @xs@ failed/retry/blocked, @ys@ ended-                That (Left _) -> case py of-                    Nothing -> xs P.>-> PP.map (LeftOnly OtherDead)-                    Just y -> xs P.>-> PP.map (\x -> Coupled (FromLeft OtherDead) x y)-                -- @xs@ produced something,   @ys@ failed/retry/blocked-                This (Right (x, xs')) -> do-                    case py of-                        Nothing -> P.yield $ LeftOnly OtherLive x-                        Just y -> P.yield $ Coupled (FromLeft OtherLive) x y-                    go (Just x) py xs' ys-                -- @xs@ failed/retry/blocked, @ys@ produced something-                That (Right (y, ys')) -> do-                    case px of-                        Nothing -> P.yield $ RightOnly OtherLive y-                        Just x -> P.yield $ Coupled (FromRight OtherLive) x y-                    go px (Just y) xs ys'-                -- @xs@ produced something,   @ys@ ended-                These (Right (x, xs')) (Left _) ->-                    case py of-                        Nothing -> do-                            P.yield $ LeftOnly OtherDead x-                            xs' P.>-> PP.map (LeftOnly OtherDead)-                        Just y -> do-                            P.yield $ Coupled (FromLeft OtherDead) x y-                            xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)-                -- @fs@ ended,                @as@ produced something-                These (Left _) (Right (y, ys')) ->-                    case px of-                        Nothing -> do-                            P.yield $ RightOnly OtherDead y-                            ys' P.>-> PP.map (RightOnly OtherDead)-                        Just x -> do-                            P.yield $ Coupled (FromRight OtherDead) x y-                            ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)-                -- both @fs@ and @as@ produced something-                These (Right (x, xs')) (Right (y, ys')) -> do-                    P.yield $ Coupled FromBoth x y-                    go (Just x) (Just y) xs' ys'-    {-# INLINABLE merge' #-}---- | Used internally by React and ReactIO identifying which side (or both) returned values-bothOrEither :: Alternative f => f a -> f b -> f (These a b)-bothOrEither left right =-  (These <$> left <*> right)-  <|>-  (This <$> left)-  <|>-  (That <$> right)-{-# INLINABLE bothOrEither #-}

@@ -1,161 +0,0 @@-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE FunctionalDependencies #-}-{-# LANGUAGE GADTs #-}-{-# LANGUAGE ImpredicativeTypes #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE UndecidableInstances #-}--module Pipes.Fluid.ReactIO-    ( ReactIO(..)-    , module Pipes.Fluid.Merge-    ) where--import Control.Applicative-import qualified Control.Concurrent.Async.Lifted.Safe as A-import qualified Control.Concurrent.STM as S-import Control.Lens-import Control.Monad.Base-import Control.Monad.Trans.Class-import Control.Monad.Trans.Control-import Data.Constraint.Forall (Forall)-import Data.These-import qualified Pipes as P-import Pipes.Fluid.Merge-import qualified Pipes.Prelude as PP---- | The applicative instance of this combines multiple Producers reactively--- ie, yields a value as soon as either or both of the input producers yields a value.--- This creates two threads each time this combinator is used.--- Warning: This means that the monadic effects are run in isolation from each other--- so if the monad is something like (StateT s IO), then the state will alternate--- between the two input producers, which is most likely not what you want.-newtype ReactIO m a = ReactIO-    { reactivelyIO :: P.Producer a m ()-    }--makeWrapped ''ReactIO--instance Monad m => Functor (ReactIO m) where-  fmap f (ReactIO as) = ReactIO $ as P.>-> PP.map f-  {-# INLINABLE fmap #-}--instance (MonadBaseControl IO m, Forall (A.Pure m)) => Applicative (ReactIO m) where-    pure = ReactIO . P.yield-    {-# INLINABLE pure #-}--    -- 'ap' doesn't know about initial values-    fs <*> as = ReactIO $ P.for (reactivelyIO $ merge fs as) $ \r ->-        case r of-            Coupled _ f a -> P.yield $ f a-            -- never got anything from one of the signals, can't do anything yet.-            -- drop the event-            LeftOnly _ _ -> pure ()-            RightOnly _ _-> pure ()-    {-# INLINABLE (<*>) #-}---- | Reactively combines two producers, given initial values to use when the produce hasn't produced anything yet--- Combine two signals, and returns a signal that emits--- @Either bothfired (Either (leftFired, previousRight) (previousLeft, rightFired))@.--- This creates two threads each time this combinator is used.--- Warning: This means that the monadic effects are run in isolation from each other--- so if the monad is something like (StateT s IO), then the state will alternate--- between the two input producers, which is most likely not what you want.--- This will be detect as a compile error due to use of Control.Concurrent.Async.Lifted.Safe-instance (MonadBaseControl IO m, Forall (A.Pure m)) => Merge (ReactIO m) where-    merge' px_ py_ (ReactIO xs_) (ReactIO ys_) = ReactIO $ do-        ax <- lift $ A.async $ P.next xs_-        ay <- lift $ A.async $ P.next ys_-        doMergeIO px_ py_ ax ay-      where-        doMergeIO :: (MonadBaseControl IO m, Forall (A.Pure m)) =>-             Maybe x-          -> Maybe y-          -> A.Async (Either () (x, P.Producer x m ()))-          -> A.Async (Either () (y, P.Producer y m ()))-          -> P.Producer (Merged x y) m ()-        doMergeIO px py ax ay = do-            r <--                lift $-                liftBase . S.atomically $ bothOrEither (A.waitSTM ax) (A.waitSTM ay)-            case r of-                -- both @ax@ and @ay@ have ended-                These (Left _) (Left _) -> pure ()-                -- @ax@ ended,                @ay@ still waiting-                This (Left _) -> do-                    ry <- lift $ A.wait ay -- wait for @ay@ to return and then-                                           -- only use @ys@-                    case ry of-                        Left _ -> pure ()-                        Right (y, ys') -> case px of-                            Nothing -> do-                                P.yield $ RightOnly OtherDead y-                                ys' P.>-> PP.map (RightOnly OtherDead)-                            Just x -> do-                                P.yield $ Coupled (FromRight OtherDead) x y-                                ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)-                -- @ax@ still waiting,        @ay@ ended-                That (Left _) -> do-                    rx <- lift $ A.wait ax -- wait for @ax@ to retrun and then-                                           -- only use @xs@-                    case rx of-                        Left _ -> pure ()-                        Right (x, xs') -> case py of-                            Nothing -> do-                                P.yield $ LeftOnly OtherDead x-                                xs' P.>-> PP.map (LeftOnly OtherDead)-                            Just y -> do-                                P.yield $ Coupled (FromLeft OtherDead) x y-                                xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)-                -- @ax@ produced something,   @ay@ still waiting-                This (Right (x, xs')) -> do-                    case py of-                        Nothing -> P.yield $ LeftOnly OtherLive x-                        Just y -> P.yield $ Coupled (FromLeft OtherLive) x y-                    ax' <- lift $ A.async $ P.next xs'-                    doMergeIO (Just x) py ax' ay-                -- @ax@ still waiting,        @ay@ produced something-                That (Right (y, ys')) -> do-                    case px of-                        Nothing -> P.yield $ RightOnly OtherLive y-                        Just x -> P.yield $ Coupled (FromRight OtherLive) x y-                    ay' <- lift $ A.async $ P.next ys'-                    doMergeIO px (Just y) ax ay'-                -- @ax@ produced something,   @ay@ ended-                These (Right (x, xs')) (Left _) ->-                    case py of-                        Nothing -> do-                            P.yield $ LeftOnly OtherDead x-                            xs' P.>-> PP.map (LeftOnly OtherDead)-                        Just y -> do-                            P.yield $ Coupled (FromLeft OtherDead) x y-                            xs' P.>-> PP.map (\x' -> Coupled (FromLeft OtherDead) x' y)-                -- @af@ ended,                @aa@ produced something-                These (Left _) (Right (y, ys')) ->-                    case px of-                        Nothing -> do-                            P.yield $ RightOnly OtherDead y-                            ys' P.>-> PP.map (RightOnly OtherDead)-                        Just x -> do-                            P.yield $ Coupled (FromRight OtherDead) x y-                            ys' P.>-> PP.map (Coupled (FromRight OtherDead) x)-                -- both @fs@ and @as@ produced something-                These (Right (x, xs')) (Right (y, ys')) -> do-                    P.yield $ Coupled FromBoth x y-                    ax' <- lift $ A.async $ P.next xs'-                    ay' <- lift $ A.async $ P.next ys'-                    doMergeIO (Just x) (Just y) ax' ay'-    {-# INLINABLE merge' #-}---- | Used internally by React and ReactIO identifying which side (or both) returned values-bothOrEither :: Alternative f => f a -> f b -> f (These a b)-bothOrEither left right =-  (These <$> left <*> right)-  <|>-  (This <$> left)-  <|>-  (That <$> right)-{-# INLINABLE bothOrEither #-}

@@ -19,8 +19,8 @@ import Data.Maybe import qualified Pipes as P import qualified Pipes.Concurrent as PC-import qualified Pipes.Fluid.React as PF-import qualified Pipes.Fluid.ReactIO as PF+import qualified Pipes.Fluid.Impulse as PF+import qualified Pipes.Fluid.ImpulseIO as PF import qualified Pipes.Fluid.Sync as PF import qualified Pipes.Misc.Concurrent as PM import qualified Pipes.Misc.State.Strict as PM@@ -101,71 +101,71 @@                         PF.synchronously $                         (\a b -> (a, b, a + b)) <$> PF.Sync as <*> PF.Sync bs                 xs `shouldBe` (\(a, b) -> (a, b, a + b)) <$> zip data1 data2-        describe "React" $ do-            it "React STM: yield a value whenever any producer yields a value" $ do+        describe "Impulse" $ do+            it "Impulse STM: yield a value whenever any producer yields a value" $ do                 xs <-                     testSig' False $ \as bs ->                         PP.toListM $                         hoist atomically $-                        PF.reactively $-                        (\a b -> (a, b, a + b)) <$> PF.React as <*> PF.React bs+                        PF.impulsively $+                        (\a b -> (a, b, a + b)) <$> PF.Impulse as <*> PF.Impulse bs                 xs `shouldSatisfy` isBigger-            it "React IdentityT STM: reactively yields under 't STM'" $ do+            it "Impulse IdentityT STM: impulsively yields under 't STM'" $ do                 xs <-                     testSig' False $ \as bs ->                         runIdentityT $                         PP.toListM $                         hoist (hoist atomically) $-                        PF.reactively $-                        (\a b -> (a, b, a + b)) <$> (PF.React $ hoist lift as) <*>-                        (PF.React $ hoist lift bs)+                        PF.impulsively $+                        (\a b -> (a, b, a + b)) <$> (PF.Impulse $ hoist lift as) <*>+                        (PF.Impulse $ hoist lift bs)                 xs `shouldSatisfy` isBigger-            it "React StateT STM: reactively yields under 'StateT STM'" $ do+            it "Impulse StateT STM: impulsively yields under 'StateT STM'" $ do                 xs <-                     testSig' False $ \as bs ->                         (`evalStateT` (Model 0 0)) $                         PP.toListM $                         (hoist (hoist atomically) $-                         PF.reactively $+                         PF.impulsively $                          (\a b -> (a, b, a + b)) <$>-                         (PF.React $ hoist lift as P.>-> PM.store id counter1) <*>-                         (PF.React $ hoist lift bs P.>-> PM.store id counter2)) P.>->+                         (PF.Impulse $ hoist lift as P.>-> PM.store id counter1) <*>+                         (PF.Impulse $ hoist lift bs P.>-> PM.store id counter2)) P.>->                         PM.retrieve id                 xs `shouldSatisfy` isBigger-            it "React Merge: yield a Left/Right value depending on which producer yields a value" $ do+            it "Impulse Merge: yield a Left/Right value depending on which producer yields a value" $ do                 xs <-                     testSig' False $ \as bs ->                         PP.toListM $                         hoist atomically $-                        PF.reactively $-                        PF.React as `PF.merge`PF.React bs+                        PF.impulsively $+                        PF.Impulse as `PF.merge`PF.Impulse bs                 xs `shouldSatisfy` isDifferent-        describe "ReactIO" $ do-            it "React IO: reactively yield under IO using lifted-async" $ do+        describe "ImpulseIO" $ do+            it "Impulse IO: impulsively yield under IO using lifted-async" $ do                 xs <-                     testSig' False $ \as bs ->                         PP.toListM $-                        PF.reactivelyIO $-                        (\a b -> (a, b, a + b)) <$> (PF.ReactIO $ hoist atomically as) <*>-                        (PF.ReactIO $ hoist atomically bs)+                        PF.impulsivelyIO $+                        (\a b -> (a, b, a + b)) <$> (PF.ImpulseIO $ hoist atomically as) <*>+                        (PF.ImpulseIO $ hoist atomically bs)                 xs `shouldSatisfy` isBigger-            it "React IdentityT IO: reactively yield under 't IO' using lifted-async" $ do+            it "Impulse IdentityT IO: impulsively yield under 't IO' using lifted-async" $ do                 xs <-                     testSig' False $ \as bs ->                         runIdentityT $                         PP.toListM $-                        PF.reactivelyIO $-                        (\a b -> (a, b, a + b)) <$> (PF.ReactIO $ hoist (lift . atomically) as) <*>-                        (PF.ReactIO $ hoist (lift . atomically) bs)+                        PF.impulsivelyIO $+                        (\a b -> (a, b, a + b)) <$> (PF.ImpulseIO $ hoist (lift . atomically) as) <*>+                        (PF.ImpulseIO $ hoist (lift . atomically) bs)                 xs `shouldSatisfy` isBigger-            it "\nReact StateT IO is unsafe (lift-async detect this as a compile error)" $ do+            it "\nImpulse StateT IO is unsafe (lift-async detect this as a compile error)" $ do                 pure () `shouldReturn` ()-            it "ReactIO Merge: yield a Left/Right value depending on which producer yields a value" $ do+            it "ImpulseIO Merge: yield a Left/Right value depending on which producer yields a value" $ do                 xs <-                     testSig' False $ \as bs ->                         PP.toListM $-                        PF.reactivelyIO $-                        (PF.ReactIO $ hoist atomically as) `PF.merge` (PF.ReactIO $ hoist atomically bs)+                        PF.impulsivelyIO $+                        (PF.ImpulseIO $ hoist atomically as) `PF.merge` (PF.ImpulseIO $ hoist atomically bs)                 xs `shouldSatisfy` isDifferent  isBigger :: Ord a => [a] -> Bool