pipes-fluid 0.3.0.0 → 0.3.1.0

raw patch · 7 files changed

+295/−154 lines, 7 filesdep +thesedep ~pipes-miscPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: these

Dependency ranges changed: pipes-misc

API changes (from Hackage documentation)

Files

• pipes-fluid.cabal +8/−6
• src/Pipes/Fluid.hs +11/−0
• src/Pipes/Fluid/Alternative.hs +0/−11
• src/Pipes/Fluid/Merge.hs +98/−0
• src/Pipes/Fluid/React.hs +71/−53
• src/Pipes/Fluid/ReactIO.hs +104/−82
• test/Spec.hs +3/−2

@@ -1,5 +1,5 @@ name:                pipes-fluid-version:             0.3.0.0+version:             0.3.1.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@@ -15,10 +15,11 @@  library   hs-source-dirs:      src-  exposed-modules:     Pipes.Fluid.React-                     , Pipes.Fluid.ReactIO-                     , Pipes.Fluid.Sync-                     , Pipes.Fluid.Alternative+  exposed-modules:     Pipes.Fluid+                       Pipes.Fluid.Merge+                       Pipes.Fluid.React+                       Pipes.Fluid.ReactIO+                       Pipes.Fluid.Sync   build-depends:       base >= 4.7 && < 5                      , constraints >= 0.4 && < 1                      , lens >= 4 && < 5@@ -28,6 +29,7 @@                      , stm >= 2.4 && < 3                      , transformers >= 0.4 && < 6                      , transformers-base >= 0.4 && < 1+                     , these >= 0.7 && < 1   ghc-options:         -Wall   default-language:    Haskell2010 @@ -47,7 +49,7 @@                      , pipes >= 4 && < 5                      , pipes-concurrency >= 2 && < 3                      , pipes-fluid-                     , pipes-misc >= 0.2.0.0 && < 1+                     , pipes-misc >= 0.2.3.0 && < 1                      , stm >= 2.4 && < 3                      , transformers >= 0.4 && < 0.6   ghc-options:         -Wall -threaded -rtsopts -with-rtsopts=-N

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

@@ -1,11 +0,0 @@-module Pipes.Fluid.Alternative where--import Control.Applicative--bothOrEither :: Alternative f => f a -> f b -> f (Either (a, b) (Either a b))-bothOrEither left right =-  (curry Left <$> left <*> right)-  <|>-  (Right . Left <$> left)-  <|>-  (Right . Right <$> right)

@@ -0,0 +1,98 @@+{-# LANGUAGE DeriveGeneric #-}++module Pipes.Fluid.Merge where++import qualified GHC.Generics as G++-- | Differentiates whether a value from either or both producers.+-- In the case of one producer, additional identify if the other producer is live or dead.+data Source = FromBoth | FromLeft OtherStatus | FromRight OtherStatus+    deriving (Eq, Show, Ord, G.Generic)++-- | The other producer can be live (still yielding values), or dead+data OtherStatus = OtherLive | OtherDead+    deriving (Eq, Show, Ord, G.Generic)++-- | Differentiates when only one side is available (due to initial merge values of Nothing)+-- or if two values (one of which may be a previous values) are availabe.+data Merged a b =+    Coupled Source a b+    | LeftOnly OtherStatus a+    | RightOnly OtherStatus b+    deriving (Eq, Show, Ord, G.Generic)++-- | This can be used with 'Pipes.Prelude.takeWhile'+isBothLive :: Merged x y -> Bool+isBothLive (Coupled FromBoth _ _) = True+isBothLive (Coupled (FromLeft OtherLive) _ _) = True+isBothLive (Coupled (FromRight OtherLive) _ _) = True+isBothLive (LeftOnly OtherLive _) = True+isBothLive (RightOnly OtherLive _) = True+isBothLive _ = False+{-# INLINABLE isBothLive #-}++-- | This can be used with 'Pipes.Prelude.takeWhile'+isLeftLive :: Merged x y -> Bool+isLeftLive (Coupled FromBoth _ _) = True+isLeftLive (Coupled (FromLeft _) _ _) = True+isLeftLive (Coupled (FromRight OtherLive) _ _) = True+isLeftLive (LeftOnly _ _) = True+isLeftLive (RightOnly OtherLive _) = True+isLeftLive _ = False+{-# INLINABLE isLeftLive #-}++-- | This can be used with 'Pipes.Prelude.takeWhile'+isRightLive :: Merged x y -> Bool+isRightLive (Coupled FromBoth _ _) = True+isRightLive (Coupled (FromRight _) _ _) = True+isRightLive (Coupled (FromLeft OtherLive) _ _) = True+isRightLive (RightOnly _ _) = True+isRightLive (LeftOnly OtherLive _) = True+isRightLive _ = False+{-# INLINABLE isRightLive #-}++-- | This can be used with 'Pipes.Prelude.takeWhile'+isRightDead :: Merged x y -> Bool+isRightDead (Coupled (FromLeft OtherDead) _ _) = True+isRightDead (LeftOnly OtherDead _) = True+isRightDead _ = False+{-# INLINABLE isRightDead #-}++-- | This can be used with 'Pipes.Prelude.takeWhile'+isLeftDead :: Merged x y -> Bool+isLeftDead (Coupled (FromRight OtherDead) _ _) = True+isLeftDead (RightOnly OtherDead _) = True+isLeftDead _ = False+{-# INLINABLE isLeftDead #-}++class Merge f where+    merge' :: Maybe x -> Maybe y -> f x -> f y -> f (Merged x y)++merge :: Merge f => f x -> f y -> f (Merged x y)+merge = merge' Nothing Nothing+{-# INLINABLE merge #-}++-- | Keep only the values originated from the left, replacing other yields with Nothing.+-- This is useful when React is based on STM, since filtering with Producer STM results in+-- larger STM transactions which may result in blocking.+discreteLeft :: Merged x y -> Maybe x+discreteLeft (LeftOnly _ x) = Just x+discreteLeft (Coupled FromBoth  x _) = Just x+discreteLeft (Coupled (FromLeft _)  x _) = Just x+discreteLeft _ = Nothing++-- | Keep only the values originated from the right, replacing other yields with Nothing.+-- This is useful when React is based on STM, since filtering with Producer STM results in+-- larger STM transactions which may result in blocking.+discreteRight :: Merged x y -> Maybe y+discreteRight (RightOnly _ y) = Just y+discreteRight (Coupled FromBoth  _ y) = Just y+discreteRight (Coupled (FromRight _)  _ y) = Just y+discreteRight _ = Nothing++-- | Keep only the values originated from both, replacing other yields with Nothing.+-- This is useful when React is based on STM, since filtering with Producer STM results in+-- larger STM transactions which may result in blocking.+discreteBoth :: Merged x y -> Maybe (x, y)+discreteBoth (Coupled FromBoth  x y) = Just (x, y)+discreteBoth _ = Nothing

@@ -4,22 +4,21 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE ImpredicativeTypes #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoMonomorphismRestriction #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeFamilies #-}  module Pipes.Fluid.React     ( React(..)-    , merge-    , merge'+    , 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 qualified Pipes.Fluid.Alternative as PFA+import Pipes.Fluid.Merge import qualified Pipes.Prelude as PP  -- | The applicative instance of this combines multiple Producers reactively@@ -33,6 +32,7 @@ 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@.@@ -40,66 +40,84 @@ 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-                Left (f, a) -> P.yield $ f a-                Right (Left (f, Just a)) -> P.yield $ f a-                Right (Right (Just f, a)) -> P.yield $ f a+                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-                Right (Left (_, Nothing)) -> lift empty-                Right (Right (Nothing, _)) -> lift empty+                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.-merge' :: (Alternative m, Monad m) =>-  Maybe x-  -> Maybe y-  -> React m x-  -> React m y-  -> React m (Either (x, y) (Either (x, Maybe y) (Maybe x, y)))-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 $ PFA.bothOrEither (P.next xs) (P.next ys)-        case r-            -- both fs and as have ended-              of-            Left (Left _, Left _) -> pure ()-            -- @xs@ ended,                @ys@ failed/retry/blocked-            Right (Left (Left _)) -> ys P.>-> PP.map useRight-            -- @xs@ failed/retry/blocked, @ys@ ended-            Right (Right (Left _)) -> xs P.>-> PP.map useLeft-            -- @xs@ produced something,   @ys@ failed/retry/blocked-            Right (Left (Right (x, xs'))) -> do-                P.yield $ Right (Left (x, py))-                go (Just x) py xs' ys-            -- @xs@ failed/retry/blocked, @ys@ produced something-            Right (Right (Right (y, ys'))) -> do-                P.yield $ useRight y-                go px (Just y) xs ys'-            -- @xs@ produced something,   @ys@ ended-            Left (Right (x, xs'), Left _) -> do-                P.yield $ useLeft x-                xs' P.>-> PP.map useLeft-            -- @fs@ ended,                @as@ produced something-            Left (Left _, Right (y, ys')) -> do-                P.yield $ useRight y-                ys' P.>-> PP.map useRight-            -- both @fs@ and @as@ produced something-            Left (Right (x, xs'), Right (y, ys')) -> do-                P.yield $ Left (x, y)-                go (Just x) (Just y) xs' ys'+instance (Alternative m, Monad m) => Merge (React m) where+    merge' px_ py_ (React xs_) (React ys_) = React $ go px_ py_ xs_ ys_       where-        useRight y = Right (Right (px, y))-        useLeft x = Right (Left (x, py))+        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' #-} --- | A simpler version of merge', with the initial values as Nothing-merge :: (Alternative m, Monad m) => React m x -> React m y -> React m (Either (x, y) (Either (x, Maybe y) (Maybe x, y)))-merge = merge' Nothing Nothing+-- | 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 #-}

@@ -4,7 +4,6 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE ImpredicativeTypes #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE NoMonomorphismRestriction #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeFamilies #-}@@ -12,10 +11,10 @@  module Pipes.Fluid.ReactIO     ( ReactIO(..)-    , mergeIO-    , mergeIO'+    , 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@@ -23,8 +22,9 @@ import Control.Monad.Trans.Class import Control.Monad.Trans.Control import Data.Constraint.Forall (Forall)+import Data.These import qualified Pipes as P-import qualified Pipes.Fluid.Alternative as PFA+import Pipes.Fluid.Merge import qualified Pipes.Prelude as PP  -- | The applicative instance of this combines multiple Producers reactively@@ -41,19 +41,21 @@  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 $ mergeIO fs as) $ \r ->+    fs <*> as = ReactIO $ P.for (reactivelyIO $ merge fs as) $ \r ->         case r of-            Left (f, a) -> P.yield $ f a-            Right (Left (f, Just a)) -> P.yield $ f a-            Right (Right (Just f, a)) -> P.yield $ f a+            Coupled _ f a -> P.yield $ f a             -- never got anything from one of the signals, can't do anything yet.             -- drop the event-            Right (Left (_, Nothing)) -> pure ()-            Right (Right (Nothing, _)) -> pure ()+            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@@ -63,77 +65,97 @@ -- 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-mergeIO' :: (MonadBaseControl IO m, Forall (A.Pure m)) =>-     Maybe x-  -> Maybe y-  -> ReactIO m x-  -> ReactIO m y-  -> ReactIO m (Either (x, y) (Either (x, Maybe y) (Maybe x, y)))-mergeIO' 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 (Either (x, y) (Either (x, Maybe y) (Maybe x, y))) m ()-    doMergeIO px py ax ay = do-        r <--            lift $-            liftBase . S.atomically $ PFA.bothOrEither (A.waitSTM ax) (A.waitSTM ay)-        case r of-            Left (Left _, Left _) -> pure () -- both @ax@ and @ay@ have ended-            -- @ax@ ended,                @ay@ still waiting-            Right (Left (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') -> do-                        P.yield $ useRight y'-                        ys' P.>-> PP.map useRight-            -- @ax@ still waiting,        @ay@ ended-            Right (Right (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') -> do-                        P.yield $ useLeft x'-                        xs' P.>-> PP.map useLeft-            -- @ax@ produced something,   @ay@ still waiting-            Right (Left (Right (x, xs'))) -> do-                P.yield $ useLeft x-                ax' <- lift $ A.async $ P.next xs'-                doMergeIO (Just x) py ax' ay-            -- @ax@ still waiting,        @ay@ produced something-            Right (Right (Right (y, ys'))) -> do-                P.yield $ useRight y-                ay' <- lift $ A.async $ P.next ys'-                doMergeIO px (Just y) ax ay'-            -- @ax@ produced something,   @ay@ ended-            Left (Right (x, xs'), Left _) -> do-                P.yield $ useLeft x-                xs' P.>-> PP.map useLeft-            -- @af@ ended,                @aa@ produced something-            Left (Left _, Right (y, ys')) -> do-                P.yield $ useRight y-                ys' P.>-> PP.map useRight-            -- both @fs@ and @as@ produced something-            Left (Right (x, xs'), Right (y, ys')) -> do-                P.yield $ Left (x, y)-                ax' <- lift $ A.async $ P.next xs'-                ay' <- lift $ A.async $ P.next ys'-                doMergeIO (Just x) (Just y) ax' ay'+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-        useRight y = Right (Right (px, y))-        useLeft x = Right (Left (x, py))+        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' #-} -mergeIO :: (MonadBaseControl IO m, Forall (A.Pure m))-    => ReactIO m x-    -> ReactIO m y-    -> ReactIO m (Either (x, y) (Either (x, Maybe y) (Maybe x, y)))-mergeIO = mergeIO' Nothing Nothing+-- | 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 #-}

@@ -22,7 +22,8 @@ import qualified Pipes.Fluid.React as PF import qualified Pipes.Fluid.ReactIO as PF import qualified Pipes.Fluid.Sync as PF-import qualified Pipes.Misc as PM+import qualified Pipes.Misc.Concurrent as PM+import qualified Pipes.Misc.State.Strict as PM import qualified Pipes.Prelude as PP import Test.Hspec @@ -164,7 +165,7 @@                     testSig' False $ \as bs ->                         PP.toListM $                         PF.reactivelyIO $-                        (PF.ReactIO $ hoist atomically as) `PF.mergeIO` (PF.ReactIO $ hoist atomically bs)+                        (PF.ReactIO $ hoist atomically as) `PF.merge` (PF.ReactIO $ hoist atomically bs)                 xs `shouldSatisfy` isDifferent  isBigger :: Ord a => [a] -> Bool