diff --git a/examples/poodle/Engine.hs b/examples/poodle/Engine.hs --- a/examples/poodle/Engine.hs +++ b/examples/poodle/Engine.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable, EmptyDataDecls #-} +{-# LANGUAGE ScopedTypeVariables, EmptyDataDecls #-} module Engine where import FRP.Sodium @@ -6,7 +6,6 @@ import Control.Applicative import Control.Monad import Data.List -import Data.Typeable import Graphics.Rendering.OpenGL as GL hiding (Triangle, Rect, translate) import qualified Graphics.Rendering.OpenGL as GL import qualified Graphics.UI.GLUT as GLUT hiding (Rect, translate) @@ -62,22 +61,20 @@ (fromIntegral sec) + (fromIntegral pico) / 1000000000000 -data M deriving Typeable - -- | Game, which takes mouse event and time as input, and a list of sprites to draw -- as output. Time is updated once per animation frame. -type Game p = Event p MouseEvent -> Behaviour p Double -> Reactive p (Behaviour p [Sprite]) +type Game = Event MouseEvent -> Behaviour Double -> Reactive (Behaviour [Sprite]) -runGame :: String -> Game M -> IO () +runGame :: String -> Game -> IO () runGame title game = do - (eMouse, pushMouse) <- newEvent - (eTime, pushTime) <- newEvent + (eMouse, pushMouse) <- sync newEvent + (eTime, pushTime) <- sync newEvent spritesRef <- newIORef [] - _ <- synchronously $ do + _ <- sync $ do time <- hold 0 eTime sprites <- game eMouse time - listenValueIO sprites (writeIORef spritesRef) + listenValue sprites (writeIORef spritesRef) _ <- GLUT.getArgsAndInitialize GLUT.initialDisplayMode $= [GLUT.DoubleBuffered] @@ -97,15 +94,15 @@ GLUT.displayCallback $= display texturesRef t0 pushTime spritesRef let motion (GLUT.Position x y) = do pt <- toScreen x y - synchronously $ pushMouse (MouseMove pt) + sync $ pushMouse (MouseMove pt) GLUT.motionCallback $= Just motion GLUT.passiveMotionCallback $= Just motion GLUT.keyboardMouseCallback $= Just (\key keyState mods pos -> do case (key, keyState, pos) of (GLUT.MouseButton GLUT.LeftButton, GLUT.Down, GLUT.Position x y) -> - synchronously . pushMouse . MouseDown =<< toScreen x y + sync . pushMouse . MouseDown =<< toScreen x y (GLUT.MouseButton GLUT.LeftButton, GLUT.Up, GLUT.Position x y) -> - synchronously . pushMouse . MouseUp =<< toScreen x y + sync . pushMouse . MouseUp =<< toScreen x y _ -> return () ) GLUT.addTimerCallback (1000 `div` frameRate) $ repaint @@ -128,13 +125,13 @@ display :: IORef (Map String (TextureImage, TextureObject)) -> Double - -> (Double -> Reactive M ()) + -> (Double -> Reactive ()) -> IORef [Sprite] -> IO () display texturesRef t0 pushTime spritesRef = do t <- subtract t0 <$> getTime - synchronously $ pushTime t + sync $ pushTime t sprites <- readIORef spritesRef diff --git a/examples/poodle/poodle.hs b/examples/poodle/poodle.hs --- a/examples/poodle/poodle.hs +++ b/examples/poodle/poodle.hs @@ -8,7 +8,6 @@ import Control.Applicative import Control.Monad.Trans import Data.Maybe -import Data.Typeable import Engine import System.Random @@ -16,12 +15,11 @@ poodleSprite pt = ((pt,(120,120)), "poodle.png") -- | Active poodle logic (which could be made much more interesting). -poodle :: Typeable p => - PoodleID +poodle :: PoodleID -> Point - -> Event p MouseEvent - -> Behaviour p Double - -> Reactive p (Behaviour p (PoodleID, Sprite)) + -> Event MouseEvent + -> Behaviour Double + -> Reactive (Behaviour (PoodleID, Sprite)) poodle iD pos@(x0,y0) eMouse time = do t0 <- sample time let dt = subtract t0 <$> time @@ -32,29 +30,29 @@ return sprite -- | Peel a new item off the list each time the event fires. -peelList :: Typeable p => Event p x -> [a] -> Reactive p (Behaviour p a) +peelList :: Event x -> [a] -> Reactive (Behaviour a) peelList ev xs0 = hold (head xs0) =<< collectE (\_ (x:xs) -> (x, xs)) (tail xs0) ev -- | Generate events at random intervals. -randomTimes :: Typeable p => StdGen -> Behaviour p Double -> Reactive p (Event p Double) +randomTimes :: StdGen -> Behaviour Double -> Reactive (Event Double) randomTimes rng time = do -- Infinite list of random intervals from 0.25 to 1.2 seconds. let intervals = randomRs (0.25, 1.2) rng rec tLast <- hold 0 eAppear interval <- peelList eAppear intervals - let eTime = valueEvent time - eAppear = justE $ attachWith (\t (tLast, interval) -> + let eTime = values time + eAppear = filterJust $ snapshotWith (\t (tLast, interval) -> if t >= tLast + interval then Just t else Nothing ) eTime ((,) <$> tLast <*> interval) return eAppear newtype PoodleID = PoodleID Int deriving (Eq, Enum, Show) -data Action p = Create PoodleID (Behaviour p (PoodleID, Sprite)) | Destroy PoodleID +data Action = Create PoodleID (Behaviour (PoodleID, Sprite)) | Destroy PoodleID -poodleGame :: forall p . Typeable p => StdGen -> Game p +poodleGame :: StdGen -> Game poodleGame rng eMouse time = do -- Random times for appearance of new poodles @@ -80,7 +78,7 @@ rec -- Destroy poodles that are clicked on - let eDestructions = justE $ attachWith (\mev poodles -> + let eDestructions = filterJust $ snapshotWith (\mev poodles -> case mev of MouseDown clickPos -> listToMaybe [ Destroy iD | (iD, (rect, _)) <- poodles, diff --git a/examples/tests.hs b/examples/tests.hs --- a/examples/tests.hs +++ b/examples/tests.hs @@ -1,42 +1,37 @@ -{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable, EmptyDataDecls, DoRec #-} +{-# LANGUAGE ScopedTypeVariables, EmptyDataDecls, DoRec #-} import FRP.Sodium -import FRP.Sodium.Internal import Control.Applicative import Control.Monad import Control.Monad.Trans import Data.Char import Data.IORef -import Data.Typeable import Test.HUnit -data M deriving Typeable -- Define a main partition -data N deriving Typeable -- Define a secondary partition - event1 = TestCase $ do - (ev :: Event M Char, push) <- newEvent + (ev, push) <- sync newEvent outRef <- newIORef "" - synchronously $ do + sync $ do push '?' - unlisten <- synchronously $ do + unlisten <- sync $ do push 'h' + unlisten <- listen ev $ \letter -> modifyIORef outRef (++ [letter]) push 'e' - unlisten <- listenIO ev $ \letter -> modifyIORef outRef (++ [letter]) - push 'l' return unlisten - synchronously $ do + sync $ do push 'l' + push 'l' push 'o' unlisten - synchronously $ do + sync $ do push '!' out <- readIORef outRef assertEqual "event1" "hello" =<< readIORef outRef fmap1 = TestCase $ do - (ev :: Event M Char, push) <- newEvent + (ev, push) <- sync newEvent outRef <- newIORef "" - synchronously $ do - listenIO (toUpper `fmap` ev) $ \letter -> modifyIORef outRef (++ [letter]) + sync $ do + listen (toUpper `fmap` ev) $ \letter -> modifyIORef outRef (++ [letter]) push 'h' push 'e' push 'l' @@ -46,35 +41,35 @@ assertEqual "fmap1" "HELLO" =<< readIORef outRef merge1 = TestCase $ do - (ev1 :: Event M String, push1) <- newEvent - (ev2, push2) <- newEvent + (ev1, push1) <- sync newEvent + (ev2, push2) <- sync newEvent let ev = merge ev1 ev2 outRef <- newIORef [] - unlisten <- synchronously $ listenIO ev $ \a -> modifyIORef outRef (++ [a]) - synchronously $ do + unlisten <- sync $ listen ev $ \a -> modifyIORef outRef (++ [a]) + sync $ do push1 "hello" push2 "world" - synchronously $ push1 "people" - synchronously $ push1 "everywhere" + sync $ push1 "people" + sync $ push1 "everywhere" unlisten assertEqual "merge1" ["hello","world","people","everywhere"] =<< readIORef outRef -justE1 = TestCase $ do - (ema :: Event M (Maybe String), push) <- newEvent +filterJust1 = TestCase $ do + (ema, push) <- sync newEvent outRef <- newIORef [] - synchronously $ do - listenIO (justE ema) $ \a -> modifyIORef outRef (++ [a]) + sync $ do + listen (filterJust ema) $ \a -> modifyIORef outRef (++ [a]) push (Just "yes") push Nothing push (Just "no") - assertEqual "justE1" ["yes", "no"] =<< readIORef outRef + assertEqual "filterJust1" ["yes", "no"] =<< readIORef outRef filterE1 = TestCase $ do - (ec, push) <- newEvent + (ec, push) <- sync newEvent outRef <- newIORef "" - synchronously $ do - let ed = filterE isDigit (ec :: Event M Char) - listenIO ed $ \a -> modifyIORef outRef (++ [a]) + sync $ do + let ed = filterE isDigit ec + listen ed $ \a -> modifyIORef outRef (++ [a]) push 'a' push '2' push 'X' @@ -82,34 +77,34 @@ assertEqual "filterE1" "23" =<< readIORef outRef beh1 = TestCase $ do - (ea :: Event M String, push) <- newEvent outRef <- newIORef [] - unlisten <- synchronously $ do - beh <- hold "init" ea - listenValueIO beh $ \a -> modifyIORef outRef (++ [a]) - synchronously $ do + (push, unlisten) <- sync $ do + (beh, push) <- newBehavior "init" + unlisten <- listenValue beh $ \a -> modifyIORef outRef (++ [a]) + return (push, unlisten) + sync $ do push "next" unlisten assertEqual "beh1" ["init", "next"] =<< readIORef outRef beh2 = TestCase $ do - (ea :: Event M String, push) <- newEvent outRef <- newIORef [] - unlisten <- synchronously $ do - beh <- hold "init" ea - listenValueIO beh $ \a -> modifyIORef outRef (++ [a]) + (push, unlisten) <- sync $ do + (beh, push) <- newBehavior "init" + unlisten <- listenValue beh $ \a -> modifyIORef outRef (++ [a]) + return (push, unlisten) unlisten - synchronously $ do + sync $ do push "next" assertEqual "beh2" ["init"] =<< readIORef outRef beh3 = TestCase $ do - (ea :: Event M String, push) <- newEvent outRef <- newIORef [] - unlisten <- synchronously $ do - beh <- hold "init" ea - listenValueIO beh $ \a -> modifyIORef outRef (++ [a]) - synchronously $ do + (push, unlisten) <- sync $ do + (beh, push) <- newBehavior "init" + unlisten <- listenValue beh $ \a -> modifyIORef outRef (++ [a]) + return (push, unlisten) + sync $ do push "first" push "second" unlisten @@ -118,315 +113,300 @@ -- | This demonstrates the fact that if there are multiple updates to a behaviour -- in a given transaction, the last one prevails. beh4 = TestCase $ do - (ea :: Event M String, push) <- newEvent outRef <- newIORef [] - unlisten <- synchronously $ do - beh <- hold "init" ea - unlisten <- listenValueIO beh $ \a -> modifyIORef outRef (++ [a]) + (push, unlisten) <- sync $ do + (beh, push) <- newBehavior "init" + unlisten <- listenValue beh $ \a -> modifyIORef outRef (++ [a]) push "other" - return unlisten - synchronously $ do + return (push, unlisten) + sync $ do push "first" push "second" unlisten assertEqual "beh4" ["other", "second"] =<< readIORef outRef beh5 = TestCase $ do - (ea :: Event M String, push) <- newEvent - outRef <- newIORef [] - unlisten <- synchronously $ do - beh <- hold "init" ea - unlisten <- listenIO (valueEvent beh) $ \a -> modifyIORef outRef (++ [a]) - push "other" - return unlisten - synchronously $ do - push "first" - push "second" - unlisten - assertEqual "beh5" ["other", "second"] =<< readIORef outRef - -beh6 = TestCase $ do - (ea :: Event M String, push) <- newEvent + (ea, push) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do beh <- hold "init" ea - unlisten <- listenIO (map toUpper <$> valueEvent beh) $ \a -> modifyIORef outRef (++ [a]) + unlisten <- listen (map toUpper <$> values beh) $ \a -> modifyIORef outRef (++ [a]) push "other" return unlisten - synchronously $ do + sync $ do push "first" push "second" unlisten - assertEqual "beh6" ["OTHER", "SECOND"] =<< readIORef outRef + assertEqual "beh5" ["OTHER", "SECOND"] =<< readIORef outRef appl1 = TestCase $ do - (ea :: Event M Int, pusha) <- newEvent - ba <- synchronously $ hold 0 ea - (eb, pushb) <- newEvent - bb <- synchronously $ hold 0 eb + (ea, pusha) <- sync newEvent + ba <- sync $ hold 0 ea + (eb, pushb) <- sync newEvent + bb <- sync $ hold 0 eb let esum = (+) <$> ba <*> bb outRef <- newIORef [] - unlisten <- synchronously $ listenValueIO esum $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ pusha 5 - synchronously $ pushb 100 - synchronously $ pusha 10 >> pushb 200 + unlisten <- sync $ listenValue esum $ \sum -> modifyIORef outRef (++ [sum]) + sync $ pusha 5 + sync $ pushb 100 + sync $ pusha 10 >> pushb 200 unlisten assertEqual "appl1" [0, 5, 105, 210] =<< readIORef outRef -appl2 = TestCase $ do -- variant that uses listenIO (valueEvent esum) instead of listenValueIO - (ea :: Event M Int, pusha) <- newEvent - ba <- synchronously $ hold 0 ea - (eb, pushb) <- newEvent - bb <- synchronously $ hold 0 eb +appl2 = TestCase $ do -- variant that uses listen (valueEvent esum) instead of listenValue + (ea, pusha) <- sync newEvent + ba <- sync $ hold 0 ea + (eb, pushb) <- sync newEvent + bb <- sync $ hold 0 eb let esum = (+) <$> ba <*> bb outRef <- newIORef [] - unlisten <- synchronously $ listenIO (valueEvent esum) $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ pusha 5 - synchronously $ pushb 100 - synchronously $ pusha 10 >> pushb 200 + unlisten <- sync $ listen (values esum) $ \sum -> modifyIORef outRef (++ [sum]) + sync $ pusha 5 + sync $ pushb 100 + sync $ pusha 10 >> pushb 200 unlisten assertEqual "appl2" [0, 5, 105, 210] =<< readIORef outRef -attach1 = TestCase $ do - (ea :: Event M Char, pusha) <- newEvent - (eb :: Event M Int, pushb) <- newEvent - bb <- synchronously $ hold 0 eb - let ec = attach ea bb +snapshot1 = TestCase $ do + (ea, pusha) <- sync newEvent + (eb, pushb) <- sync newEvent + bb <- sync $ hold 0 eb + let ec = snapshotWith (,) ea bb outRef <- newIORef [] - unlisten <- synchronously $ listenIO ec $ \c -> modifyIORef outRef (++ [c]) - synchronously $ pusha 'A' - synchronously $ pushb 50 - synchronously $ pusha 'B' - synchronously $ pusha 'C' >> pushb 60 - synchronously $ pusha 'D' + unlisten <- sync $ listen ec $ \c -> modifyIORef outRef (++ [c]) + sync $ pusha 'A' + sync $ pushb 50 + sync $ pusha 'B' + sync $ pusha 'C' >> pushb 60 + sync $ pusha 'D' unlisten - assertEqual "attach1" [('A',0),('B',50),('C',50),('D',60)] =<< readIORef outRef + assertEqual "snapshot1" [('A',0),('B',50),('C',50),('D',60)] =<< readIORef outRef count1 = TestCase $ do - (ea :: Event M (), push) <- newEvent + (ea, push) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do eCount <- countE ea - listenIO eCount $ \c -> modifyIORef outRef (++ [c]) - synchronously $ push () - synchronously $ push () - synchronously $ push () + listen eCount $ \c -> modifyIORef outRef (++ [c]) + sync $ push () + sync $ push () + sync $ push () unlisten assertEqual "count1" [1,2,3] =<< readIORef outRef collect1 = TestCase $ do - (ea :: Event M Int, push) <- newEvent + (ea, push) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do ba <- hold 100 ea sum <- collect (\a s -> (a+s, a+s)) 0 ba - listenValueIO sum $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ push 5 - synchronously $ push 7 - synchronously $ push 1 - synchronously $ push 2 - synchronously $ push 3 + listenValue sum $ \sum -> modifyIORef outRef (++ [sum]) + sync $ push 5 + sync $ push 7 + sync $ push 1 + sync $ push 2 + sync $ push 3 unlisten assertEqual "collect1" [100, 105, 112, 113, 115, 118] =<< readIORef outRef collect2 = TestCase $ do - (ea :: Event M Int, push) <- newEvent outRef <- newIORef [] - -- This behaviour is a little bit odd but difficult to fix in the - -- implementation. However, it shouldn't be too much of a problem in - -- practice. Here we are defining it. - unlisten <- synchronously $ do - ba <- hold 100 ea + -- This is a bit of an edge case. + (unlisten, push) <- sync $ do + (ba, push) <- newBehavior 100 sum <- collect (\a s -> (a + s, a + s)) 0 ba push 5 - listenValueIO sum $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ push 7 - synchronously $ push 1 + unlisten <- listenValue sum $ \sum -> modifyIORef outRef (++ [sum]) + return (unlisten, push) + sync $ push 7 + sync $ push 1 unlisten - assertEqual "collect2" [105, 112, 113] =<< readIORef outRef + assertEqual "collect2" [100, 105, 112, 113] =<< readIORef outRef collectE1 = TestCase $ do - (ea :: Event M Int, push) <- newEvent + (ea, push) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do sum <- collectE (\a s -> (a+s, a+s)) 100 ea - listenIO sum $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ push 5 - synchronously $ push 7 - synchronously $ push 1 - synchronously $ push 2 - synchronously $ push 3 + listen sum $ \sum -> modifyIORef outRef (++ [sum]) + sync $ push 5 + sync $ push 7 + sync $ push 1 + sync $ push 2 + sync $ push 3 unlisten assertEqual "collectE1" [105, 112, 113, 115, 118] =<< readIORef outRef collectE2 = TestCase $ do - (ea :: Event M Int, push) <- newEvent + (ea, push) <- sync newEvent outRef <- newIORef [] -- This behaviour is a little bit odd but difficult to fix in the -- implementation. However, it shouldn't be too much of a problem in -- practice. Here we are defining it. - unlisten <- synchronously $ do + unlisten <- sync $ do sum <- collectE (\a s -> (a + s, a + s)) 100 ea push 5 - listenIO sum $ \sum -> modifyIORef outRef (++ [sum]) - synchronously $ push 7 - synchronously $ push 1 + listen sum $ \sum -> modifyIORef outRef (++ [sum]) + sync $ push 7 + sync $ push 1 unlisten assertEqual "collectE2" [105, 112, 113] =<< readIORef outRef switchE1 = TestCase $ do - (ea :: Event M Char, pusha) <- newEvent - (eb :: Event M Char, pushb) <- newEvent - (esw :: Event M (Event M Char), pushsw) <- newEvent + (ea, pusha) <- sync newEvent + (eb, pushb) <- sync newEvent + (esw, pushsw) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do sw <- hold ea esw let eo = switchE sw - unlisten <- listenIO eo $ \o -> modifyIORef outRef (++ [o]) + unlisten <- listen eo $ \o -> modifyIORef outRef (++ [o]) pusha 'A' pushb 'a' return unlisten - synchronously $ pusha 'B' >> pushb 'b' - synchronously $ pushsw eb >> pusha 'C' >> pushb 'c' - synchronously $ pusha 'D' >> pushb 'd' - synchronously $ pusha 'E' >> pushb 'e' >> pushsw ea - synchronously $ pusha 'F' >> pushb 'f' - synchronously $ pusha 'G' >> pushb 'g' >> pushsw eb - synchronously $ pusha 'H' >> pushb 'h' >> pushsw ea - synchronously $ pusha 'I' >> pushb 'i' >> pushsw ea + sync $ pusha 'B' >> pushb 'b' + sync $ pushsw eb >> pusha 'C' >> pushb 'c' + sync $ pusha 'D' >> pushb 'd' + sync $ pusha 'E' >> pushb 'e' >> pushsw ea + sync $ pusha 'F' >> pushb 'f' + sync $ pusha 'G' >> pushb 'g' >> pushsw eb + sync $ pusha 'H' >> pushb 'h' >> pushsw ea + sync $ pusha 'I' >> pushb 'i' >> pushsw ea unlisten assertEqual "switchE1" "ABCdeFGhI" =<< readIORef outRef switch1 = TestCase $ do - (ea :: Event M Char, pusha) <- newEvent - (eb :: Event M Char, pushb) <- newEvent - (esw :: Event M (Behaviour M Char), pushsw) <- newEvent + (ea, pusha) <- sync newEvent + (eb, pushb) <- sync newEvent + (esw, pushsw) <- sync newEvent outRef <- newIORef [] - (ba, bb, unlisten) <- synchronously $ do + (ba, bb, unlisten) <- sync $ do ba <- hold 'A' ea bb <- hold 'a' eb bsw <- hold ba esw bo <- switch bsw - unlisten <- listenValueIO bo $ \o -> modifyIORef outRef (++ [o]) + unlisten <- listenValue bo $ \o -> modifyIORef outRef (++ [o]) return (ba, bb, unlisten) - synchronously $ pusha 'B' >> pushb 'b' - synchronously $ pushsw bb >> pusha 'C' >> pushb 'c' - synchronously $ pusha 'D' >> pushb 'd' - synchronously $ pusha 'E' >> pushb 'e' >> pushsw ba - synchronously $ pusha 'F' >> pushb 'f' - synchronously $ pushsw bb - synchronously $ pushsw ba - synchronously $ pusha 'G' >> pushb 'g' >> pushsw bb - synchronously $ pusha 'H' >> pushb 'h' >> pushsw ba - synchronously $ pusha 'I' >> pushb 'i' >> pushsw ba + sync $ pusha 'B' >> pushb 'b' + sync $ pushsw bb >> pusha 'C' >> pushb 'c' + sync $ pusha 'D' >> pushb 'd' + sync $ pusha 'E' >> pushb 'e' >> pushsw ba + sync $ pusha 'F' >> pushb 'f' + sync $ pushsw bb + sync $ pushsw ba + sync $ pusha 'G' >> pushb 'g' >> pushsw bb + sync $ pusha 'H' >> pushb 'h' >> pushsw ba + sync $ pusha 'I' >> pushb 'i' >> pushsw ba unlisten assertEqual "switch1" "ABcdEFfFgHI" =<< readIORef outRef once1 = TestCase $ do - (ea :: Event M Char, pusha) <- newEvent + (ea, pusha) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do oea <- once ea - listenIO oea $ \a -> modifyIORef outRef (++ [a]) - synchronously $ pusha 'A' - synchronously $ pusha 'B' - synchronously $ pusha 'C' + listen oea $ \a -> modifyIORef outRef (++ [a]) + sync $ pusha 'A' + sync $ pusha 'B' + sync $ pusha 'C' unlisten assertEqual "switch1" "A" =<< readIORef outRef once2 = TestCase $ do - (ea :: Event M Char, pusha) <- newEvent + (ea, pusha) <- sync newEvent outRef <- newIORef [] - unlisten <- synchronously $ do + unlisten <- sync $ do oea <- once ea pusha 'A' - listenIO oea $ \a -> modifyIORef outRef (++ [a]) - synchronously $ pusha 'B' - synchronously $ pusha 'C' + listen oea $ \a -> modifyIORef outRef (++ [a]) + sync $ pusha 'B' + sync $ pusha 'C' unlisten assertEqual "switch1" "A" =<< readIORef outRef +{- crossE1 = TestCase $ do outRef <- newIORef [] - (ema :: Event M Char, push) <- newEvent - (ena :: Event N Char) <- synchronously $ crossE ema - unlisten <- synchronously $ listenIO ena $ \a -> modifyIORef outRef (++ [a]) - synchronously $ push 'A' - synchronously $ push 'M' - synchronously $ push 'T' + (ema :: Event Plain Char, push) <- newEvent + (ena :: Event N Char) <- sync $ crossE ema + unlisten <- sync $ listen ena $ \a -> modifyIORef outRef (++ [a]) + sync $ push 'A' + sync $ push 'M' + sync $ push 'T' -- Flush processing on partition N before unlistening - synchronously (return () :: Reactive N ()) + sync (return () :: Reactive N ()) unlisten assertEqual "crossE1" "AMT" =<< readIORef outRef cross1 = TestCase $ do outRef <- newIORef [] - (ema :: Event M Char, push) <- newEvent - bma <- synchronously $ hold 'A' ema - synchronously $ push 'B' - (bna :: Behaviour N Char) <- synchronously $ cross bma - unlisten <- synchronously $ listenValueIO bna $ \a -> modifyIORef outRef (++ [a]) - synchronously $ push 'C' - synchronously $ push 'D' - synchronously $ push 'E' + (ema :: Event Plain Char, push) <- newEvent + bma <- sync $ hold 'A' ema + sync $ push 'B' + (bna :: Behavior N Char) <- sync $ cross bma + unlisten <- sync $ listenValue bna $ \a -> modifyIORef outRef (++ [a]) + sync $ push 'C' + sync $ push 'D' + sync $ push 'E' -- Flush processing on partition N before unlistening - synchronously (return () :: Reactive N ()) + sync (return () :: Reactive N ()) unlisten assertEqual "cross1" "BCDE" =<< readIORef outRef cross2 = TestCase $ do outRef <- newIORef [] - (ema :: Event M Char, push) <- newEvent - bma <- synchronously $ hold 'A' ema - (bna :: Behaviour N Char) <- synchronously $ cross bma - unlisten <- synchronously $ listenValueIO bna $ \a -> modifyIORef outRef (++ [a]) - synchronously $ push 'B' - synchronously $ push 'C' - synchronously $ push 'D' - synchronously $ push 'E' + (ema :: Event Plain Char, push) <- newEvent + bma <- sync $ hold 'A' ema + (bna :: Behavior N Char) <- sync $ cross bma + unlisten <- sync $ listenValue bna $ \a -> modifyIORef outRef (++ [a]) + sync $ push 'B' + sync $ push 'C' + sync $ push 'D' + sync $ push 'E' -- Flush processing on partition N before unlistening - synchronously (return () :: Reactive N ()) + sync (return () :: Reactive N ()) unlisten assertEqual "cross1" "ABCDE" =<< readIORef outRef +-} -data Page p = Page { unPage :: Reactive p (Char, Event p (Page p)) } +data Page = Page { unPage :: Reactive (Char, Event Page) } cycle1 = TestCase $ do outRef <- newIORef [] - (ep :: Event M (Page M), push) <- newEvent - bo <- synchronously $ do + (ep, push) <- sync newEvent + bo <- sync $ do let initPair = ('a', ep) rec bPair <- hold initPair ePage let ePage = execute $ unPage <$> switchE (snd <$> bPair) return (fst <$> bPair) - unlisten <- synchronously $ listenValueIO bo $ \o -> modifyIORef outRef (++ [o]) - synchronously $ push (Page $ return ('b', ep)) - synchronously $ push (Page $ return ('c', ep)) + unlisten <- sync $ listenValue bo $ \o -> modifyIORef outRef (++ [o]) + sync $ push (Page $ return ('b', ep)) + sync $ push (Page $ return ('c', ep)) unlisten assertEqual "cycle1" "abc" =<< readIORef outRef mergeWith1 = TestCase $ do outRef <- newIORef [] - (ea :: Event M Int, pushA) <- newEvent - (eb :: Event M Int, pushB) <- newEvent - unlisten <- synchronously $ do + (ea, pushA) <- sync newEvent + (eb, pushB) <- sync newEvent + unlisten <- sync $ do pushA 5 - listenIO (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) - synchronously $ pushA 2 - synchronously $ pushB 3 - synchronously $ pushA 10 >> pushB 4 - synchronously $ pushB 7 >> pushA 1 + listen (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) + sync $ pushA 2 + sync $ pushB 3 + sync $ pushA 10 >> pushB 4 + sync $ pushB 7 >> pushA 1 unlisten assertEqual "mergeWith1" [5,2,3,14,8] =<< readIORef outRef mergeWith2 = TestCase $ do outRef <- newIORef [] - (ea :: Event M Int, pushA) <- newEvent - (eb :: Event M Int, pushB) <- newEvent - unlisten <- synchronously $ do + (ea, pushA) <- sync newEvent + (eb, pushB) <- sync newEvent + unlisten <- sync $ do pushA 5 - unlisten <- listenIO (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) + unlisten <- listen (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) pushB 99 return unlisten unlisten @@ -434,20 +414,32 @@ mergeWith3 = TestCase $ do outRef <- newIORef [] - (ea :: Event M Int, pushA) <- newEvent - (eb :: Event M Int, pushB) <- newEvent - unlisten <- synchronously $ do - listenIO (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) - synchronously $ pushA 2 - synchronously $ pushB 3 >> pushB 1 >> pushA 10 - synchronously $ pushB 9 >> pushB 11 >> pushB 12 - synchronously $ pushA 32 >> pushA 11 >> pushA 12 + (ea, pushA) <- sync newEvent + (eb, pushB) <- sync newEvent + unlisten <- sync $ do + listen (mergeWith (+) ea eb) $ \o -> modifyIORef outRef (++ [o]) + sync $ pushA 2 + sync $ pushB 3 >> pushB 1 >> pushA 10 + sync $ pushB 9 >> pushB 11 >> pushB 12 + sync $ pushA 32 >> pushA 11 >> pushA 12 unlisten assertEqual "mergeWith3" [2,14,32,55] =<< readIORef outRef -tests = test [ event1, fmap1, merge1, justE1, filterE1, beh1, beh2, beh3, beh4, beh5, beh6, - appl1, appl2, attach1, count1, collect1, collect2, collectE1, collectE2, switchE1, - switch1, once1, once2, crossE1, cross1, cross2, cycle1, mergeWith1, mergeWith2, mergeWith3 ] +coalesce1 = TestCase $ do + outRef <- newIORef [] + (ea, pushA) <- sync newEvent + (eb, pushB) <- sync newEvent + unlisten <- sync $ do + listen (coalesce (+) (merge ea eb)) $ \o -> modifyIORef outRef (++ [o]) + sync $ pushA 2 + sync $ pushA 5 >> pushB 6 + unlisten + assertEqual "coalesce1" [2, 11] =<< readIORef outRef + +tests = test [ event1, fmap1, merge1, filterJust1, filterE1, beh1, beh2, beh3, beh4, beh5, + appl1, appl2, snapshot1, count1, collect1, collect2, collectE1, collectE2, switchE1, + switch1, once1, once2, {-crossE1, cross1, cross2,-} cycle1, mergeWith1, mergeWith2, mergeWith3, + coalesce1 ] main = {-forever $-} runTestTT tests diff --git a/sodium.cabal b/sodium.cabal --- a/sodium.cabal +++ b/sodium.cabal @@ -1,17 +1,15 @@ name: sodium -version: 0.3.0.1 +version: 0.4.0.0 synopsis: Sodium Reactive Programming (FRP) System description: - A general purpose Reactive Programming (FRP) system. + A general purpose Reactive Programming (FRP) system. This is part of a project to + implement reactive libraries with similar interfaces across a range of programming + languages. . - * Goals include simplicity and completeness, but it is currently not built for speed. + * Goals include simplicity and completeness. . * Applicative style: Event implements Functor and Behaviour implements Applicative. . - * FRP logic is tied to partitions, within which consistency is guaranteed. - This concept allows you to selectively relax consistency guarantees to - facilitate parallelism. - . * Instead of the common approach where inputs are fed into the front of a monolithic \'reactimate\', Sodium allows you to push inputs in from scattered places in IO. . @@ -23,7 +21,8 @@ See the /examples/ directory for test cases and examples. . Changes: 0.2.0.0 fix some value recursion deadlocks and improve docs; - 0.3.0.0 add mergeWith, make cross asynchronous + 0.3.0.0 add mergeWith, make cross asynchronous; + 0.4.0.0 API revamp to remove an excess type variable. Parallelism stuff to be rethought. license: BSD3 license-file: LICENSE author: Stephen Blackheath @@ -40,12 +39,12 @@ source-repository head type: git - location: https://github.com/the-real-blackh/sodium + location: https://github.com/kentuckyfriedtakahe/sodium library hs-source-dirs: src - exposed-modules: FRP.Sodium, FRP.Sodium.Internal - other-modules: FRP.Sodium.Impl + exposed-modules: FRP.Sodium, FRP.Sodium.Internal, FRP.Sodium.Context + other-modules: FRP.Sodium.Plain build-depends: base >= 4.3.0.0 && < 4.6.0.0, containers >= 0.4.0.0 && < 0.5.0.0, mtl >= 2.0.0.0 && < 2.2.0.0 diff --git a/src/FRP/Sodium.hs b/src/FRP/Sodium.hs --- a/src/FRP/Sodium.hs +++ b/src/FRP/Sodium.hs @@ -1,38 +1,21 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving, ScopedTypeVariables, DoRec, GADTs, + TypeFamilies, EmptyDataDecls, FlexibleInstances #-} +{-# OPTIONS_GHC -fno-cse -fno-full-laziness #-} -- | Sodium Reactive Programming (FRP) system. -- -- See the /examples/ directory for test cases and examples. -- --- The @p@ type parameter determines the /partition/ that your FRP is running --- on. A thread is automatically created for each partition used in the system based --- on the unique concrete p type, which must be an instance of Typeable. FRP --- processing runs on this thread, but 'synchronously' will block the calling thread --- while it waits for FRP processing to complete. --- --- In most cases you would just use one concrete partition type for everything, defined --- like this: --- --- > {-# LANGUAGE DeriveDataTypeable, EmptyDataDecls #-} --- > import Data.Typeable --- > --- > data M deriving Typeable --- --- Later, if you want your code to be more parallel, you can add more partitions: --- The 'cross' and 'crossE' functions are used to move events and behaviours between --- partitions. The separation thus created allows your FRP logic to be partitioned --- so that the different partitions can run in parallel, with more relaxed guarantees --- of consistency between partitions. --- -- Some functions are pure, and others need to run under the 'Reactive' monad via --- 'synchronously' or 'asynchronously'. An 'Event' /p (/'Reactive' /p a)/ can be flattened --- to an 'Event' /p a/ using the 'execute' primitive. +-- 'sync'. An 'Event' /(/'Reactive' /a)/ can be flattened +-- to an 'Event' /a/ using the 'execute' primitive. -- -- In addition to the explicit functions in the language, note that you can use -- --- * Functor on 'Event' and 'Behaviour' +-- * Functor on 'Event' and 'Behavior' -- -- * Applicative on 'behaviour', e.g. @let bsum = (+) \<$\> ba \<*\> bb@ -- --- * Applicative 'pure' is used to give a constant 'Behaviour'. +-- * Applicative 'pure' is used to give a constant 'Behavior'. -- -- * Recursive do (via DoRec) to make state loops with the @rec@ keyword. -- @@ -41,39 +24,41 @@ -- -- > {-# LANGUAGE DoRec #-} -- > -- | Accumulate on input event, outputting the new state each time. --- > accumE :: Typeable p => (a -> s -> s) -> s -> Event p a -> Reactive p (Event p s) +-- > accumE :: (a -> s -> s) -> s -> Event a -> Reactive (Event s) -- > accumE f z ea = do -- > rec -- > let es = attachWith f ea s -- > s <- hold z es -- > return es module FRP.Sodium ( + Plain, -- * Running FRP code Reactive, - synchronously, - asynchronously, + sync, newEvent, - listenIO, - listenValueIO, + newBehavior, + listen, + listenValue, -- * FRP core language Event, - Behaviour, Behavior, + Behaviour, never, merge, - mergeWith, - justE, + filterJust, hold, - valueEvent, - attachWith, + changes, + values, + snapshotWith, switchE, switch, execute, sample, + coalesce, -- * Derived FRP functions + mergeWith, filterE, - attach, - tag, + snapshot, gate, collectE, collect, @@ -82,10 +67,7 @@ countE, count, once, - -- * Partitions - crossE, - cross ) where -import FRP.Sodium.Impl +import FRP.Sodium.Plain diff --git a/src/FRP/Sodium/Context.hs b/src/FRP/Sodium/Context.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Sodium/Context.hs @@ -0,0 +1,165 @@ +{-# LANGUAGE TypeFamilies, DoRec, FlexibleContexts, ScopedTypeVariables #-} +-- | Generalization of the Sodium API to allow for parallel processing. +module FRP.Sodium.Context where + +import Control.Applicative +import Control.Monad +import Control.Monad.Fix + +class ( + Applicative (Reactive r), + Monad (Reactive r), + MonadFix (Reactive r), + Functor (Event r), + Applicative (Behavior r) + ) => + Context r where + -- | A monad for transactional reactive operations. Execute it from 'IO' using 'sync'. + data Reactive r :: * -> * + -- | A stream of events. The individual firings of events are called \'event occurrences\'. + data Event r :: * -> * + -- | A time-varying value, American spelling. + data Behavior r :: * -> * + -- | Execute the specified 'Reactive' within a new transaction, blocking the caller + -- until all resulting processing is complete and all callbacks have been called. + -- This operation is thread-safe, so it may be called from any thread. + -- + -- State changes to 'hold' values occur after processing of the transaction is complete. + sync :: Reactive r a -> IO a + -- Lift an arbitrary IO action into a 'Reactive'. + ioReactive :: IO a -> Reactive r a + -- | Returns an event, and a push action for pushing a value into the event. + newEvent :: Reactive r (Event r a, a -> Reactive r ()) + -- | Listen for firings of this event. The returned @IO ()@ is an IO action + -- that unregisters the listener. This is the observer pattern. + listen :: Event r a -> (a -> IO ()) -> Reactive r (IO ()) + -- | An event that never fires. + never :: Event r a + -- | Merge two streams of events of the same type. + -- + -- In the case where two event occurrences are simultaneous (i.e. both + -- within the same transaction), both will be delivered in the same + -- transaction. + -- + -- The order is not defined, because simultaneous events should be considered + -- to be order-agnostic. + merge :: Event r a -> Event r a -> Event r a + -- | Unwrap Just values, and discard event occurrences with Nothing values. + filterJust :: Event r (Maybe a) -> Event r a + -- | Create a behaviour with the specified initial value, that gets updated + -- by the values coming through the event. The \'current value\' of the behaviour + -- is notionally the value as it was 'at the start of the transaction'. + -- That is, state updates caused by event firings get processed at the end of + -- the transaction. + hold :: a -> Event r a -> Reactive r (Behavior r a) + -- | An event that gives the updates for the behaviour. It doesn't do any equality + -- comparison as the name might imply. + changes :: Behavior r a -> Event r a + -- | An event that is guaranteed to fires once when you listen to it, giving + -- the current value of the behaviour, and thereafter behaves like 'changes', + -- firing for each update to the behaviour's value. + values :: Behavior r a -> Event r a + -- | Sample the behaviour at the time of the event firing. Note that the 'current value' + -- of the behaviour that's sampled is the value as at the start of the transaction + -- before any state changes of the current transaction are applied through 'hold's. + snapshotWith :: (a -> b -> c) -> Event r a -> Behavior r b -> Event r c + -- | Unwrap an event inside a behaviour to give a time-varying event implementation. + switchE :: Behavior r (Event r a) -> Event r a + -- | Unwrap a behaviour inside another behaviour to give a time-varying behaviour implementation. + switch :: Behavior r (Behavior r a) -> Reactive r (Behavior r a) + -- | Execute the specified 'Reactive' action inside an event. + execute :: Event r (Reactive r a) -> Event r a + -- | Obtain the current value of a behaviour. + sample :: Behavior r a -> Reactive r a + -- | If there's more than one firing in a single transaction, combine them into + -- one using the specified combining function. + coalesce :: (a -> a -> a) -> Event r a -> Event r a + +newBehavior :: forall r a . Context r => + a -- ^ Initial behaviour value + -> Reactive r (Behavior r a, a -> Reactive r ()) +newBehavior initA = do + (ev, push) <- newEvent + beh <- hold initA ev + return (beh, push) + +listenValue :: Context r => Behavior r a -> (a -> IO ()) -> Reactive r (IO ()) +listenValue b handler = listen (values b) handler + +-- | Merge two streams of events of the same type, combining simultaneous +-- event occurrences. +-- +-- In the case where multiple event occurrences are simultaneous (i.e. all +-- within the same transaction), they are combined using the supplied +-- function. The output event is guaranteed not to have more than one +-- event occurrence per transaction. +-- +-- The combine function should be commutative, because simultaneous events +-- should be considered to be order-agnostic. +mergeWith :: Context r => (a -> a -> a) -> Event r a -> Event r a -> Event r a +mergeWith f ea eb = coalesce f $ merge ea eb + +-- | Only keep event occurrences for which the predicate is true. +filterE :: Context r => (a -> Bool) -> Event r a -> Event r a +filterE pred = filterJust . ((\a -> if pred a then Just a else Nothing) <$>) + +-- | Variant of snapshotWith that throws away the event's value and captures the behaviour's. +snapshot :: Context r => Event r a -> Behavior r b -> Event r b +snapshot = snapshotWith (flip const) + +-- | Let event occurrences through only when the behaviour's value is True. +-- Note that the behaviour's value is as it was at the start of the transaction, +-- that is, no state changes from the current transaction are taken into account. +gate :: Context r => Event r a -> Behavior r Bool -> Event r a +gate ea = filterJust . snapshotWith (\a b -> if b then Just a else Nothing) ea + +-- | Transform an event with a generalized state loop (a mealy machine). The function +-- is passed the input and the old state and returns the new state and output value. +collectE :: Context r => (a -> s -> (b, s)) -> s -> Event r a -> Reactive r (Event r b) +collectE f z ea = do + rec + s <- hold z es + let ebs = snapshotWith f ea s + eb = fst <$> ebs + es = snd <$> ebs + return eb + +-- | Transform a behaviour with a generalized state loop (a mealy machine). The function +-- is passed the input and the old state and returns the new state and output value. +collect :: Context r => (a -> s -> (b, s)) -> s -> Behavior r a -> Reactive r (Behavior r b) +collect f zs bea = do + let ea = coalesce (flip const) (changes bea) + za <- sample bea + let (zb, zs') = f za zs + rec + bs <- hold (zb, zs') ebs + let ebs = snapshotWith f ea (snd <$> bs) + return (fst <$> bs) + +-- | Accumulate on input event, outputting the new state each time. +accumE :: Context r => (a -> s -> s) -> s -> Event r a -> Reactive r (Event r s) +accumE f z ea = do + rec + let es = snapshotWith f ea s + s <- hold z es + return es + +-- | Accumulate on input event, holding state. +accum :: Context r => (a -> s -> s) -> s -> Event r a -> Reactive r (Behavior r s) +accum f z ea = do + rec + s <- hold z (snapshotWith f ea s) + return s + +-- | Count event occurrences, starting with 1 for the first occurrence. +countE :: Context r => Event r a -> Reactive r (Event r Int) +countE = accumE (+) 0 . (const 1 <$>) + +-- | Count event occurrences, giving a behaviour that starts with 0 before the first occurrence. +count :: Context r => Event r a -> Reactive r (Behavior r Int) +count = hold 0 <=< countE + +-- | Throw away all event occurrences except for the first one. +once :: Context r => Event r a -> Reactive r (Event r a) +once ea = filterJust <$> collectE (\a active -> (if active then Just a else Nothing, False)) True ea + diff --git a/src/FRP/Sodium/Impl.hs b/src/FRP/Sodium/Impl.hs deleted file mode 100644 --- a/src/FRP/Sodium/Impl.hs +++ /dev/null @@ -1,741 +0,0 @@ -{-# LANGUAGE GeneralizedNewtypeDeriving, ScopedTypeVariables, DoRec, GADTs #-} -{-# OPTIONS_GHC -fno-cse -fno-full-laziness #-} -module FRP.Sodium.Impl where - --- Note: the 'full-laziness' optimization messes up finalizers, so we're --- disabling it. It'd be nice to find a really robust solution to this. --- -fno-cse just in case, since we're using unsafePerformIO. - -import Control.Applicative -import Control.Concurrent -import Control.Concurrent.Chan -import Control.Concurrent.MVar -import Control.Exception (evaluate) -import Control.Monad -import Control.Monad.State.Strict -import Control.Monad.Trans -import Data.Int -import Data.IORef -import Data.Map (Map) -import qualified Data.Map as M -import Data.Maybe -import Data.Set (Set) -import qualified Data.Set as S -import Data.Sequence (Seq, (|>)) -import qualified Data.Sequence as Seq -import Data.Typeable -import GHC.Exts -import System.Mem.Weak -import System.IO.Unsafe -import Unsafe.Coerce - -type ID = Int64 - -data ReactiveState p = ReactiveState { - asQueue1 :: Seq (Reactive p ()), - asQueue2 :: Map Int64 (Reactive p ()), - asFinal :: IO () - } - -{- -newtype Reactive p a = Reactive (StateT (ReactiveState p) IO a) - deriving (Functor, Applicative, Monad, MonadFix) --} - -data Reactive p a where - Reactive :: StateT (ReactiveState p) IO a -> Reactive p a - -instance Functor (Reactive p) where - fmap f rm = Reactive (fmap f (unReactive rm)) - -unReactive :: Reactive p a -> StateT (ReactiveState p) IO a -unReactive (Reactive m) = m - -instance Applicative (Reactive p) where - pure a = Reactive $ return a - rf <*> rm = Reactive $ unReactive rf <*> unReactive rm - -instance Monad (Reactive p) where - return a = Reactive $ return a - rma >>= kmb = Reactive $ do - a <- unReactive rma - unReactive (kmb a) - -instance MonadFix (Reactive p) where - mfix f = Reactive $ mfix $ \a -> unReactive (f a) - -ioReactive :: IO a -> Reactive p a -ioReactive io = Reactive $ liftIO io - -newtype NodeID = NodeID Int deriving (Eq, Ord, Enum) - -data Partition p = Partition { - paRun :: Reactive p () -> IO () -> IO (), - paNextNodeID :: IORef NodeID - } - --- | Queue the specified atomic to run at the end of the priority 1 queue -schedulePriority1 :: Reactive p () -> Reactive p () -schedulePriority1 task = Reactive $ modify $ \as -> as { asQueue1 = asQueue1 as |> task } - -onFinal :: IO () -> Reactive p () -onFinal task = Reactive $ modify $ \as -> as { asFinal = asFinal as >> task } - -partitionRegistry :: MVar (Map String Any) -{-# NOINLINE partitionRegistry #-} -partitionRegistry = unsafePerformIO $ newMVar M.empty - --- | Get the globally unique partition handle for this partition type. -partition :: forall p . Typeable p => IO (Partition p) -partition = do - let typ = show $ typeOf (undefined :: p) - modifyMVar partitionRegistry $ \reg -> - case M.lookup typ reg of - Just part -> return (reg, unsafeCoerce part) - Nothing -> do - part <- createPartition - return (M.insert typ (unsafeCoerce part) reg, part) - -createPartition :: IO (Partition p) -createPartition = do - ch <- newChan - forkIO $ forever $ do - (task, onCompletion) <- readChan ch - let loop = do - queue1 <- gets asQueue1 - if not $ Seq.null queue1 then do - let Reactive task = Seq.index queue1 0 - modify $ \as -> as { asQueue1 = Seq.drop 1 queue1 } - task - loop - else do - queue2 <- gets asQueue2 - if not $ M.null queue2 then do - let (k, Reactive task) = M.findMin queue2 - modify $ \as -> as { asQueue2 = M.delete k queue2 } - task - loop - else do - final <- gets asFinal - liftIO final - return () - runStateT loop $ ReactiveState { - asQueue1 = Seq.singleton task, - asQueue2 = M.empty, - asFinal = return () - } - onCompletion - - nextNodeIDRef <- newIORef (NodeID 0) - return $ Partition { - paRun = \task onCompletion -> writeChan ch (task, onCompletion), - paNextNodeID = nextNodeIDRef - } - --- | Execute the specified 'Reactive' within a new transaction, firing it off without --- waiting for it to complete. It will be queued for executing on the FRP thread --- for the selected partition. --- --- State changes to 'hold' values occur after processing of the transaction is complete. -asynchronously :: Typeable p => Reactive p () -> IO () -asynchronously task = do - part <- partition - paRun part task (return ()) - --- | Execute the specified 'Reactive' within a new transaction, blocking the caller --- until all resulting processing is complete and all callbacks have been called. --- --- State changes to 'hold' values occur after processing of the transaction is complete. -synchronously :: Typeable p => Reactive p a -> IO a -synchronously task = do - mvOutput <- newEmptyMVar - mvCompleted <- newEmptyMVar - part <- partition - paRun part (task >>= ioReactive . putMVar mvOutput) (putMVar mvCompleted ()) - takeMVar mvCompleted - takeMVar mvOutput - -data Listen p a = Listen { runListen_ :: Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) } - -runListen :: Listen p a -> Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) -{-# NOINLINE runListen #-} -runListen l mv handle = do - o <- runListen_ l mv handle - _ <- ioReactive $ evaluate l - return o - --- | A stream of events. The individual firings of events are called \'event occurrences\'. -data Event p a = Event { -- Must be data not newtype, because we need to attach finalizers to it - -- | Listen for event occurrences on this event, to be handled by the specified - -- handler. The returned action is used to unregister the listener. - getListenRaw :: Reactive p (Listen p a), - evCacheRef :: IORef (Maybe (Listen p a)) - } - --- | An event that never fires. -never :: Event p a -never = Event { - getListenRaw = return $ Listen $ \_ _ -> return (return ()), - evCacheRef = unsafePerformIO $ newIORef Nothing - } - --- | Unwrap an event's listener machinery. -getListen :: Event p a -> Reactive p (Listen p a) -getListen (Event getLRaw cacheRef) = do - mL <- ioReactive $ readIORef cacheRef - case mL of - Just l -> return l - Nothing -> do - l <- getLRaw - ioReactive $ writeIORef cacheRef (Just l) - return l - --- | Listen for firings of this event. The returned @IO ()@ is an IO action --- that unregisters the listener. This is the observer pattern. -linkedListen :: Event p a -> Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) -linkedListen ev mMvTarget handle = do - l <- getListen ev - runListen l mMvTarget handle - --- | Variant of 'listenIO' that allows you to initiate more activity in the current --- transaction. Useful for implementing new primitives. -listen :: Event p a -> (a -> Reactive p ()) -> Reactive p (IO ()) -listen ev handle = linkedListen ev Nothing handle - --- | Listen for firings of this event. The returned @IO ()@ is an IO action --- that unregisters the listener. This is the observer pattern. -listenIO :: Event p a -> (a -> IO ()) -> Reactive p (IO ()) -listenIO ev handle = listen ev (ioReactive . handle) - -data Observer p a = Observer { - obNextID :: ID, - obListeners :: Map ID (a -> Reactive p ()), - obFirings :: [a] - } - -data Node p = Node { - noID :: NodeID, - noSerial :: Int64, - noListeners :: Map ID (MVar (Node p)) - } - -newNode :: forall p . Typeable p => IO (MVar (Node p)) -newNode = do - part <- partition :: IO (Partition p) - nodeID <- readIORef (paNextNodeID part) - modifyIORef (paNextNodeID part) succ - newMVar (Node nodeID 0 M.empty) - -wrap :: (Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ())) -> IO (Listen p a) -{-# NOINLINE wrap #-} -wrap l = return (Listen l) - -touch :: Listen p a -> IO () -{-# NOINLINE touch #-} -touch l = evaluate l >> return () - -linkNode :: MVar (Node p) -> ID -> MVar (Node p) -> IO () -linkNode mvNode iD mvTarget = do - no <- readMVar mvNode - ensureBiggerThan S.empty mvTarget (noSerial no) - modifyMVar_ mvNode $ \no -> return $ - no { noListeners = M.insert iD mvTarget (noListeners no) } - -ensureBiggerThan :: Set NodeID -> MVar (Node p) -> Int64 -> IO () -ensureBiggerThan visited mvNode limit = do - no <- readMVar mvNode - if noID no `S.member` visited || noSerial no > limit then - return () - else do - let newSerial = succ limit - --putStrLn $ show (noSerial no) ++ " -> " ++ show newSerial - modifyMVar_ mvNode $ \no -> return $ no { noSerial = newSerial } - forM_ (M.elems . noListeners $ no) $ \mvTarget -> do - ensureBiggerThan (S.insert (noID no) visited) mvTarget newSerial - -unlinkNode :: MVar (Node p) -> ID -> IO () -unlinkNode mvNode iD = do - modifyMVar_ mvNode $ \no -> return $ - no { noListeners = M.delete iD (noListeners no) } - --- | Returns a 'Listen' for registering listeners, and a push action for pushing --- a value into the event. -newSink :: forall p a . Typeable p => IO (Listen p a, a -> Reactive p (), MVar (Node p)) -newSink = do - mvNode <- newNode - mvObs <- newMVar (Observer 0 M.empty []) - cacheRef <- newIORef Nothing - rec - let l mMvTarget handle = do - (firings, unlisten, iD) <- ioReactive $ modifyMVar mvObs $ \ob -> return $ - let iD = obNextID ob - handle' a = handle a >> ioReactive (touch listen) - ob' = ob { obNextID = succ iD, - obListeners = M.insert iD handle' (obListeners ob) } - unlisten = do - modifyMVar_ mvObs $ \ob -> return $ ob { - obListeners = M.delete iD (obListeners ob) - } - unlinkNode mvNode iD - return () - in (ob', (reverse . obFirings $ ob, unlisten, iD)) - case mMvTarget of - Just mvTarget -> ioReactive $ linkNode mvNode iD mvTarget - Nothing -> return () - mapM_ handle firings - return unlisten - listen <- wrap l -- defeat optimizer on ghc-7.0.4 - let push a = do - ob <- ioReactive $ modifyMVar mvObs $ \ob -> return $ - (ob { obFirings = a : obFirings ob }, ob) - -- If this is the first firing... - when (null (obFirings ob)) $ onFinal $ do - modifyMVar_ mvObs $ \ob -> return $ ob { obFirings = [] } - let seqa = seq a a - mapM_ ($ seqa) (M.elems . obListeners $ ob) - return (listen, push, mvNode) - --- | Returns an event, and a push action for pushing a value into the event. -newEventLinked :: Typeable p => IO (Event p a, a -> Reactive p (), MVar (Node p)) -newEventLinked = do - (listen, push, mvNode) <- newSink - cacheRef <- newIORef Nothing - let ev = Event { - getListenRaw = return listen, - evCacheRef = cacheRef - } - return (ev, push, mvNode) - --- | Returns an event, and a push action for pushing a value into the event. -newEvent :: Typeable p => IO (Event p a, a -> Reactive p ()) -newEvent = do - (ev, push, _) <- newEventLinked - return (ev, push) - -instance Functor (Event p) where - f `fmap` Event getListen cacheRef = Event getListen' cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - getListen' = do - return $ Listen $ \mMvNode handle -> do - l <- getListen - runListen l mMvNode (handle . f) - --- | Merge two streams of events of the same type. --- --- In the case where two event occurrences are simultaneous (i.e. both --- within the same transaction), both will be delivered in the same --- transaction. --- --- The order is not defined, because simultaneous events should be considered --- to be order-agnostic. -merge :: Typeable p => Event p a -> Event p a -> Event p a -merge ea eb = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - l1 <- getListen ea - l2 <- getListen eb - (l, push, mvNode) <- ioReactive newSink - unlistener1 <- unlistenize $ runListen l1 (Just mvNode) push - unlistener2 <- unlistenize $ runListen l2 (Just mvNode) push - (finalerize unlistener1 <=< finalerize unlistener2) l - --- | Merge two streams of events of the same type, combining simultaneous --- event occurrences. --- --- In the case where multiple event occurrences are simultaneous (i.e. all --- within the same transaction), they are combined using the supplied --- function. The output event is guaranteed not to have more than one --- event occurrence per transaction. --- --- The combine function should be commutative, because simultaneous events --- should be considered to be order-agnostic. -mergeWith :: Typeable p => (a -> a -> a) -> Event p a -> Event p a -> Event p a -mergeWith f ea eb = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - l1 <- getListen ea - l2 <- getListen eb - (l, push, mvNode) <- ioReactive newSink - outRef <- ioReactive $ newIORef Nothing - let process a = do - mOut <- ioReactive $ readIORef outRef - ioReactive $ modifyIORef outRef $ \mOut -> Just $ case mOut of - Just out -> f out a - Nothing -> a - when (isNothing mOut) $ schedulePriority2 (Just mvNode) $ do - Just out <- ioReactive $ readIORef outRef - ioReactive $ writeIORef outRef Nothing - push out - unlistener1 <- unlistenize $ runListen l1 (Just mvNode) process - unlistener2 <- unlistenize $ runListen l2 (Just mvNode) process - (finalerize unlistener1 <=< finalerize unlistener2) l - --- | Unwrap Just values, and discard event occurrences with Nothing values. -justE :: Typeable p => Event p (Maybe a) -> Event p a -justE ema = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - (l', push, mvNode) <- ioReactive newSink - l <- getListen ema - unlistener <- unlistenize $ runListen l (Just mvNode) $ \ma -> case ma of - Just a -> push a - Nothing -> return () - finalerize unlistener l' - --- | Only keep event occurrences for which the predicate is true. -filterE :: Typeable p => (a -> Bool) -> Event p a -> Event p a -filterE pred = justE . ((\a -> if pred a then Just a else Nothing) <$>) - --- | A time-varying value, American spelling. -type Behavior p a = Behaviour p a - --- | A time-varying value, British spelling. -data Behaviour p a = Behaviour { - -- | Internal: Extract the underlyingEvent event for this behaviour. - underlyingEvent :: Event p a, - -- | Obtain the current value of a behaviour. - sample :: Reactive p a - } - -instance Functor (Behaviour p) where - f `fmap` Behaviour underlyingEvent sample = - Behaviour (f `fmap` underlyingEvent) (f `fmap` sample) - -constant :: a -> Behaviour p a -constant a = Behaviour { - underlyingEvent = never, - sample = return a - } - -data BehaviourState p a = BehaviourState { - bsCurrent :: a, - bsUpdate :: Maybe a - } - --- | Add a finalizer to an event. -finalizeEvent :: Event p a -> IO () -> Event p a -{-# NOINLINE finalizeEvent #-} -finalizeEvent ea unlisten = Event gl (evCacheRef ea) - where - gl = do - l <- getListen ea - ioReactive $ finalizeListen l unlisten - --- | Add a finalizer to a listener. -finalizeListen :: Listen p a -> IO () -> IO (Listen p a) -{-# NOINLINE finalizeListen #-} -finalizeListen l unlisten = do - addFinalizer l unlisten - return l - -newtype Unlistener = Unlistener (MVar (Maybe (IO ()))) - --- | Listen to an input event/behaviour and return an 'Unlistener' that can be --- attached to an output event using 'finalerize'. -unlistenize :: Reactive p (IO ()) -> Reactive p Unlistener -unlistenize doListen = do - unlistener@(Unlistener ref) <- newUnlistener - schedulePriority1 $ do - mOldUnlisten <- ioReactive $ takeMVar ref - case mOldUnlisten of - Just _ -> do - unlisten <- doListen - ioReactive $ putMVar ref (Just unlisten) - Nothing -> ioReactive $ putMVar ref mOldUnlisten - return unlistener - where - newUnlistener :: Reactive p Unlistener - newUnlistener = Unlistener <$> ioReactive (newMVar (Just $ return ())) - --- | Cause the things listened to with unlistenize to be unlistened when the --- specified listener is not referenced any more. -finalerize :: Unlistener -> Listen p a -> Reactive p (Listen p a) -finalerize (Unlistener ref) l = ioReactive $ finalizeListen l $ do - mUnlisten <- takeMVar ref - fromMaybe (return ()) mUnlisten - putMVar ref Nothing - --- | Create a behaviour with the specified initial value, that gets updated --- by the values coming through the event. The \'current value\' of the behaviour --- is notionally the value as it was 'at the start of the transaction'. --- That is, state updates caused by event firings get processed at the end of --- the transaction. -hold :: a -> Event p a -> Reactive p (Behaviour p a) -hold initA ea = do - bsRef <- ioReactive $ newIORef (BehaviourState initA Nothing) - unlistener <- unlistenize $ listen ea $ \a -> do - bs <- ioReactive $ readIORef bsRef - ioReactive $ writeIORef bsRef $ bs { bsUpdate = Just a } - when (isNothing (bsUpdate bs)) $ onFinal $ do - bs <- readIORef bsRef - let newCurrent = fromJust (bsUpdate bs) - bs' = newCurrent `seq` BehaviourState newCurrent Nothing - evaluate bs' - writeIORef bsRef bs' - let gl = do - l <- getListen ea - finalerize unlistener l - beh = Behaviour { - underlyingEvent = Event gl (evCacheRef ea), - sample = ioReactive $ bsCurrent <$> readIORef bsRef - } - return beh - --- | Sample the behaviour at the time of the event firing. Note that the 'current value' --- of the behaviour that's sampled is the value as at the start of the transaction --- before any state changes of the current transaction are applied through 'hold's. -attachWith :: Typeable p => (a -> b -> c) -> Event p a -> Behaviour p b -> Event p c -attachWith f ea bb = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - (l, push, mvNode) <- ioReactive newSink - unlistener <- unlistenize $ linkedListen ea (Just mvNode) $ \a -> do - b <- sample bb - push (f a b) - finalerize unlistener l - --- | Variant of attachWith defined as /attachWith (,)/ -attach :: Typeable p => Event p a -> Behaviour p b -> Event p (a,b) -attach = attachWith (,) - --- | Variant of attachWith that throws away the event's value and captures the behaviour's. -tag :: Typeable p => Event p a -> Behaviour p b -> Event p b -tag = attachWith (flip const) - --- | Listen to the value of this behaviour with an initial callback giving --- the current value. Can get multiple values per transaction, the last of --- which is considered valid. You would normally want to use 'listenValue', --- which removes the extra unwanted values. -listenValueRaw :: Behaviour p a -> Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) -listenValueRaw ba mMvNode handle = do - a <- sample ba - handle a - linkedListen (underlyingEvent ba) mMvNode handle - --- | Queue the specified atomic to run at the end of the priority 2 queue -schedulePriority2 :: Maybe (MVar (Node p)) - -> Reactive p () - -> Reactive p () -schedulePriority2 mMvNode task = do - mNode <- case mMvNode of - Just mvNode -> Just <$> ioReactive (readMVar mvNode) - Nothing -> pure Nothing - let priority = maybe maxBound noSerial mNode - Reactive $ modify $ \as -> as { - asQueue2 = M.alter (\mOldTask -> Just $ case mOldTask of - Just oldTask -> oldTask >> task - Nothing -> task) priority (asQueue2 as) - } - --- Clean up the listener so it gives only one value per transaction, specifically --- the last one. -tidy :: (Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ())) - -> Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) -tidy listen mMvNode handle = do - aRef <- ioReactive $ newIORef Nothing - listen mMvNode $ \a -> do - ma <- ioReactive $ readIORef aRef - ioReactive $ writeIORef aRef (Just a) - when (isNothing ma) $ schedulePriority2 mMvNode $ do - Just a <- ioReactive $ readIORef aRef - ioReactive $ writeIORef aRef Nothing - handle a - --- | Listen to the value of this behaviour with a guaranteed initial callback --- giving the current value, followed by callbacks for any updates. -linkedListenValue :: Behaviour p a -> Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ()) -linkedListenValue ba = tidy (listenValueRaw ba) - --- | Variant of 'listenValueIO' that allows you to initiate more activity in the current --- transaction. Useful for implementing new primitives. -listenValue :: Behaviour p a -> (a -> Reactive p ()) -> Reactive p (IO ()) -listenValue ba = linkedListenValue ba Nothing - --- | Listen to the value of this behaviour with a guaranteed initial callback --- giving the current value, followed by callbacks for any updates. -listenValueIO :: Behaviour p a -> (a -> IO ()) -> Reactive p (IO ()) -listenValueIO ba handle = listenValue ba (ioReactive . handle) - -eventify :: Typeable p => (Maybe (MVar (Node p)) -> (a -> Reactive p ()) -> Reactive p (IO ())) -> Event p a -eventify listen = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - (l, push, mvNode) <- ioReactive newSink - unlistener <- unlistenize $ listen (Just mvNode) push - finalerize unlistener l - --- | An event that fires once for the current value of the behaviour, and then --- for all changes that occur after that. -valueEvent :: Typeable p => Behaviour p a -> Event p a -valueEvent ba = eventify (linkedListenValue ba) - -instance Typeable p => Applicative (Behaviour p) where - pure = constant - Behaviour u1 s1 <*> Behaviour u2 s2 = Behaviour u s - where - cacheRef = unsafePerformIO $ newIORef Nothing - u = Event gl cacheRef - gl = do - fRef <- ioReactive . newIORef =<< s1 - aRef <- ioReactive . newIORef =<< s2 - l1 <- getListen u1 - l2 <- getListen u2 - (l, push, mvNode) <- ioReactive newSink - unlistener1 <- unlistenize $ runListen l1 (Just mvNode) $ \f -> do - ioReactive $ writeIORef fRef f - a <- ioReactive $ readIORef aRef - push (f a) - unlistener2 <- unlistenize $ runListen l2 (Just mvNode) $ \a -> do - f <- ioReactive $ readIORef fRef - ioReactive $ writeIORef aRef a - push (f a) - (finalerize unlistener1 <=< finalerize unlistener2) l - s = ($) <$> s1 <*> s2 - --- | Let event occurrences through only when the behaviour's value is True. --- Note that the behaviour's value is as it was at the start of the transaction, --- that is, no state changes from the current transaction are taken into account. -gate :: Typeable p => Event p a -> Behaviour p Bool -> Event p a -gate ea = justE . attachWith (\a b -> if b then Just a else Nothing) ea - --- | Transform an event with a generalized state loop (a mealy machine). The function --- is passed the input and the old state and returns the new state and output value. -collectE :: Typeable p => (a -> s -> (b, s)) -> s -> Event p a -> Reactive p (Event p b) -collectE f z ea = do - rec - s <- hold z es - let ebs = attachWith f ea s - eb = fst <$> ebs - es = snd <$> ebs - return eb - --- | Transform a behaviour with a generalized state loop (a mealy machine). The function --- is passed the input and the old state and returns the new state and output value. -collect :: Typeable p => (a -> s -> (b, s)) -> s -> Behaviour p a -> Reactive p (Behaviour p b) -collect f zs bea = do - let ea = eventify . tidy . linkedListen $ underlyingEvent bea - za <- sample bea - let (zb, zs') = f za zs - rec - bs <- hold (zb, zs') ebs - let ebs = attachWith f ea (snd <$> bs) - return (fst <$> bs) - --- | Accumulate on input event, outputting the new state each time. -accumE :: Typeable p => (a -> s -> s) -> s -> Event p a -> Reactive p (Event p s) -accumE f z ea = do - rec - let es = attachWith f ea s - s <- hold z es - return es - --- | Accumulate on input event, holding state. -accum :: Typeable p => (a -> s -> s) -> s -> Event p a -> Reactive p (Behaviour p s) -accum f z ea = do - rec - s <- hold z (attachWith f ea s) - return s - --- | Count event occurrences, starting with 1 for the first occurrence. -countE :: Typeable p => Event p a -> Reactive p (Event p Int) -countE = accumE (+) 0 . (const 1 <$>) - --- | Count event occurrences, giving a behaviour that starts with 0 before the first occurrence. -count :: Typeable p => Event p a -> Reactive p (Behaviour p Int) -count = hold 0 <=< countE - -splitLessThan :: Ord k => k -> Map k a -> (Map k a, Map k a) -splitLessThan k m = - let (lt, mEq, gt) = M.splitLookup k m - in (lt, case mEq of - Just eq -> M.insert k eq gt - Nothing -> gt) - -unlistenLessThan :: IORef (Map ID (IO ())) -> ID -> IO () -unlistenLessThan unlistensRef iD = do - uls <- readIORef unlistensRef - let (toDelete, uls') = splitLessThan iD uls - do - writeIORef unlistensRef uls' - {-when (M.size toDelete > 0) $ - putStrLn $ "deleting "++show (M.size toDelete) -} - forM_ (M.elems toDelete) $ \unl -> unl - --- | Unwrap an event inside a behaviour to give a time-varying event implementation. -switchE :: Typeable p => Behaviour p (Event p a) -> Event p a -switchE bea = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - unlistensRef = unsafePerformIO $ newIORef M.empty - gl = do - -- assign ID numbers to the incoming events - beaId <- collect (\ea nxtID -> ((ea, nxtID), succ nxtID)) (0 :: ID) bea - (l, push, mvNode) <- ioReactive newSink - unlistener1 <- unlistenize $ linkedListenValue beaId (Just mvNode) $ \(ea, iD) -> do - let filtered = justE $ attachWith (\a activeID -> - if activeID == iD - then Just a - else Nothing - ) ea (snd <$> beaId) - unlisten2 <- listen filtered $ \a -> do - push a - ioReactive $ unlistenLessThan unlistensRef iD - ioReactive $ modifyIORef unlistensRef (M.insert iD unlisten2) - finalerize unlistener1 l - --- | Unwrap a behaviour inside another behaviour to give a time-varying behaviour implementation. -switch :: Typeable p => Behaviour p (Behaviour p a) -> Reactive p (Behaviour p a) -switch bba = do - ba <- sample bba - za <- sample ba - (ev, push, mvNode) <- ioReactive newEventLinked - activeIDRef <- ioReactive $ newIORef (0 :: ID) - unlistensRef <- ioReactive $ newIORef M.empty - unlisten1 <- listenValueRaw bba (Just mvNode) $ \ba -> do - iD <- ioReactive $ do - modifyIORef activeIDRef succ - readIORef activeIDRef - unlisten2 <- listenValueRaw ba (Just mvNode) $ \a -> do - activeID <- ioReactive $ readIORef activeIDRef - when (activeID == iD) $ do - push a - ioReactive $ unlistenLessThan unlistensRef iD - ioReactive $ modifyIORef unlistensRef (M.insert iD unlisten2) - hold za (finalizeEvent ev unlisten1) - --- | Throw away all event occurrences except for the first one. -once :: Typeable p => Event p a -> Reactive p (Event p a) -once ea = justE <$> collectE (\a active -> (if active then Just a else Nothing, False)) True ea - --- | Execute the specified 'Reactive' action inside an event. -execute :: Typeable p => Event p (Reactive p a) -> Event p a -execute ev = Event gl cacheRef - where - cacheRef = unsafePerformIO $ newIORef Nothing - gl = do - (l', push, mvNode) <- ioReactive newSink - unlistener <- unlistenize $ do - l <- getListen ev - runListen l (Just mvNode) $ \action -> action >>= push - finalerize unlistener l' - --- | Cross the specified event over to a different partition. -crossE :: (Typeable p, Typeable q) => Event p a -> Reactive p (Event q a) -crossE epa = do - (ev, push) <- ioReactive newEvent - unlisten <- listenIO epa $ asynchronously . push - return $ finalizeEvent ev unlisten - --- | Cross the specified behaviour over to a different partition. -cross :: (Typeable p, Typeable q) => Behaviour p a -> Reactive p (Behaviour q a) -cross bpa = do - a <- sample bpa - ea <- crossE (underlyingEvent bpa) - ioReactive $ synchronously $ hold a ea - diff --git a/src/FRP/Sodium/Internal.hs b/src/FRP/Sodium/Internal.hs --- a/src/FRP/Sodium/Internal.hs +++ b/src/FRP/Sodium/Internal.hs @@ -1,26 +1,24 @@ {-# LANGUAGE GeneralizedNewtypeDeriving, ScopedTypeVariables, DoRec #-} {-# OPTIONS_GHC -fno-cse -fno-full-laziness #-} module FRP.Sodium.Internal ( - listen, - listenValue, - Event(..), - Behaviour(..), - schedulePriority1, - schedulePriority2, + listenTrans, + listenValueTrans, + schedulePrioritized, + scheduleLast, Listen(..), getListen, runListen, linkedListen, Node, newEventLinked, - newSink, + newEvent, finalizeEvent, finalizeListen, ioReactive, Unlistener, - finalerize, + addCleanup, unlistenize ) where -import FRP.Sodium.Impl +import FRP.Sodium.Plain diff --git a/src/FRP/Sodium/Plain.hs b/src/FRP/Sodium/Plain.hs new file mode 100644 --- /dev/null +++ b/src/FRP/Sodium/Plain.hs @@ -0,0 +1,730 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving, ScopedTypeVariables, DoRec, GADTs, + TypeFamilies, EmptyDataDecls, FlexibleInstances #-} +{-# OPTIONS_GHC -fno-cse -fno-full-laziness #-} +module FRP.Sodium.Plain where + +import qualified FRP.Sodium.Context as R + +-- Note: the 'full-laziness' optimization messes up finalizers, so we're +-- disabling it. It'd be nice to find a really robust solution to this. +-- -fno-cse just in case, since we're using unsafePerformIO. + +import Control.Applicative +import Control.Concurrent +import Control.Concurrent.Chan +import Control.Concurrent.MVar +import Control.Exception (evaluate) +import Control.Monad +import Control.Monad.State.Strict +import Control.Monad.Trans +import Data.Int +import Data.IORef +import Data.Map (Map) +import qualified Data.Map as M +import Data.Maybe +import Data.Set (Set) +import qualified Data.Set as S +import Data.Sequence (Seq, (|>)) +import qualified Data.Sequence as Seq +import GHC.Exts +import System.Mem.Weak +import System.IO.Unsafe + +-- | Phantom type for use with 'R.Context' type class. +data Plain + +partition :: Partition +{-# NOINLINE partition #-} +partition = unsafePerformIO createPartition + where + createPartition :: IO Partition + createPartition = do + lock <- newEmptyMVar + nextNodeIDRef <- newIORef (NodeID 0) + return $ Partition { + paLock = lock, + paNextNodeID = nextNodeIDRef + } + +-- | A monad for transactional reactive operations. Execute it from 'IO' using 'sync'. +type Reactive a = R.Reactive Plain a + +-- | A stream of events. The individual firings of events are called \'event occurrences\'. +type Event a = R.Event Plain a + +-- | A time-varying value, American spelling. +type Behavior a = R.Behavior Plain a + +-- | A time-varying value, British spelling. +type Behaviour a = R.Behavior Plain a + +instance R.Context Plain where + + data Reactive Plain a = Reactive (StateT ReactiveState IO a) + + data Event Plain a = Event { -- Must be data not newtype, because we need to attach finalizers to it + -- | Listen for event occurrences on this event, to be handled by the specified + -- handler. The returned action is used to unregister the listener. + getListenRaw :: Reactive (Listen a), + evCacheRef :: IORef (Maybe (Listen a)) + } + + data Behavior Plain a = Behavior { + -- | Internal: Extract the underlyingEvent event for this behaviour. + underlyingEvent :: Event a, + -- | Obtain the current value of a behaviour. + behSample :: Reactive a + } + sync = sync + ioReactive = ioReactive + newEvent = newEvent + listen = listen + never = never + merge = merge + filterJust = filterJust + hold = hold + changes = changes + values = values + snapshotWith = snapshotWith + switchE = switchE + switch = switch + execute = execute + sample = sample + coalesce = coalesce + +-- | Execute the specified 'Reactive' within a new transaction, blocking the caller +-- until all resulting processing is complete and all callbacks have been called. +-- This operation is thread-safe, so it may be called from any thread. +-- +-- State changes to 'hold' values occur after processing of the transaction is complete. +sync :: Reactive a -> IO a +sync task = do + let loop :: StateT ReactiveState IO () = do + queue1 <- gets asQueue1 + if not $ Seq.null queue1 then do + let Reactive task = Seq.index queue1 0 + modify $ \as -> as { asQueue1 = Seq.drop 1 queue1 } + task + loop + else do + queue2 <- gets asQueue2 + if not $ M.null queue2 then do + let (k, Reactive task) = M.findMin queue2 + modify $ \as -> as { asQueue2 = M.delete k queue2 } + task + loop + else do + final <- gets asFinal + liftIO final + return () + outVar <- newIORef undefined + let lock = paLock partition + putMVar lock () + evalStateT loop $ ReactiveState { + asQueue1 = Seq.singleton (task >>= ioReactive . writeIORef outVar), + asQueue2 = M.empty, + asFinal = return () + } + takeMVar lock + readIORef outVar + +-- | Returns an event, and a push action for pushing a value into the event. +newEvent :: Reactive (Event a, a -> Reactive ()) +newEvent = do + (ev, push, _) <- ioReactive newEventLinked + return (ev, push) + +-- | Listen for firings of this event. The returned @IO ()@ is an IO action +-- that unregisters the listener. This is the observer pattern. +listen :: Event a -> (a -> IO ()) -> Reactive (IO ()) +listen ev handle = listenTrans ev (ioReactive . handle) + +-- | An event that never fires. +never :: Event a +never = Event { + getListenRaw = return $ Listen $ \_ _ -> return (return ()), + evCacheRef = unsafePerformIO $ newIORef Nothing + } + +-- | Merge two streams of events of the same type. +-- +-- In the case where two event occurrences are simultaneous (i.e. both +-- within the same transaction), both will be delivered in the same +-- transaction. +-- +-- The order is not defined, because simultaneous events should be considered +-- to be order-agnostic. +merge :: Event a -> Event a -> Event a +merge ea eb = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + l1 <- getListen ea + l2 <- getListen eb + (l, push, nodeRef) <- ioReactive newEventImpl + unlistener1 <- unlistenize $ runListen l1 (Just nodeRef) push + unlistener2 <- unlistenize $ runListen l2 (Just nodeRef) push + (addCleanup unlistener1 <=< addCleanup unlistener2) l + +-- | Unwrap Just values, and discard event occurrences with Nothing values. +filterJust :: Event (Maybe a) -> Event a +filterJust ema = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + (l', push, nodeRef) <- ioReactive newEventImpl + l <- getListen ema + unlistener <- unlistenize $ runListen l (Just nodeRef) $ \ma -> case ma of + Just a -> push a + Nothing -> return () + addCleanup unlistener l' + +-- | Create a behaviour with the specified initial value, that gets updated +-- by the values coming through the event. The \'current value\' of the behaviour +-- is notionally the value as it was 'at the start of the transaction'. +-- That is, state updates caused by event firings get processed at the end of +-- the transaction. +hold :: a -> Event a -> Reactive (Behavior a) +hold initA ea = do + bsRef <- ioReactive $ newIORef (BehaviorState initA Nothing) + unlistener <- unlistenize $ listenTrans ea $ \a -> do + bs <- ioReactive $ readIORef bsRef + ioReactive $ writeIORef bsRef $ bs { bsUpdate = Just a } + when (isNothing (bsUpdate bs)) $ onFinal $ do + bs <- readIORef bsRef + let newCurrent = fromJust (bsUpdate bs) + bs' = newCurrent `seq` BehaviorState newCurrent Nothing + evaluate bs' + writeIORef bsRef bs' + let gl = do + l <- getListen ea + addCleanup unlistener l + beh = Behavior { + underlyingEvent = Event gl (evCacheRef ea), + behSample = ioReactive $ bsCurrent <$> readIORef bsRef + } + return beh + +-- | An event that gives the updates for the behaviour. It doesn't do any equality +-- comparison as the name might imply. +changes :: Behavior a -> Event a +changes = underlyingEvent + +-- | An event that is guaranteed to fires once when you listen to it, giving +-- the current value of the behaviour, and thereafter behaves like 'changes', +-- firing for each update to the behaviour's value. +values :: Behavior a -> Event a +values = eventify . linkedListenValue + +-- | Sample the behaviour at the time of the event firing. Note that the 'current value' +-- of the behaviour that's sampled is the value as at the start of the transaction +-- before any state changes of the current transaction are applied through 'hold's. +snapshotWith :: (a -> b -> c) -> Event a -> Behavior b -> Event c +snapshotWith f ea bb = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + (l, push, nodeRef) <- ioReactive newEventImpl + unlistener <- unlistenize $ linkedListen ea (Just nodeRef) $ \a -> do + b <- sample bb + push (f a b) + addCleanup unlistener l + +-- | Unwrap an event inside a behaviour to give a time-varying event implementation. +switchE :: Behavior (Event a) -> Event a +switchE bea = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + unlistensRef = unsafePerformIO $ newIORef M.empty + gl = do + -- assign ID numbers to the incoming events + beaId <- R.collect (\ea nxtID -> ((ea, nxtID), succ nxtID)) (0 :: ID) bea + (l, push, nodeRef) <- ioReactive newEventImpl + unlistener1 <- unlistenize $ linkedListenValue beaId (Just nodeRef) $ \(ea, iD) -> do + let filtered = filterJust $ snapshotWith (\a activeID -> + if activeID == iD + then Just a + else Nothing + ) ea (snd <$> beaId) + unlisten2 <- listenTrans filtered $ \a -> do + push a + ioReactive $ unlistenLessThan unlistensRef iD + ioReactive $ modifyIORef unlistensRef (M.insert iD unlisten2) + addCleanup unlistener1 l + +-- | Unwrap a behaviour inside another behaviour to give a time-varying behaviour implementation. +switch :: Behavior (Behavior a) -> Reactive (Behavior a) +switch bba = do + ba <- sample bba + za <- sample ba + (ev, push, nodeRef) <- ioReactive newEventLinked + activeIDRef <- ioReactive $ newIORef (0 :: ID) + unlistensRef <- ioReactive $ newIORef M.empty + unlisten1 <- listenValueRaw bba (Just nodeRef) $ \ba -> do + iD <- ioReactive $ do + modifyIORef activeIDRef succ + readIORef activeIDRef + unlisten2 <- listenValueRaw ba (Just nodeRef) $ \a -> do + activeID <- ioReactive $ readIORef activeIDRef + when (activeID == iD) $ do + push a + ioReactive $ unlistenLessThan unlistensRef iD + ioReactive $ modifyIORef unlistensRef (M.insert iD unlisten2) + hold za (finalizeEvent ev unlisten1) + +-- | Execute the specified 'Reactive' action inside an event. +execute :: Event (Reactive a) -> Event a +execute ev = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + (l', push, nodeRef) <- ioReactive newEventImpl + unlistener <- unlistenize $ do + l <- getListen ev + runListen l (Just nodeRef) $ \action -> action >>= push + addCleanup unlistener l' + +-- | Obtain the current value of a behaviour. +sample :: Behavior a -> Reactive a +sample = behSample + +-- | If there's more than one firing in a single transaction, combine them into +-- one using the specified combining function. +coalesce :: (a -> a -> a) -> Event a -> Event a +coalesce combine e = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + l1 <- getListen e + (l, push, nodeRef) <- ioReactive newEventImpl + outRef <- ioReactive $ newIORef Nothing + unlistener <- unlistenize $ runListen l1 (Just nodeRef) $ \a -> do + first <- isNothing <$> ioReactive (readIORef outRef) + ioReactive $ modifyIORef outRef $ \ma -> Just $ case ma of + Just a0 -> a0 `combine` a + Nothing -> a + when first $ scheduleLast (Just nodeRef) $ do + Just out <- ioReactive $ readIORef outRef + ioReactive $ writeIORef outRef Nothing + push out + addCleanup unlistener l + +newBehavior :: a -- ^ Initial behaviour value + -> Reactive (Behavior a, a -> Reactive ()) +newBehavior = R.newBehavior + +listenValue :: Behavior a -> (a -> IO ()) -> Reactive (IO ()) +listenValue = R.listenValue + +-- | Merge two streams of events of the same type, combining simultaneous +-- event occurrences. +-- +-- In the case where multiple event occurrences are simultaneous (i.e. all +-- within the same transaction), they are combined using the supplied +-- function. The output event is guaranteed not to have more than one +-- event occurrence per transaction. +-- +-- The combine function should be commutative, because simultaneous events +-- should be considered to be order-agnostic. +mergeWith :: (a -> a -> a) -> Event a -> Event a -> Event a +mergeWith = R.mergeWith + +-- | Only keep event occurrences for which the predicate is true. +filterE :: (a -> Bool) -> Event a -> Event a +filterE = R.filterE + +-- | Variant of 'snapshotWith' that throws away the event's value and captures the behaviour's. +snapshot :: Event a -> Behavior b -> Event b +snapshot = R.snapshot + +-- | Let event occurrences through only when the behaviour's value is True. +-- Note that the behaviour's value is as it was at the start of the transaction, +-- that is, no state changes from the current transaction are taken into account. +gate :: Event a -> Behavior Bool -> Event a +gate = R.gate + +-- | Transform an event with a generalized state loop (a mealy machine). The function +-- is passed the input and the old state and returns the new state and output value. +collectE :: (a -> s -> (b, s)) -> s -> Event a -> Reactive (Event b) +collectE = R.collectE + +-- | Transform a behaviour with a generalized state loop (a mealy machine). The function +-- is passed the input and the old state and returns the new state and output value. +collect :: (a -> s -> (b, s)) -> s -> Behavior a -> Reactive (Behavior b) +collect = R.collect + +-- | Accumulate on input event, outputting the new state each time. +accumE :: (a -> s -> s) -> s -> Event a -> Reactive (Event s) +accumE = R.accumE + +-- | Accumulate on input event, holding state. +accum :: (a -> s -> s) -> s -> Event a -> Reactive (Behavior s) +accum = R.accum + +-- | Count event occurrences, starting with 1 for the first occurrence. +countE :: Event a -> Reactive (Event Int) +countE = R.countE + +-- | Count event occurrences, giving a behaviour that starts with 0 before the first occurrence. +count :: Event a -> Reactive (Behavior Int) +count = R.count + +-- | Throw away all event occurrences except for the first one. +once :: Event a -> Reactive (Event a) +once = R.once + +type ID = Int64 + +data ReactiveState = ReactiveState { + asQueue1 :: Seq (Reactive ()), + asQueue2 :: Map Int64 (Reactive ()), + asFinal :: IO () + } + +instance Functor (R.Reactive Plain) where + fmap f rm = Reactive (fmap f (unReactive rm)) + +unReactive :: Reactive a -> StateT ReactiveState IO a +unReactive (Reactive m) = m + +instance Applicative (R.Reactive Plain) where + pure a = Reactive $ return a + rf <*> rm = Reactive $ unReactive rf <*> unReactive rm + +instance Monad (R.Reactive Plain) where + return a = Reactive $ return a + rma >>= kmb = Reactive $ do + a <- unReactive rma + unReactive (kmb a) + +instance MonadFix (R.Reactive Plain) where + mfix f = Reactive $ mfix $ \a -> unReactive (f a) + +ioReactive :: IO a -> Reactive a +ioReactive io = Reactive $ liftIO io + +newtype NodeID = NodeID Int deriving (Eq, Ord, Enum) + +data Partition = Partition { + paLock :: MVar (), + paNextNodeID :: IORef NodeID + } + +-- | Queue the specified atomic to run at the end of the priority 1 queue +schedulePrioritized :: Reactive () -> Reactive () +schedulePrioritized task = Reactive $ modify $ \as -> as { asQueue1 = asQueue1 as |> task } + +onFinal :: IO () -> Reactive () +onFinal task = Reactive $ modify $ \as -> as { asFinal = asFinal as >> task } + +data Listen a = Listen { runListen_ :: Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) } + +runListen :: Listen a -> Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) +{-# NOINLINE runListen #-} +runListen l mv handle = do + o <- runListen_ l mv handle + _ <- ioReactive $ evaluate l + return o + +-- | Unwrap an event's listener machinery. +getListen :: Event a -> Reactive (Listen a) +getListen (Event getLRaw cacheRef) = do + mL <- ioReactive $ readIORef cacheRef + case mL of + Just l -> return l + Nothing -> do + l <- getLRaw + ioReactive $ writeIORef cacheRef (Just l) + return l + +-- | Listen for firings of this event. The returned @IO ()@ is an IO action +-- that unregisters the listener. This is the observer pattern. +linkedListen :: Event a -> Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) +linkedListen ev mMvTarget handle = do + l <- getListen ev + runListen l mMvTarget handle + +-- | Variant of 'listen' that allows you to initiate more activity in the current +-- transaction. Useful for implementing new primitives. +listenTrans :: Event a -> (a -> Reactive ()) -> Reactive (IO ()) +listenTrans ev handle = linkedListen ev Nothing handle + +data Observer p a = Observer { + obNextID :: ID, + obListeners :: Map ID (a -> Reactive ()), + obFirings :: [a] + } + +data Node = Node { + noID :: NodeID, + noRank :: Int64, + noListeners :: Map ID (IORef Node) + } + +newNode :: IO (IORef Node) +newNode = do + nodeID <- readIORef (paNextNodeID partition) + modifyIORef (paNextNodeID partition) succ + newIORef (Node nodeID 0 M.empty) + +wrap :: (Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ())) -> IO (Listen a) +{-# NOINLINE wrap #-} +wrap l = return (Listen l) + +touch :: Listen a -> IO () +{-# NOINLINE touch #-} +touch l = evaluate l >> return () + +linkNode :: IORef Node -> ID -> IORef Node -> IO () +linkNode nodeRef iD mvTarget = do + no <- readIORef nodeRef + ensureBiggerThan S.empty mvTarget (noRank no) + modifyIORef nodeRef $ \no -> + no { noListeners = M.insert iD mvTarget (noListeners no) } + +ensureBiggerThan :: Set NodeID -> IORef Node -> Int64 -> IO () +ensureBiggerThan visited nodeRef limit = do + no <- readIORef nodeRef + if noRank no > limit || noID no `S.member` visited then + return () + else do + let newSerial = succ limit + --putStrLn $ show (noRank no) ++ " -> " ++ show newSerial + modifyIORef nodeRef $ \no -> no { noRank = newSerial } + forM_ (M.elems . noListeners $ no) $ \mvTarget -> do + ensureBiggerThan (S.insert (noID no) visited) mvTarget newSerial + +unlinkNode :: IORef Node -> ID -> IO () +unlinkNode nodeRef iD = do + modifyIORef nodeRef $ \no -> + no { noListeners = M.delete iD (noListeners no) } + +-- | Returns a 'Listen' for registering listeners, and a push action for pushing +-- a value into the event. +newEventImpl :: forall p a . IO (Listen a, a -> Reactive (), IORef Node) +newEventImpl = do + nodeRef <- newNode + mvObs <- newMVar (Observer 0 M.empty []) + cacheRef <- newIORef Nothing + rec + let l mMvTarget handle = do + (firings, unlisten, iD) <- ioReactive $ modifyMVar mvObs $ \ob -> return $ + let iD = obNextID ob + handle' a = handle a >> ioReactive (touch listen) + ob' = ob { obNextID = succ iD, + obListeners = M.insert iD handle' (obListeners ob) } + unlisten = do + modifyMVar_ mvObs $ \ob -> return $ ob { + obListeners = M.delete iD (obListeners ob) + } + unlinkNode nodeRef iD + return () + in (ob', (reverse . obFirings $ ob, unlisten, iD)) + case mMvTarget of + Just mvTarget -> ioReactive $ linkNode nodeRef iD mvTarget + Nothing -> return () + mapM_ handle firings + return unlisten + listen <- wrap l -- defeat optimizer on ghc-7.0.4 + let push a = do + ob <- ioReactive $ modifyMVar mvObs $ \ob -> return $ + (ob { obFirings = a : obFirings ob }, ob) + -- If this is the first firing... + when (null (obFirings ob)) $ onFinal $ do + modifyMVar_ mvObs $ \ob -> return $ ob { obFirings = [] } + let seqa = seq a a + mapM_ ($ seqa) (M.elems . obListeners $ ob) + return (listen, push, nodeRef) + +-- | Returns an event, and a push action for pushing a value into the event. +newEventLinked :: IO (Event a, a -> Reactive (), IORef Node) +newEventLinked = do + (listen, push, nodeRef) <- newEventImpl + cacheRef <- newIORef Nothing + let ev = Event { + getListenRaw = return listen, + evCacheRef = cacheRef + } + return (ev, push, nodeRef) + +instance Functor (R.Event Plain) where + f `fmap` Event getListen cacheRef = Event getListen' cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + getListen' = do + return $ Listen $ \mNodeRef handle -> do + l <- getListen + runListen l mNodeRef (handle . f) + +instance Functor (R.Behavior Plain) where + f `fmap` Behavior underlyingEvent sample = + Behavior (f `fmap` underlyingEvent) (f `fmap` sample) + +constant :: a -> Behavior a +constant a = Behavior { + underlyingEvent = never, + behSample = return a + } + +data BehaviorState a = BehaviorState { + bsCurrent :: a, + bsUpdate :: Maybe a + } + +-- | Add a finalizer to an event. +finalizeEvent :: Event a -> IO () -> Event a +{-# NOINLINE finalizeEvent #-} +finalizeEvent ea unlisten = Event gl (evCacheRef ea) + where + gl = do + l <- getListen ea + ioReactive $ finalizeListen l unlisten + +-- | Add a finalizer to a listener. +finalizeListen :: Listen a -> IO () -> IO (Listen a) +{-# NOINLINE finalizeListen #-} +finalizeListen l unlisten = do + addFinalizer l unlisten + return l + +newtype Unlistener = Unlistener (MVar (Maybe (IO ()))) + +-- | Listen to an input event/behaviour and return an 'Unlistener' that can be +-- attached to an output event using 'addCleanup'. +unlistenize :: Reactive (IO ()) -> Reactive Unlistener +unlistenize doListen = do + unlistener@(Unlistener ref) <- newUnlistener + schedulePrioritized $ do + mOldUnlisten <- ioReactive $ takeMVar ref + case mOldUnlisten of + Just _ -> do + unlisten <- doListen + ioReactive $ putMVar ref (Just unlisten) + Nothing -> ioReactive $ putMVar ref mOldUnlisten + return unlistener + where + newUnlistener :: Reactive Unlistener + newUnlistener = Unlistener <$> ioReactive (newMVar (Just $ return ())) + +-- | Cause the things listened to with unlistenize to be unlistened when the +-- specified listener is not referenced any more. +addCleanup :: Unlistener -> Listen a -> Reactive (Listen a) +addCleanup (Unlistener ref) l = ioReactive $ finalizeListen l $ do + mUnlisten <- takeMVar ref + fromMaybe (return ()) mUnlisten + putMVar ref Nothing + +-- | Listen to the value of this behaviour with an initial callback giving +-- the current value. Can get multiple values per transaction, the last of +-- which is considered valid. You would normally want to use 'listenValue', +-- which removes the extra unwanted values. +listenValueRaw :: Behavior a -> Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) +listenValueRaw ba mNodeRef handle = do + a <- sample ba + handle a + linkedListen (underlyingEvent ba) mNodeRef handle + +-- | Queue the specified atomic to run at the end of the priority 2 queue +scheduleLast :: Maybe (IORef Node) + -> Reactive () + -> Reactive () +scheduleLast mNodeRef task = do + mNode <- case mNodeRef of + Just nodeRef -> Just <$> ioReactive (readIORef nodeRef) + Nothing -> pure Nothing + let priority = maybe maxBound noRank mNode + Reactive $ modify $ \as -> as { + asQueue2 = M.alter (\mOldTask -> Just $ case mOldTask of + Just oldTask -> oldTask >> task + Nothing -> task) priority (asQueue2 as) + } + +-- Clean up the listener so it gives only one value per transaction, specifically +-- the last one. +tidy :: (Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ())) + -> Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) +tidy listen mNodeRef handle = do + aRef <- ioReactive $ newIORef Nothing + listen mNodeRef $ \a -> do + ma <- ioReactive $ readIORef aRef + ioReactive $ writeIORef aRef (Just a) + when (isNothing ma) $ scheduleLast mNodeRef $ do + Just a <- ioReactive $ readIORef aRef + ioReactive $ writeIORef aRef Nothing + handle a + +-- | Listen to the value of this behaviour with a guaranteed initial callback +-- giving the current value, followed by callbacks for any updates. +linkedListenValue :: Behavior a -> Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ()) +linkedListenValue ba = tidy (listenValueRaw ba) + +-- | Variant of 'listenValue' that allows you to initiate more activity in the current +-- transaction. Useful for implementing new primitives. +listenValueTrans :: Behavior a -> (a -> Reactive ()) -> Reactive (IO ()) +listenValueTrans ba = linkedListenValue ba Nothing + +eventify :: (Maybe (IORef Node) -> (a -> Reactive ()) -> Reactive (IO ())) -> Event a +eventify listen = Event gl cacheRef + where + cacheRef = unsafePerformIO $ newIORef Nothing + gl = do + (l, push, nodeRef) <- ioReactive newEventImpl + unlistener <- unlistenize $ listen (Just nodeRef) push + addCleanup unlistener l + +instance Applicative (R.Behavior Plain) where + pure = constant + Behavior u1 s1 <*> Behavior u2 s2 = Behavior u s + where + cacheRef = unsafePerformIO $ newIORef Nothing + u = Event gl cacheRef + gl = do + fRef <- ioReactive . newIORef =<< s1 + aRef <- ioReactive . newIORef =<< s2 + l1 <- getListen u1 + l2 <- getListen u2 + (l, push, nodeRef) <- ioReactive newEventImpl + unlistener1 <- unlistenize $ runListen l1 (Just nodeRef) $ \f -> do + ioReactive $ writeIORef fRef f + a <- ioReactive $ readIORef aRef + push (f a) + unlistener2 <- unlistenize $ runListen l2 (Just nodeRef) $ \a -> do + f <- ioReactive $ readIORef fRef + ioReactive $ writeIORef aRef a + push (f a) + (addCleanup unlistener1 <=< addCleanup unlistener2) l + s = ($) <$> s1 <*> s2 + +splitLessThan :: Ord k => k -> Map k a -> (Map k a, Map k a) +splitLessThan k m = + let (lt, mEq, gt) = M.splitLookup k m + in (lt, case mEq of + Just eq -> M.insert k eq gt + Nothing -> gt) + +unlistenLessThan :: IORef (Map ID (IO ())) -> ID -> IO () +unlistenLessThan unlistensRef iD = do + uls <- readIORef unlistensRef + let (toDelete, uls') = splitLessThan iD uls + do + writeIORef unlistensRef uls' + {-when (M.size toDelete > 0) $ + putStrLn $ "deleting "++show (M.size toDelete) -} + forM_ (M.elems toDelete) $ \unl -> unl + +{- +-- | Cross the specified event over to a different partition. +crossE :: (Typeable p, Typeable q) => Event a -> Reactive (Event q a) +crossE epa = do + (ev, push) <- ioReactive newEvent + unlisten <- listen epa $ async . push + return $ finalizeEvent ev unlisten + +-- | Cross the specified behaviour over to a different partition. +cross :: (Typeable p, Typeable q) => Behavior a -> Reactive (Behavior q a) +cross bpa = do + a <- sample bpa + ea <- crossE (underlyingEvent bpa) + ioReactive $ sync $ hold a ea +-} +