diff --git a/Disposable.hs b/Disposable.hs new file mode 100644 --- /dev/null +++ b/Disposable.hs @@ -0,0 +1,83 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE Safe #-} + +module Disposable ( Disposable(EmptyDisposable) + , newDisposable + , dispose + , DisposableSet + , newDisposableSet + , addDisposable + , removeDisposable + , toDisposable + ) where + +import Control.Applicative +import Control.Monad +import Data.Maybe +import Data.Foldable as F +import Data.Sequence as Seq +import Data.IORef +import Data.Unique + +-- | Allows disposal of a resource by running an action in the monad @m@. +data Disposable where + EmptyDisposable :: Disposable + Disposable :: Unique -> IORef (Maybe (IO ())) -> Disposable + +instance Eq Disposable where + EmptyDisposable == EmptyDisposable = True + (Disposable u _) == (Disposable u' _) = u == u' + _ == _ = False + +-- | Disposes a disposable. +dispose :: Disposable -> IO () +dispose EmptyDisposable = return () +dispose (Disposable _ mref) = do + m <- atomicModifyIORef mref $ \m -> (Nothing, m) + fromMaybe (return ()) m + +-- | Creates a disposable which runs the given action upon disposal. +newDisposable :: IO () -> IO Disposable +newDisposable action = do + u <- newUnique + mref <- newIORef $ Just action + return $ Disposable u mref + +-- | @Just s@ when not yet disposed. @Nothing@ after disposal. +type MaybeSet = Maybe (Seq Disposable) + +-- | A synchronized set of disposables. +newtype DisposableSet = DisposableSet (IORef MaybeSet) + +-- | Creates a set of disposables. +newDisposableSet :: IO DisposableSet +newDisposableSet = do + mref <- newIORef $ Just Seq.empty + return $ DisposableSet mref + +-- | Converts a set of disposables into a disposable. +-- The constructed disposable will dispose of all disposables in the set. +toDisposable :: DisposableSet -> IO Disposable +toDisposable (DisposableSet mref) = + let disposeSet = F.mapM_ dispose + action = do + m <- atomicModifyIORef mref $ \m -> (Nothing, m) + maybe (return ()) disposeSet m + in newDisposable action + +-- | Adds a disposable to a set. +addDisposable :: DisposableSet -> Disposable -> IO () +addDisposable _ EmptyDisposable = return () +addDisposable (DisposableSet mref) d = + let addDisposable' Nothing = (Nothing, True) + addDisposable' (Just s) = (Just $ s |> d, False) + in do + b <- atomicModifyIORef mref addDisposable' + when b $ dispose d + +-- | Removes a disposable from a set. +removeDisposable :: DisposableSet -> Disposable -> IO () +removeDisposable _ EmptyDisposable = return () +removeDisposable (DisposableSet mref) d = + let removeDisposable' = liftM $ Seq.filter (/= d) + in atomicModifyIORef mref $ \m -> (removeDisposable' m, ()) diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,8 @@ +**Copyright (c) 2013 Justin Spahr-Summers.** +**All rights reserved.** + +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. diff --git a/RxHaskell.cabal b/RxHaskell.cabal new file mode 100644 --- /dev/null +++ b/RxHaskell.cabal @@ -0,0 +1,33 @@ +name: RxHaskell +version: 0.1 +stability: alpha +synopsis: Reactive Extensions for Haskell +description: An implementation of functional reactive programming based on Microsoft's Reactive Extensions for .NET: . + . + RxHaskell offers a monadic API, making it easier to interleave side effects and imperative-style code. +homepage: https://github.com/jspahrsummers/RxHaskell +bug-reports: https://github.com/jspahrsummers/RxHaskell/issues +category: FRP +author: Justin Spahr-Summers +maintainer: justin.spahrsummers@gmail.com +license: MIT +license-file: LICENSE.md +copyright: Copyright (c) 2013 Justin Spahr-Summers +cabal-version: >= 1.14 +build-type: Simple + +library + default-language: Haskell2010 + build-depends: base >= 4.5 && < 5, + containers >= 0.4, + stm >= 2.4, + transformers >= 0.3 + exposed-modules: Disposable, + Scheduler, Scheduler.Internal, Scheduler.Main, Scheduler.Unsafe, + Signal, Signal.Channel, Signal.Command, Signal.Connection, Signal.Event, Signal.Operators, Signal.Scheduled + Signal.Subscriber, Signal.Subscriber.Internal + default-extensions: + +source-repository head + type: git + location: git://github.com/jspahrsummers/RxHaskell.git diff --git a/Scheduler.hs b/Scheduler.hs new file mode 100644 --- /dev/null +++ b/Scheduler.hs @@ -0,0 +1,15 @@ +{-# LANGUAGE Safe #-} + +module Scheduler ( Scheduler(schedule) + , SchedulerIO + , BackgroundScheduler + , newScheduler + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Scheduler.Internal + +-- | Creates a new background scheduler. +newScheduler :: IO BackgroundScheduler +newScheduler = BackgroundScheduler <$> atomically newTQueue diff --git a/Scheduler/Internal.hs b/Scheduler/Internal.hs new file mode 100644 --- /dev/null +++ b/Scheduler/Internal.hs @@ -0,0 +1,101 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE Safe #-} + +module Scheduler.Internal ( SchedulerIO(..) + , unsafeRunSchedulerIO + , Scheduler(..) + , BackgroundScheduler(..) + , ScheduledAction + , newScheduledAction + , executeScheduledAction + ) where + +import Control.Applicative +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.IO.Class +import Data.Functor +import Data.IORef +import Disposable + +-- | An 'IO' computation that must be performed in a scheduler of type @s@. +data SchedulerIO s a where + SchedulerIO :: Scheduler s => IO a -> SchedulerIO s a + +-- | Extracts the underlying 'IO' action from a 'SchedulerIO' action. +-- +-- This can be unsafe because the type system does not have enough information to determine +-- whether the calling code is running on an appropriate scheduler. +unsafeRunSchedulerIO :: Scheduler s => SchedulerIO s a -> IO a +unsafeRunSchedulerIO (SchedulerIO action) = action + +instance Functor (SchedulerIO s) where + fmap f (SchedulerIO action) = SchedulerIO $ fmap f action + +instance Scheduler s => Monad (SchedulerIO s) where + return = SchedulerIO . return + (SchedulerIO action) >>= f = + SchedulerIO $ do + v <- action + unsafeRunSchedulerIO $ f v + +instance Scheduler s => MonadIO (SchedulerIO s) where + liftIO = SchedulerIO + +instance Scheduler s => Applicative (SchedulerIO s) where + pure = return + (<*>) = ap + +-- | Represents an action on a scheduler, along with a flag indicating whether it should be canceled. +data ScheduledAction s where + ScheduledAction :: Scheduler s => IORef Bool -> SchedulerIO s () -> ScheduledAction s + +-- | Creates a new scheduled action, and returns a disposable which can be used to cancel it. +newScheduledAction :: Scheduler s => SchedulerIO s () -> IO (ScheduledAction s, Disposable) +newScheduledAction action = do + ref <- newIORef False + d <- newDisposable $ atomicModifyIORef ref $ const (True, ()) + return (ScheduledAction ref action, d) + +-- | Represents a queue of 'IO' actions which can be executed in FIFO order. +class Scheduler s where + -- | Schedules an action on the scheduler. Returns a disposable which can be used to cancel it. + schedule :: s -> SchedulerIO s () -> IO Disposable + + -- | Executes all current and future actions enqueued on the given scheduler. + schedulerMain :: s -> IO () + +-- | A scheduler which runs enqueued actions in a dedicated background thread. +newtype BackgroundScheduler = BackgroundScheduler (TQueue (ScheduledAction BackgroundScheduler)) + +instance Scheduler BackgroundScheduler where + schedule s@(BackgroundScheduler q) action = do + (sa, d) <- newScheduledAction action + + let schedule' = do + e <- isEmptyTQueue q + writeTQueue q sa + return e + + e <- atomically schedule' + + -- If the queue was previously empty, spin up a thread for the scheduler. + when e $ void $ forkIO $ schedulerMain s + + return d + + schedulerMain s@(BackgroundScheduler q) = do + m <- atomically $ tryReadTQueue q + + -- If the queue is empty, let this thread die. + maybe (return ()) (executeScheduledAction s) m + +-- | Executes the given action, then re-enters 'schedulerMain'. +executeScheduledAction :: Scheduler s => s -> ScheduledAction s -> IO () +executeScheduledAction s (ScheduledAction ref (SchedulerIO action)) = do + d <- readIORef ref + unless d action + + yield + schedulerMain s diff --git a/Scheduler/Main.hs b/Scheduler/Main.hs new file mode 100644 --- /dev/null +++ b/Scheduler/Main.hs @@ -0,0 +1,49 @@ +{-# LANGUAGE Trustworthy #-} + +module Scheduler.Main ( MainScheduler + , getMainScheduler + , runMainScheduler + ) where + +import Control.Concurrent +import Control.Concurrent.STM +import Data.IORef +import Scheduler +import Scheduler.Internal +import System.IO.Unsafe + +-- | A scheduler which runs enqueued actions on the main thread. +newtype MainScheduler = MainScheduler (TQueue (ScheduledAction MainScheduler)) + +instance Scheduler MainScheduler where + schedule (MainScheduler q) action = do + (sa, d) <- newScheduledAction action + atomically $ writeTQueue q sa + return d + + schedulerMain s@(MainScheduler q) = do + sa <- atomically $ readTQueue q + executeScheduledAction s sa + +-- ohai global variable +mainSchedulerRef :: IORef MainScheduler +{-# NOINLINE mainSchedulerRef #-} +mainSchedulerRef = + unsafePerformIO $ do + q <- atomically newTQueue + newIORef $ MainScheduler q + +-- | Returns a scheduler representing the main thread. +-- +-- Note that 'runMainScheduler' must be called for enqueued actions to actually execute. +getMainScheduler :: IO MainScheduler +getMainScheduler = readIORef mainSchedulerRef + +-- | Runs the main scheduler indefinitely using the current thread. +-- The current thread will be bound if possible. +runMainScheduler :: IO () +runMainScheduler = + let run = getMainScheduler >>= schedulerMain + in if rtsSupportsBoundThreads + then runInBoundThread run + else run diff --git a/Scheduler/Unsafe.hs b/Scheduler/Unsafe.hs new file mode 100644 --- /dev/null +++ b/Scheduler/Unsafe.hs @@ -0,0 +1,12 @@ +{-# LANGUAGE Unsafe #-} + +module Scheduler.Unsafe ( + ) where + +import Scheduler.Internal +import System.IO.Unsafe (unsafePerformIO) + +-- | Unsafely executes a 'SchedulerIO' action and shows the result. +-- This is for /DEBUGGING PURPOSES ONLY/. +instance (Scheduler s, Show v) => Show (SchedulerIO s v) where + show action = show $ unsafePerformIO $ unsafeRunSchedulerIO action diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,3 @@ +import Distribution.Simple + +main = defaultMain diff --git a/Signal.hs b/Signal.hs new file mode 100644 --- /dev/null +++ b/Signal.hs @@ -0,0 +1,206 @@ +{-# LANGUAGE GADTs #-} +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal ( Signal + , signal + , subscribe + , (>>:) + , never + , Signal.empty + , Event(..) + , Disposable + , Scheduler + , SchedulerIO + ) where + +import Control.Applicative +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.IO.Class +import Control.Monad.Zip +import Data.IORef +import Data.Monoid +import Data.Sequence as Seq +import Data.Word +import Disposable +import Prelude hiding (length, drop, zip) +import Scheduler.Internal +import Signal.Event +import Signal.Subscriber +import Signal.Subscriber.Internal + +-- | A signal which will send values of type @v@ on a scheduler of type @s@. +data Signal s v where + Signal :: Scheduler s => (Subscriber s v -> SchedulerIO s Disposable) -> Signal s v + +-- | Constructs a signal which sends its values to new subscribers synchronously. +signal + :: Scheduler s + => (Subscriber s v -> SchedulerIO s Disposable) -- ^ An action to run on each subscription. + -> Signal s v -- ^ The constructed signal. + +signal = Signal + +-- | Subscribes to a signal. +subscribe + :: Scheduler s + => Signal s v -- ^ The signal to subscribe to. + -> Subscriber s v -- ^ The subscriber to attach. + -> SchedulerIO s Disposable -- ^ A disposable which can be used to terminate the subscription. + +subscribe (Signal f) sub = do + d <- f sub + liftIO $ addSubscriptionDisposable sub d + return d + +-- | Returns a signal which never sends any events. +never :: Scheduler s => Signal s v +never = Signal $ const $ return EmptyDisposable + +-- | Returns a signal which immediately completes. +empty :: Scheduler s => Signal s v +empty = mempty + +-- | Creates a subscriber and subscribes to the signal. +(>>:) + :: Scheduler s + => Signal s v -- ^ The signal to subscribe to. + -> (Event v -> SchedulerIO s ()) -- ^ An action to run when each event is received. + -> SchedulerIO s Disposable -- ^ A disposable which can be used to terminate the subscription. + +(>>:) s f = do + sub <- liftIO $ subscriber f + subscribe s sub + +infixl 1 >>: + +instance Scheduler s => Monad (Signal s) where + return v = + Signal $ \sub -> do + send sub (NextEvent v) + send sub CompletedEvent + return EmptyDisposable + + s >>= f = + Signal $ \sub -> do + sc <- liftIO $ newIORef (1 :: Word32) + ds <- liftIO newDisposableSet + + let decSubscribers = do + rem <- liftIO $ atomicModifyIORef sc $ \n -> + let n' = n - 1 + in (n', n') + + when (rem == 0) $ send sub CompletedEvent + + onInner CompletedEvent = decSubscribers + onInner ev = send sub ev + + onOuter CompletedEvent = decSubscribers + onOuter (ErrorEvent e) = send sub $ ErrorEvent e + onOuter (NextEvent v) = do + liftIO $ atomicModifyIORef sc $ \n -> (n + 1, ()) + + d <- f v >>: onInner + liftIO $ ds `addDisposable` d + + d <- s >>: onOuter + liftIO $ ds `addDisposable` d + liftIO $ toDisposable ds + +instance Scheduler s => Functor (Signal s) where + fmap = liftM + +instance Scheduler s => Applicative (Signal s) where + pure = return + (<*>) = ap + +instance Scheduler s => Monoid (Signal s v) where + mempty = Signal $ \sub -> do + send sub CompletedEvent + return EmptyDisposable + + a `mappend` b = + Signal $ \sub -> do + ds <- liftIO newDisposableSet + + let onEvent CompletedEvent = do + d <- b `subscribe` sub + liftIO $ ds `addDisposable` d + + onEvent e = send sub e + + d <- a >>: onEvent + liftIO $ ds `addDisposable` d + liftIO $ toDisposable ds + +instance Scheduler s => MonadPlus (Signal s) where + mzero = mempty + a `mplus` b = + join $ Signal $ \sub -> do + send sub $ NextEvent a + send sub $ NextEvent b + send sub CompletedEvent + return EmptyDisposable + +-- Implementing mzip outside of an instance declaration so we can get some scoped type variables +-- (without needing the InstanceSigs extension). +szip :: forall a b s. Scheduler s => Signal s a -> Signal s b -> Signal s (a, b) +szip a b = + Signal $ \sub -> do + aVals <- liftIO $ atomically $ newTVar (Seq.empty :: Seq a) + aDone <- liftIO $ atomically $ newTVar False + + bVals <- liftIO $ atomically $ newTVar (Seq.empty :: Seq b) + bDone <- liftIO $ atomically $ newTVar False + + ds <- liftIO newDisposableSet + + let completed :: STM [Event (a, b)] + completed = do + ac <- readTVar aDone + al <- length <$> readTVar aVals + + bc <- readTVar bDone + bl <- length <$> readTVar bVals + + return $ if (ac && al == 0) || (bc && bl == 0) then [CompletedEvent] else [] + + onEvent' :: (TVar (Seq x), TVar Bool) -> (TVar (Seq y), TVar Bool) -> (Seq x -> Seq y -> Seq (a, b)) -> Event x -> STM [Event (a, b)] + onEvent' vt@(vals, _) ot@(otherVals, _) f (NextEvent v) = do + modifyTVar' vals (|> v) + + vs <- readTVar vals + os <- readTVar otherVals + + case viewl $ f vs os of + (t :< _) -> do + modifyTVar' vals $ drop 1 + modifyTVar' otherVals $ drop 1 + + (:) (NextEvent t) <$> completed + + _ -> return [] + + onEvent' (_, done) _ _ CompletedEvent = writeTVar done True >> completed + onEvent' _ _ _ (ErrorEvent e) = return [ErrorEvent e] + + onEvent :: (TVar (Seq x), TVar Bool) -> (TVar (Seq y), TVar Bool) -> (Seq x -> Seq y -> Seq (a, b)) -> Event x -> SchedulerIO s () + onEvent vt ot f ev = do + evl <- liftIO $ atomically $ onEvent' vt ot f ev + mapM_ (send sub) evl + + let at = (aVals, aDone) + bt = (bVals, bDone) + + ad <- a >>: onEvent at bt zip + liftIO $ ds `addDisposable` ad + + bd <- b >>: onEvent bt at (flip zip) + liftIO $ ds `addDisposable` bd + + liftIO $ toDisposable ds + +instance Scheduler s => MonadZip (Signal s) where + mzip = szip diff --git a/Signal/Channel.hs b/Signal/Channel.hs new file mode 100644 --- /dev/null +++ b/Signal/Channel.hs @@ -0,0 +1,99 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Channel ( Channel + , newChannel + , ChannelCapacity(..) + , newReplayChannel + , Signal + , Subscriber + , Scheduler + ) where + +import Control.Concurrent.STM +import Control.Monad hiding (mapM_) +import Control.Monad.IO.Class +import Data.Foldable +import Data.Sequence as Seq +import Disposable +import Prelude hiding (mapM_, length, drop) +import Scheduler +import Signal +import Signal.Subscriber +import Signal.Subscriber.Internal + +-- | A controllable signal, represented by a 'Subscriber' and 'Signal' pair. +-- +-- Values sent to the subscriber will automatically be broadcast to all of the signal's subscribers. +-- In effect, the subscriber is the write end, while the signal is the read end. +type Channel s v = (Subscriber s v, Signal s v) + +-- | Determines how many events a replay channel will save. +data ChannelCapacity = LimitedCapacity Int -- ^ The channel will only save the specified number of events. + | UnlimitedCapacity -- ^ The channel will save an unlimited number of events. + deriving (Eq, Show) + +-- | Creates a simple channel which broadcasts all values sent to it. +-- +-- Sending an 'ErrorEvent' or 'CompletedEvent' will terminate the channel. +newChannel :: Scheduler s => IO (Channel s v) +newChannel = newReplayChannel $ LimitedCapacity 0 + +-- | Like 'newChannel', but new subscriptions to the returned signal will receive all values +-- (up to the specified capacity) which have been sent thus far. +-- +-- Sending an 'ErrorEvent' or 'CompletedEvent' will terminate the channel. Any terminating event +-- will be replayed to future subscribers, assuming sufficient capacity. +newReplayChannel :: forall s v. Scheduler s => ChannelCapacity -> IO (Channel s v) +newReplayChannel cap = do + subs <- atomically $ newTVar Seq.empty + disposed <- atomically $ newTVar False + events <- atomically $ newTVar Seq.empty + + let addSubscriber :: Subscriber s v -> STM (Seq (Event v), Bool) + addSubscriber sub = do + d <- readTVar disposed + unless d $ modifyTVar' subs (|> sub) + + seq <- readTVar events + return (seq, d) + + s :: Signal s v + s = + signal $ \sub -> do + (events, d) <- liftIO $ atomically $ addSubscriber sub + + -- TODO: Allow these sends to be interrupted through disposal. + mapM_ (send sub) events + when d $ liftIO $ disposeSubscriber sub + + return EmptyDisposable + + limit :: Int -> Seq (Event v) -> Seq (Event v) + limit n seq + | n <= 0 = Seq.empty + | length seq > n = drop (length seq - n) seq + | otherwise = seq + + addEvent' :: ChannelCapacity -> Event v -> Seq (Event v) -> Seq (Event v) + addEvent' (LimitedCapacity c) ev seq = limit c $ seq |> ev + addEvent' UnlimitedCapacity ev seq = seq |> ev + + addEvent :: Event v -> STM (Seq (Subscriber s v)) + addEvent ev@(NextEvent _) = do + d <- readTVar disposed + unless d $ modifyTVar' events $ addEvent' cap ev + readTVar subs + + addEvent ev = do + d <- swapTVar disposed True + unless d $ modifyTVar' events $ addEvent' cap ev + swapTVar subs Seq.empty + + onEvent :: Event v -> SchedulerIO s () + onEvent ev = do + subs <- liftIO $ atomically $ addEvent ev + mapM_ (`send` ev) subs + + sub <- subscriber onEvent + return (sub, s) diff --git a/Signal/Command.hs b/Signal/Command.hs new file mode 100644 --- /dev/null +++ b/Signal/Command.hs @@ -0,0 +1,201 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Command ( Command + , CommandPolicy(..) + , newCommand + , canExecute + , executing + , execute + , values + , onExecute + , errors + , Channel + , Signal + , Scheduler + , SchedulerIO + , BackgroundScheduler + , MainScheduler + ) where + +import Control.Concurrent.MVar +import Control.Exception hiding (finally) +import Control.Monad +import Control.Monad.IO.Class +import Data.Monoid +import Data.Word +import Prelude hiding (map) +import Scheduler +import Scheduler.Main +import Signal +import Signal.Channel +import Signal.Connection +import Signal.Operators +import Signal.Scheduled +import Signal.Subscriber +import System.Mem.Weak + +-- | Determines a command's behavior. +data CommandPolicy = ExecuteSerially -- ^ The command can only be executed once at a time. + -- Attempts to 'execute' while the command is already running will fail. + | ExecuteConcurrently -- ^ The command can be executed concurrently any number of times. + deriving (Eq, Show) + +-- | A signal triggered in response to some action, typically UI-related. +data Command v = Command { + policy :: CommandPolicy, + valuesChannel :: Channel MainScheduler v, + errorsChannel :: Channel MainScheduler IOException, + itemsInFlight :: MVar Word32, + executingChannel :: Channel MainScheduler Bool, + canExecuteChannel :: Channel MainScheduler Bool +} + +-- | Creates a command. +newCommand + :: CommandPolicy -- ^ Determines how the command behaves when asked to 'execute' multiple times simultaneously. + -> Signal MainScheduler Bool -- ^ A signal which sends whether the command should be enabled. + -- 'canExecute' will send 'True' before this signal sends its first value. + -> SchedulerIO MainScheduler (Command v) + +newCommand p ces = do + vChan <- liftIO newChannel + errChan <- liftIO newChannel + exChan <- liftIO $ newReplayChannel $ LimitedCapacity 1 + ceChan <- liftIO $ newReplayChannel $ LimitedCapacity 1 + items <- liftIO $ newMVar 0 + + let canExecute :: Bool -> Bool -> CommandPolicy -> Bool + canExecute ce executing ExecuteSerially = ce && not executing + canExecute ce _ ExecuteConcurrently = ce + + onEvent :: Event (Bool, Bool) -> SchedulerIO MainScheduler () + onEvent (NextEvent (ce, ex)) = send (fst ceChan) $ NextEvent $ canExecute ce ex p + onEvent _ = return () + + complete :: IO () + complete = do + sch <- getMainScheduler + void $ schedule sch $ do + send (fst vChan) CompletedEvent + send (fst errChan) CompletedEvent + send (fst exChan) CompletedEvent + send (fst ceChan) CompletedEvent + + -- Start with True for 'canExecute'. + combine (return True `mappend` ces) (snd exChan) >>: onEvent + + let command = Command { + policy = p, + valuesChannel = vChan, + errorsChannel = errChan, + itemsInFlight = items, + executingChannel = exChan, + canExecuteChannel = ceChan + } + + -- Set the starting value for 'executing'. + setExecuting command False + + liftIO $ addFinalizer command complete + return command + +-- | Sends whether this command is currently executing. +-- +-- This signal will always send at least one value immediately upon subscription. +executing :: Command v -> Signal MainScheduler Bool +executing = snd . executingChannel + +-- | A signal of errors received from all signals created by 'onExecute'. +errors :: Command v -> Signal MainScheduler IOException +errors = snd . errorsChannel + +-- | A signal of the values passed to 'execute'. +values :: Command v -> Signal MainScheduler v +values = snd . valuesChannel + +-- | Sends whether this command is able to execute. +-- +-- This signal will always send at least one value immediately upon subscription. +canExecute :: Command v -> Signal MainScheduler Bool +canExecute = snd . canExecuteChannel + +-- | Sends a new value for 'executing'. +setExecuting :: Command v -> Bool -> SchedulerIO MainScheduler () +setExecuting c b = send (fst $ executingChannel c) $ NextEvent b + +-- | Attempts to execute a command. +execute + :: Command v -- ^ The command to execute. + -> v -- ^ A value to send to the command's subscribers. + -> SchedulerIO MainScheduler Bool -- ^ Whether execution succeeded. This will be 'False' if the command is disabled. + +execute c v = do + items <- liftIO $ takeMVar $ itemsInFlight c + + let execute' :: SchedulerIO MainScheduler Bool + execute' = do + liftIO $ putMVar (itemsInFlight c) $ items + 1 + + setExecuting c True + fst (valuesChannel c) `send` NextEvent v + + items <- liftIO $ modifyMVar (itemsInFlight c) $ \items -> + return (items - 1, items - 1) + + when (items == 0) $ setExecuting c False + return True + + ce <- liftIO $ first $ canExecute c + + case ce of + (NextEvent True) -> execute' + _ -> do + liftIO $ putMVar (itemsInFlight c) items + return False + +-- | Creates a signal whenever the command executes, then subscribes to it. +onExecute + :: Command v -- ^ The command to attach behavior to. + -> (v -> Signal BackgroundScheduler ()) -- ^ A function which maps from the command's values to a signal of side-effecting work. + -> SchedulerIO MainScheduler (Signal MainScheduler (Signal BackgroundScheduler ())) -- ^ A signal of the created signals. + +onExecute c f = + let incItems :: SchedulerIO MainScheduler () + incItems = do + liftIO $ modifyMVar_ (itemsInFlight c) $ \items -> return $ items + 1 + setExecuting c True + + decItems :: SchedulerIO BackgroundScheduler () + decItems = do + mainSch <- liftIO getMainScheduler + void $ liftIO $ schedule mainSch $ do + items <- liftIO $ modifyMVar (itemsInFlight c) $ \items -> return (items - 1, items - 1) + when (items == 0) $ setExecuting c False + + sendError :: IOException -> IO () + sendError ex = do + sch <- getMainScheduler + void $ schedule sch $ send (fst $ errorsChannel c) $ NextEvent ex + + coldSignals :: Signal MainScheduler (Signal BackgroundScheduler ()) + coldSignals = + values c + `doNext` const incItems + `map` \v -> + f v + `doError` (liftIO . sendError) + `finally` decItems + + signals :: Signal MainScheduler (Signal BackgroundScheduler ()) + signals = + coldSignals >>= \sig -> + signal $ \sub -> do + sch <- liftIO newScheduler + chan <- liftIO $ newReplayChannel UnlimitedCapacity + conn <- liftIO $ multicast sig chan + + d <- liftIO $ schedule sch $ void $ connect conn + send sub $ NextEvent $ multicastedSignal conn + return d + in replayLast signals diff --git a/Signal/Connection.hs b/Signal/Connection.hs new file mode 100644 --- /dev/null +++ b/Signal/Connection.hs @@ -0,0 +1,103 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Connection ( Connection + , multicast + , publish + , connect + , multicastedSignal + , replay + , replayLast + , Channel + , Signal + , Scheduler + , SchedulerIO + , Disposable + ) where + +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.IO.Class +import Disposable +import Scheduler +import Signal +import Signal.Channel + +-- | Multicasts a signal to many subscribers, without triggering any side effects more than once. +data Connection s v = Connection { + baseSignal :: Signal s v, + disposable :: TVar Disposable, + channel :: Channel s v, + hasConnected :: TMVar Bool +} + +-- | Creates a connection that will subscribe to the given base signal, +-- and forward all events onto the given channel. +multicast :: Scheduler s => Signal s v -> Channel s v -> IO (Connection s v) +multicast sig chan = do + d <- atomically $ newTVar EmptyDisposable + hc <- atomically $ newTMVar False + + return Connection { + baseSignal = sig, + disposable = d, + channel = chan, + hasConnected = hc + } + +-- | Multicasts to a simple channel. +publish :: Scheduler s => Signal s v -> IO (Connection s v) +publish sig = newChannel >>= multicast sig + +-- | Multicasts to a replay channel of the specified capacity, then connects immediately. +replayCapacity :: Scheduler s => ChannelCapacity -> Signal s v -> SchedulerIO s (Signal s v) +replayCapacity c sig = do + chan <- liftIO $ newReplayChannel c + conn <- liftIO $ multicast sig chan + + connect conn + return $ multicastedSignal conn + +-- | Multicasts to a replay channel of unlimited capacity, then connects immediately. +replay :: Scheduler s => Signal s v -> SchedulerIO s (Signal s v) +replay = replayCapacity UnlimitedCapacity + +-- | Multicasts to a replay channel of capacity 1, then connects immediately. +replayLast :: Scheduler s => Signal s v -> SchedulerIO s (Signal s v) +replayLast = replayCapacity $ LimitedCapacity 1 + +-- | Returns the multicasted signal of a connection. +-- +-- No events will be sent on the resulting signal until 'connect' is invoked. +multicastedSignal :: Connection s v -> Signal s v +multicastedSignal conn = snd $ channel conn + +-- | Activates a connection by subscribing to its underlying signal. +-- Calling this function multiple times just returns the existing disposable. +connect :: forall s v. Scheduler s => Connection s v -> SchedulerIO s Disposable +connect conn = + let connect' :: SchedulerIO s Disposable + connect' = do + d <- baseSignal conn `subscribe` fst (channel conn) + liftIO $ atomically $ setDisposable d + return d + + setDisposable :: Disposable -> STM () + setDisposable d = do + putTMVar (hasConnected conn) True + writeTVar (disposable conn) d + + shouldConnect :: STM (Bool, Disposable) + shouldConnect = do + hc <- takeTMVar $ hasConnected conn + d <- if hc + then putTMVar (hasConnected conn) hc >> return EmptyDisposable + -- If we're going to connect, leave the MVar empty until we've filled in the disposable. + else readTVar $ disposable conn + + return (not hc, d) + in do + (shouldConnect, d) <- liftIO $ atomically shouldConnect + if shouldConnect + then connect' + else return d diff --git a/Signal/Event.hs b/Signal/Event.hs new file mode 100644 --- /dev/null +++ b/Signal/Event.hs @@ -0,0 +1,24 @@ +{-# LANGUAGE Safe #-} + +module Signal.Event ( Event(..) + ) where + +import Control.Exception + +-- | Represents an event that a signal might send. +-- +-- Signals may send any number of 'NextEvent's, followed by one 'ErrorEvent' /or/ 'CompletedEvent'. +data Event v = NextEvent v -- ^ A value @v@ in the monad. + | ErrorEvent IOException -- ^ Sent when an error or exception occurs in the signal. Outside of the monad. + | CompletedEvent -- ^ Sent when the signal completes successfully. Outside of the monad. + +instance Eq v => Eq (Event v) where + (NextEvent v) == (NextEvent v') = v == v' + (ErrorEvent e) == (ErrorEvent e') = e == e' + CompletedEvent == CompletedEvent = True + _ == _ = False + +instance Show v => Show (Event v) where + show (NextEvent v) = "NextEvent " ++ show v + show (ErrorEvent e) = "ErrorEvent " ++ show e + show CompletedEvent = "CompletedEvent" diff --git a/Signal/Operators.hs b/Signal/Operators.hs new file mode 100644 --- /dev/null +++ b/Signal/Operators.hs @@ -0,0 +1,225 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Operators ( doEvent + , doNext + , doError + , doCompleted + , finally + , materialize + , dematerialize + , fromFoldable + , map + , filter + , take + , drop + , switch + , combine + , never + , Signal.empty + , Signal + ) where + +import Control.Concurrent.STM +import Control.Exception hiding (finally) +import Control.Monad +import Control.Monad.IO.Class +import Data.Foldable +import Data.IORef +import Data.Monoid +import Prelude hiding (filter, take, drop, map) +import Disposable +import Scheduler +import Signal +import Signal.Subscriber + +-- | Turns any 'Foldable' into a signal. +fromFoldable :: (Foldable t, Scheduler s) => t v -> Signal s v +fromFoldable = foldMap return + +-- | Brings every signal event into the monad, as a 'NextEvent' containing the event itself. +materialize :: Scheduler s => Signal s v -> Signal s (Event v) +materialize s = + signal $ \sub -> + let onEvent CompletedEvent = send sub (NextEvent CompletedEvent) >> send sub CompletedEvent + onEvent ev = send sub $ NextEvent ev + in s >>: onEvent + +-- | The inverse of 'materialize'. +dematerialize :: Scheduler s => Signal s (Event v) -> Signal s v +dematerialize s = + signal $ \sub -> + let onEvent (NextEvent ev) = send sub ev + onEvent _ = return () + in s >>: onEvent + +-- | Filters the values of a signal according to a predicate. +filter :: Scheduler s => Signal s v -> (v -> Bool) -> Signal s v +filter s f = + let f' x = if f x then return x else mempty + in s >>= f' + +-- | Runs a side-effecting action whenever the signal sends an event. +doEvent :: Scheduler s => Signal s v -> (Event v -> SchedulerIO s ()) -> Signal s v +doEvent s f = + signal $ \sub -> + let onEvent e = f e >> send sub e + in s >>: onEvent + +-- | Runs a side-effecting action whenever the signal sends a value. +doNext :: Scheduler s => Signal s v -> (v -> SchedulerIO s ()) -> Signal s v +doNext s f = + let f' (NextEvent x) = f x + f' _ = return () + in doEvent s f' + +-- | Runs a side-effecting action whenever the signal sends an error. +doError :: Scheduler s => Signal s v -> (IOException -> SchedulerIO s ()) -> Signal s v +doError s f = + let f' (ErrorEvent ex) = f ex + f' _ = return () + in doEvent s f' + +-- | Runs a side-effecting action when the signal completes. +doCompleted :: Scheduler s => Signal s v -> SchedulerIO s () -> Signal s v +doCompleted s f = + let f' CompletedEvent = f + f' _ = return () + in doEvent s f' + +-- | Runs a side-effecting action when the signal completes or errors. +finally :: Scheduler s => Signal s v -> SchedulerIO s () -> Signal s v +finally s f = + let f' (NextEvent _) = return () + f' _ = f + in doEvent s f' + +-- | Returns a signal of the first @n@ elements. +take :: (Integral n, Scheduler s) => Signal s v -> n -> Signal s v +take s n = + signal $ \sub -> do + remRef <- liftIO $ newIORef n + + let onEvent ev@(NextEvent _) = do + old <- liftIO $ atomicModifyIORef remRef $ \rem -> + if rem == 0 then (0, 0) else (rem - 1, rem) + + case old of + 0 -> return () + 1 -> send sub ev >> send sub CompletedEvent + _ -> send sub ev + + onEvent ev = do + b <- liftIO $ atomicModifyIORef remRef $ \rem -> (0, rem /= 0) + when b $ send sub ev + + s >>: onEvent + +-- | Returns a signal without the first @n@ elements. +drop :: (Integral n, Scheduler s) => Signal s v -> n -> Signal s v +drop s n = + signal $ \sub -> do + remRef <- liftIO $ newIORef n + + let onEvent ev@(NextEvent _) = do + old <- liftIO $ atomicModifyIORef remRef $ \rem -> + if rem == 0 then (0, 0) else (rem - 1, rem) + + when (old == 0) $ send sub ev + + onEvent ev = send sub ev + + s >>: onEvent + +-- | Returns a signal of mapped values. +map :: Scheduler s => Signal s v -> (v -> w) -> Signal s w +map = flip fmap + +-- | Returns a signal that sends the values from the most recently sent signal. +switch :: forall s v. Scheduler s => Signal s (Signal s v) -> Signal s v +switch s = + signal $ \sub -> do + ds <- liftIO newDisposableSet + actives <- liftIO $ newIORef (True, False) -- Outer, Inner + + currD <- liftIO $ newIORef EmptyDisposable + + let disposeCurrD = do + d <- readIORef currD + dispose d + + d <- liftIO $ newDisposable disposeCurrD + liftIO $ ds `addDisposable` d + + let modifyActives :: (Maybe Bool, Maybe Bool) -> SchedulerIO s (Bool, Bool) + modifyActives (Nothing, Just ni) = liftIO $ atomicModifyIORef actives $ \(outer, _) -> ((outer, ni), (outer, ni)) + modifyActives (Just no, Nothing) = liftIO $ atomicModifyIORef actives $ \(_, inner) -> ((no, inner), (no, inner)) + + completeIfDone :: (Bool, Bool) -> SchedulerIO s () + completeIfDone (False, False) = send sub CompletedEvent + completeIfDone _ = return () + + onEvent :: Event (Signal s v) -> SchedulerIO s () + onEvent (NextEvent s') = do + let onInnerEvent :: Event v -> SchedulerIO s () + onInnerEvent CompletedEvent = modifyActives (Nothing, Just False) >>= completeIfDone + onInnerEvent ev = send sub ev + + modifyActives (Nothing, Just True) + nd <- s' >>: onInnerEvent + + oldD <- liftIO $ atomicModifyIORef currD (\oldD -> (nd, oldD)) + liftIO $ dispose oldD + + onEvent (ErrorEvent e) = send sub $ ErrorEvent e + onEvent CompletedEvent = modifyActives (Just False, Nothing) >>= completeIfDone + + d <- s >>: onEvent + liftIO $ ds `addDisposable` d + liftIO $ toDisposable ds + +-- | Combines the latest values sent by both signals. +combine :: forall a b s. Scheduler s => Signal s a -> Signal s b -> Signal s (a, b) +combine a b = + signal $ \sub -> do + aVal <- liftIO $ atomically $ newTVar (Nothing :: Maybe a) + aDone <- liftIO $ atomically $ newTVar False + + bVal <- liftIO $ atomically $ newTVar (Nothing :: Maybe b) + bDone <- liftIO $ atomically $ newTVar False + + ds <- liftIO newDisposableSet + + let completed :: STM (Maybe (Event (a, b))) + completed = do + ac <- readTVar aDone + bc <- readTVar bDone + return $ if ac && bc then Just CompletedEvent else Nothing + + onEvent' :: TVar (Maybe x) -> TVar Bool -> Event x -> STM (Maybe (Event (a, b))) + onEvent' _ _ (ErrorEvent e) = return $ Just $ ErrorEvent e + onEvent' _ done CompletedEvent = writeTVar done True >> completed + onEvent' val _ (NextEvent x) = do + writeTVar val $ Just x + + av <- readTVar aVal + bv <- readTVar bVal + + case (av, bv) of + (Just ax, Just bx) -> return $ Just $ NextEvent (ax, bx) + _ -> return Nothing + + onEvent :: TVar (Maybe x) -> TVar Bool -> Event x -> SchedulerIO s () + onEvent val done ev = do + m <- liftIO $ atomically $ onEvent' val done ev + case m of + (Just ev) -> send sub ev + Nothing -> return () + + ad <- a >>: onEvent aVal aDone + liftIO $ ds `addDisposable` ad + + bd <- b >>: onEvent bVal bDone + liftIO $ ds `addDisposable` bd + + liftIO $ toDisposable ds diff --git a/Signal/Scheduled.hs b/Signal/Scheduled.hs new file mode 100644 --- /dev/null +++ b/Signal/Scheduled.hs @@ -0,0 +1,84 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Scheduled ( start + , subscribeOn + , deliverOn + , first + , Scheduler + , SchedulerIO + , Signal + ) where + +import Control.Concurrent.MVar +import Control.Monad +import Control.Monad.IO.Class +import Disposable +import Prelude hiding (take) +import Scheduler +import Scheduler.Internal +import Signal +import Signal.Channel +import Signal.Operators +import Signal.Subscriber + +-- | Starts a signal which executes @action@ on @s@. +start :: Scheduler s => s -> (Subscriber s v -> SchedulerIO s ()) -> IO (Signal s v) +start s action = do + (sub, sig) <- newReplayChannel UnlimitedCapacity + schedule s $ action sub + return sig + +-- | Returns a signal which subscribes to @sig@ on scheduler @sch@. +subscribeOn :: forall s t v. (Scheduler s, Scheduler t) => Signal s v -> t -> Signal t v +subscribeOn sig sch = + let onSubscribe :: Subscriber t v -> SchedulerIO t Disposable + onSubscribe sub = do + ds <- liftIO newDisposableSet + + let forward :: Event v -> SchedulerIO s () + forward ev = SchedulerIO $ unsafeRunSchedulerIO $ sub `send` ev + + subscribe :: SchedulerIO t () + subscribe = do + d <- SchedulerIO $ unsafeRunSchedulerIO $ sig >>: forward + liftIO $ ds `addDisposable` d + + schD <- liftIO $ sch `schedule` subscribe + + liftIO $ ds `addDisposable` schD + liftIO $ toDisposable ds + in signal onSubscribe + +-- | Returns a signal which delivers the events of @sig@ on scheduler @sch@. +deliverOn :: forall s t v. (Scheduler s, Scheduler t) => Signal s v -> t -> Signal t v +deliverOn sig sch = + let onSubscribe :: Subscriber t v -> SchedulerIO t Disposable + onSubscribe sub = do + -- Although we could hold onto any disposable returned from scheduling, + -- the complexity of managing all of them probably isn't worth the + -- slightly faster cancellation. + let deliver :: t -> Event v -> SchedulerIO s Disposable + deliver sch ev = + let sio = SchedulerIO $ unsafeRunSchedulerIO $ sub `send` ev + in liftIO $ sch `schedule` sio + + forward :: Event v -> SchedulerIO s () + forward ev = void $ deliver sch ev + + SchedulerIO $ unsafeRunSchedulerIO $ sig >>: forward + in signal onSubscribe + +-- | Synchronously waits for the signal to send an event. +first :: forall s v. Scheduler s => Signal s v -> IO (Event v) +first s = do + var <- newEmptyMVar + + let onEvent :: Event v -> SchedulerIO s () + onEvent ev = void $ liftIO $ tryPutMVar var ev + + subscribe :: SchedulerIO s Disposable + subscribe = take s 1 >>: onEvent + + unsafeRunSchedulerIO subscribe + takeMVar var diff --git a/Signal/Subscriber.hs b/Signal/Subscriber.hs new file mode 100644 --- /dev/null +++ b/Signal/Subscriber.hs @@ -0,0 +1,92 @@ +{-# LANGUAGE Safe #-} +{-# LANGUAGE ScopedTypeVariables #-} + +module Signal.Subscriber ( Subscriber + , subscriber + , send + , Event(..) + ) where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Monad +import Control.Monad.IO.Class +import Disposable +import Scheduler +import Signal.Event +import Signal.Subscriber.Internal + +-- | Constructs a subscriber. +subscriber + :: Scheduler s + => (Event v -> SchedulerIO s ()) -- ^ An action to run when each event is received. + -> IO (Subscriber s v) -- ^ The constructed subscriber. + +subscriber f = do + b <- atomically $ newTVar False + d <- newDisposable $ atomically $ writeTVar b True + + ds <- newDisposableSet + addDisposable ds d + + tid <- myThreadId + lt <- atomically $ newTVar tid + tlc <- atomically $ newTVar 0 + + return Subscriber { + onEvent = f, + disposables = ds, + lockedThread = lt, + threadLockCounter = tlc, + disposed = b + } + +-- | Acquires a subscriber for the specified thread. +-- +-- This is used to ensure that a subscriber never receives multiple events concurrently. +acquireSubscriber :: Subscriber s v -> ThreadId -> STM Bool +acquireSubscriber sub tid = do + d <- readTVar (disposed sub) + if d + then return False + else do + -- TODO: Skip all this synchronization for singleton scheduler types. + tlc <- readTVar (threadLockCounter sub) + lt <- readTVar (lockedThread sub) + when (tlc > 0 && lt /= tid) retry + + writeTVar (lockedThread sub) tid + writeTVar (threadLockCounter sub) $ tlc + 1 + return True + +-- | Releases a subscriber from the specified thread's ownership. +-- +-- This must match a previous call to 'acquireSubscriber'. +releaseSubscriber :: Subscriber s v -> ThreadId -> STM () +releaseSubscriber sub tid = do + -- TODO: Skip all this synchronization for singleton scheduler types. + always $ fmap (== tid) $ readTVar (lockedThread sub) + + tlc <- readTVar (threadLockCounter sub) + always $ return $ tlc > 0 + + writeTVar (threadLockCounter sub) $ tlc - 1 + +-- | Synchronously sends an event to a subscriber. +send :: forall s v. Scheduler s => Subscriber s v -> Event v -> SchedulerIO s () +send s ev = + let sendAndDispose :: Event v -> SchedulerIO s () + sendAndDispose ev = do + liftIO $ disposeSubscriber s + onEvent s ev + + send' :: Event v -> SchedulerIO s () + send' ev@(NextEvent _) = onEvent s ev + send' ev = do + wasDisposed <- liftIO $ atomically $ swapTVar (disposed s) True + unless wasDisposed $ sendAndDispose ev + in do + tid <- liftIO myThreadId + b <- liftIO $ atomically $ acquireSubscriber s tid + + when b $ send' ev >> liftIO (atomically (releaseSubscriber s tid)) diff --git a/Signal/Subscriber/Internal.hs b/Signal/Subscriber/Internal.hs new file mode 100644 --- /dev/null +++ b/Signal/Subscriber/Internal.hs @@ -0,0 +1,35 @@ +{-# LANGUAGE Safe #-} + +module Signal.Subscriber.Internal ( Subscriber(..) + , addSubscriptionDisposable + , disposeSubscriber + ) where + +import Control.Applicative +import Control.Concurrent +import Control.Concurrent.STM +import Data.Word +import Disposable +import Scheduler +import Signal.Event + +-- | Receives events from a signal with values of type @v@ and running in a scheduler of type @s@. +-- +-- Note that @s@ refers to the scheduler that events must be sent on. Events are always sent +-- synchronously, regardless of @s@. +data Subscriber s v = Subscriber { + onEvent :: Event v -> SchedulerIO s (), + disposables :: DisposableSet, + lockedThread :: TVar ThreadId, + threadLockCounter :: TVar Word32, + disposed :: TVar Bool +} + +-- | Adds a disposable representing a subscription to the subscriber. +-- If the subscriber is later sent completed or error, the disposable will be disposed. +addSubscriptionDisposable :: Subscriber s v -> Disposable -> IO () +addSubscriptionDisposable = addDisposable . disposables + +-- | Disposes the subscriber, preventing it from receiving any new events. +disposeSubscriber :: Subscriber s v -> IO () +disposeSubscriber s = toDisposable (disposables s) >>= dispose