RxHaskell 0.1 → 0.2

raw patch · 7 files changed

+69/−58 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• RxHaskell.cabal +1/−1
• Scheduler.hs +1/−0
• Scheduler/Internal.hs +16/−17
• Scheduler/Unsafe.hs +19/−2
• Signal/Command.hs +1/−1
• Signal/Scheduled.hs +28/−35
• Signal/Subscriber.hs +3/−2

@@ -1,5 +1,5 @@ name:           RxHaskell-version:        0.1+version:        0.2 stability:      alpha synopsis:       Reactive Extensions for Haskell description:    An implementation of functional reactive programming based on Microsoft's Reactive Extensions for .NET: <http://msdn.microsoft.com/en-us/library/hh242985(v=VS.103).aspx>.

@@ -2,6 +2,7 @@  module Scheduler ( Scheduler(schedule)                  , SchedulerIO+                 , getCurrentScheduler                  , BackgroundScheduler                  , newScheduler                  ) where

@@ -2,7 +2,7 @@ {-# LANGUAGE Safe #-}  module Scheduler.Internal ( SchedulerIO(..)-                          , unsafeRunSchedulerIO+                          , getCurrentScheduler                           , Scheduler(..)                           , BackgroundScheduler(..)                           , ScheduledAction@@ -21,27 +21,26 @@  -- | 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+    SchedulerIO :: Scheduler s => (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+-- | Returns the scheduler that the calling code is executing on.+getCurrentScheduler :: Scheduler s => SchedulerIO s s+getCurrentScheduler = SchedulerIO return  instance Functor (SchedulerIO s) where-    fmap f (SchedulerIO action) = SchedulerIO $ fmap f action+    fmap f (SchedulerIO mf) = SchedulerIO $ fmap f . mf  instance Scheduler s => Monad (SchedulerIO s) where-    return = SchedulerIO . return-    (SchedulerIO action) >>= f =-        SchedulerIO $ do-            v <- action-            unsafeRunSchedulerIO $ f v+    return v = SchedulerIO $ \_ -> return v+    (SchedulerIO mf) >>= f =+        SchedulerIO $ \sch -> do+            v <- mf sch +            let unwrap sch (SchedulerIO mf') = mf' sch+            unwrap sch $ f v+ instance Scheduler s => MonadIO (SchedulerIO s) where-    liftIO = SchedulerIO+    liftIO action = SchedulerIO $ \_ -> action  instance Scheduler s => Applicative (SchedulerIO s) where     pure = return@@ -93,9 +92,9 @@  -- | Executes the given action, then re-enters 'schedulerMain'. executeScheduledAction :: Scheduler s => s -> ScheduledAction s -> IO ()-executeScheduledAction s (ScheduledAction ref (SchedulerIO action)) = do+executeScheduledAction s (ScheduledAction ref (SchedulerIO mf)) = do     d <- readIORef ref-    unless d action+    unless d $ mf s      yield     schedulerMain s

@@ -1,12 +1,29 @@+{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE Unsafe #-}  module Scheduler.Unsafe (                         ) where +import Scheduler (newScheduler) import Scheduler.Internal+import Scheduler.Main import System.IO.Unsafe (unsafePerformIO) +-- | 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 => s -> SchedulerIO s a -> IO a+unsafeRunSchedulerIO sch (SchedulerIO mf) = mf sch+ -- | 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+instance Show v => Show (SchedulerIO MainScheduler v) where+    show action = show $ unsafePerformIO $ do+        sch <- getMainScheduler+        unsafeRunSchedulerIO sch action++instance Show v => Show (SchedulerIO BackgroundScheduler v) where+    show action = show $ unsafePerformIO $ do+        sch <- newScheduler+        unsafeRunSchedulerIO sch action

@@ -146,7 +146,7 @@             when (items == 0) $ setExecuting c False             return True -    ce <- liftIO $ first $ canExecute c+    ce <- first $ canExecute c      case ce of         (NextEvent True) -> execute'

@@ -16,7 +16,6 @@ import Disposable import Prelude hiding (take) import Scheduler-import Scheduler.Internal import Signal import Signal.Channel import Signal.Operators@@ -30,55 +29,49 @@     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 :: forall s v. Scheduler s => Signal s v -> s -> Signal s v subscribeOn sig sch =-    let onSubscribe :: Subscriber t v -> SchedulerIO t Disposable+    let onSubscribe :: Subscriber s v -> SchedulerIO s 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+            schD <- liftIO $ schedule sch $ do+                d <- subscribe sig sub+                liftIO $ ds `addDisposable` d              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 =+-- | Returns a signal which subscribes to @sig@ on scheduler @schA@ and delivers its events onto scheduler @schB@.+deliverOn :: forall s t v. (Scheduler s, Scheduler t) => Signal s v -> s -> t -> Signal t v+deliverOn sig schA schB =     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+        onSubscribe sub =+            let forward :: Event v -> SchedulerIO s ()+                forward ev =+                    -- 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.+                    void $ liftIO $ schedule schB $ send sub ev+            in do+                ds <- liftIO newDisposableSet -                forward :: Event v -> SchedulerIO s ()-                forward ev = void $ deliver sch ev+                schD <- liftIO $ schedule schA $ do+                    d <- sig >>: forward+                    liftIO $ ds `addDisposable` d -            SchedulerIO $ unsafeRunSchedulerIO $ sig >>: forward+                liftIO $ ds `addDisposable` schD+                liftIO $ toDisposable ds     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+-- | Subscribes to @sig@ and synchronously waits for an event.+first :: forall s v. Scheduler s => Signal s v -> SchedulerIO s (Event v)+first sig = do+    var <- liftIO 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+    take sig 1 >>: onEvent+    liftIO $ takeMVar var

@@ -65,10 +65,11 @@ 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)+    ltid <- readTVar $ lockedThread sub+    unless (ltid == tid) $ throwSTM $ userError $ "Locked thread " ++ show ltid ++ " does not match current thread " ++ show tid      tlc <- readTVar (threadLockCounter sub)-    always $ return $ tlc > 0+    unless (tlc > 0) $ throwSTM $ userError "Thread lock count is not greater than zero"      writeTVar (threadLockCounter sub) $ tlc - 1