chan 0.0.3 → 0.0.4

raw patch · 8 files changed

+263/−483 lines, 8 filesdep ~basePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• chan.cabal +54/−39
• src/Control/Concurrent/Chan/Extra.hs +207/−78
• src/Control/Concurrent/Chan/Scope.hs +0/−4
• src/Control/Concurrent/Chan/Typed.hs +1/−13
• src/Control/Concurrent/Chan/Typed/Extra.hs +0/−133
• src/Control/Concurrent/STM/TChan/Extra.hs +0/−99
• src/Control/Concurrent/STM/TChan/Typed.hs +1/−13
• src/Control/Concurrent/STM/TChan/Typed/Extra.hs +0/−104

@@ -1,44 +1,59 @@-name:                chan-version:             0.0.3-synopsis:            Some extra kit for Chans--- description:-homepage:            https://github.com/athanclark/chan#readme-license:             BSD3-license-file:        LICENSE-author:              Athan Clark-maintainer:          athan.clark@gmail.com-copyright:           2017 Athan Clark-category:            Web-build-type:          Simple-extra-source-files:  README.md-cabal-version:       >=1.10+cabal-version: 1.12 -library-  hs-source-dirs:      src-  exposed-modules:     Control.Concurrent.Chan.Extra-                       Control.Concurrent.Chan.Scope-                       Control.Concurrent.Chan.Typed-                       Control.Concurrent.Chan.Typed.Extra-                       Control.Concurrent.STM.TChan.Extra-                       Control.Concurrent.STM.TChan.Typed-                       Control.Concurrent.STM.TChan.Typed.Extra-  build-depends:       base >= 4.7 && < 5-                     , async-                     , stm-                     -- , since-  default-language:    Haskell2010+-- This file has been generated from package.yaml by hpack version 0.31.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 9429247fea789bb683e6f4ed7fb173854b095d3e98824b896891eb68655e73a8 -test-suite chan-test-  type:                exitcode-stdio-1.0-  hs-source-dirs:      test-  main-is:             Spec.hs-  build-depends:       base-                     , chan-                     , async-                     , stm-  ghc-options:         -threaded -rtsopts -with-rtsopts=-N-  default-language:    Haskell2010+name:           chan+version:        0.0.4+synopsis:       Some extra kit for Chans+description:    Please see the README on Github at <https://github.com/athanclark/chan#readme>+category:       Concurrency+homepage:       https://github.com/athanclark/chan#readme+bug-reports:    https://github.com/athanclark/chan/issues+author:         Athan Clark+maintainer:     athan.clark@localcooking.com+copyright:      2018 Athan Clark+license:        BSD3+license-file:   LICENSE+build-type:     Simple+extra-source-files:+    README.md  source-repository head-  type:     git+  type: git   location: https://github.com/athanclark/chan++library+  exposed-modules:+      Control.Concurrent.Chan.Extra+      Control.Concurrent.Chan.Scope+      Control.Concurrent.Chan.Typed+      Control.Concurrent.STM.TChan.Typed+  other-modules:+      Paths_chan+  hs-source-dirs:+      src+  ghc-options: -Wall+  build-depends:+      async+    , base >=4.11 && <5.0+    , stm+  default-language: Haskell2010++test-suite chan-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules:+      Paths_chan+  hs-source-dirs:+      test+  ghc-options: -Wall -threaded -rtsopts -Wall -with-rtsopts=-N+  build-depends:+      async+    , base+    , chan+    , stm+  default-language: Haskell2010

@@ -1,20 +1,225 @@+{-# LANGUAGE+    DataKinds+  , KindSignatures+  , MultiParamTypeClasses+  , FunctionalDependencies+  , FlexibleInstances+  , ScopedTypeVariables+  , PartialTypeSignatures+  #-}+ module Control.Concurrent.Chan.Extra where +import Control.Concurrent.Chan.Typed+  ( ChanRW (..), writeChanRW, readChanRW, newChanRW)+import Control.Concurrent.STM.TChan.Typed+  ( TChanRW (..), writeTChanRW, readTChanRW, newTChanRW)+import Control.Concurrent.Chan.Scope (Scope (..), Writable, Readable)+ import Data.IORef (newIORef, readIORef, writeIORef)--- import Data.Time.Since (timeSince, newTimeSince) import Control.Monad (forever) import Control.Concurrent (threadDelay) import Control.Concurrent.MVar (newEmptyMVar, tryTakeMVar, putMVar) import Control.Concurrent.Chan (Chan, readChan, writeChan, newChan) import Control.Concurrent.Async (Async, async, cancel, wait)+import Control.Concurrent.STM+  (atomically, tryTakeTMVar, putTMVar, newEmptyTMVarIO)+import Control.Concurrent.STM.TChan (TChan)  +-- | Class for changing the access of a typed channel+class ChanScoped (c :: Scope -> * -> *) where+  readOnly :: Readable scope => c scope a -> c 'Read a+  writeOnly :: Writable scope => c scope a -> c 'Write a+  allowReading :: Writable scope => c scope a -> c 'ReadWrite a+  allowWriting :: Readable scope => c scope a -> c 'ReadWrite a + type DiffNanosec = Int --- type TotalNanosec = Int +-- | Class for extra channel techniques+class ChanExtra (inputC :: * -> *) (outputC :: * -> *)+      | inputC -> outputC, outputC -> inputC where+  -- | Throw away messages that meet the threshold+  debounceStatic :: DiffNanosec -- ^ Time to wait before attempting to send the message+                 -> outputC a -> IO (inputC a, Async ())+  -- | Refrain from relaying messages that meet the threshold+  throttleStatic :: DiffNanosec -- ^ Time to wait before sending the message, for every message+                 -> outputC a -> IO (inputC a, Async ())+  -- | Intercalate messages while threshold is met+  intersperseStatic :: DiffNanosec -- ^ Time to at-least wait to intersperse messages+                    -> IO a -- ^ Get a message to intersperse+                    -> outputC a+                    -> IO (inputC a, Async (), Async ()) ++instance ChanExtra Chan Chan where+  debounceStatic toWaitFurther outputChan = do+    (ChanRW inputChan, t) <- debounceStatic toWaitFurther (ChanRW outputChan :: ChanRW 'Read _)+    pure (inputChan, t)+  throttleStatic toWaitFurther outputChan = do+    (ChanRW inputChan, t) <- throttleStatic toWaitFurther (ChanRW outputChan :: ChanRW 'Read _)+    pure (inputChan, t)+  intersperseStatic timeBetween xM outputChan = do+    (ChanRW inputChan, writer, listener) <- intersperseStatic timeBetween xM (ChanRW outputChan :: ChanRW 'Read _)+    pure (inputChan, writer, listener)+++instance ChanExtra TChan TChan where+  debounceStatic toWaitFurther outputTChan = do+    (TChanRW inputTChan, t) <- debounceStatic toWaitFurther (TChanRW outputTChan :: TChanRW 'Read _)+    pure (inputTChan, t)+  throttleStatic toWaitFurther outputTChan = do+    (TChanRW inputTChan, t) <- throttleStatic toWaitFurther (TChanRW outputTChan :: TChanRW 'Read _)+    pure (inputTChan, t)+  intersperseStatic timeBetween xM outputTChan = do+    (TChanRW inputTChan, writer, listener) <- intersperseStatic timeBetween xM (TChanRW outputTChan :: TChanRW 'Read _)+    pure (inputTChan, writer, listener)++++instance ChanScoped TChanRW where+  readOnly (TChanRW x) = TChanRW x+  writeOnly (TChanRW x) = TChanRW x+  allowReading (TChanRW x) = TChanRW x+  allowWriting (TChanRW x) = TChanRW x++instance ChanExtra (TChanRW 'Write) (TChanRW 'Read) where+  debounceStatic toWaitFurther outputChan = do+    presentedChan <- atomically newTChanRW+    writingThread <- newEmptyTMVarIO++    writer <- async $ forever $ do+      x <- atomically (readTChanRW presentedChan)+      newWriter <- async $ do+        threadDelay toWaitFurther+        atomically (writeTChanRW (allowWriting outputChan) x)+      mInvoker <- atomically (tryTakeTMVar writingThread)+      case mInvoker of+        Nothing -> pure ()+        Just i -> cancel i -- kill the old without writing to output+      atomically (putTMVar writingThread newWriter)++    pure (writeOnly presentedChan, writer)++  throttleStatic toWaitFurther outputChan = do+    presentedChan <- atomically newTChanRW+    writingThread <- newEmptyTMVarIO++    writer <- async $ forever $ do+      x <- atomically (readTChanRW presentedChan)+      mInvoker <- atomically (tryTakeTMVar writingThread)+      case mInvoker of+        Nothing -> pure ()+        Just i -> wait i -- cascade invocations, waiting until output writes+      newWriter <- async $ do+        threadDelay toWaitFurther+        atomically (writeTChanRW (allowWriting outputChan) x)+      atomically (putTMVar writingThread newWriter)++    pure (writeOnly presentedChan, writer)++  intersperseStatic timeBetween xM outputChan = do+    presentedChan <- atomically newTChanRW+    writingThread <- newEmptyTMVarIO++    writer <- async $ forever $ do+      mInvoker <- atomically (tryTakeTMVar writingThread)+      case mInvoker of+        Nothing -> pure () -- continue immediately if nothing's waiting+        Just i -> wait i -- block until previous ping is sent+      newWriter <- async $ do+        threadDelay timeBetween+        x <- xM -- get new ping message+        atomically (writeTChanRW (allowWriting outputChan) x) -- send it+      atomically (putTMVar writingThread newWriter) -- register pinger++    listener <- async $ forever $ do+      y <- atomically (readTChanRW presentedChan)+      mInvoker <- atomically (tryTakeTMVar writingThread)+      case mInvoker of+        Nothing -> pure ()+        Just i -> cancel i -- kill pinger++      atomically (writeTChanRW (allowWriting outputChan) y) -- immediately send regular message++    pure (writeOnly presentedChan, writer, listener)+++instance ChanScoped ChanRW where+  readOnly (ChanRW x) = ChanRW x+  writeOnly (ChanRW x) = ChanRW x+  allowReading (ChanRW x) = ChanRW x+  allowWriting (ChanRW x) = ChanRW x++instance ChanExtra (ChanRW 'Write) (ChanRW 'Read) where+  debounceStatic toWaitFurther outputChan = do+    presentedChan <- newChanRW+    writingThread <- newEmptyMVar++    writer <- async $ forever $ do+      x <- readChanRW presentedChan++      newWriter <- async $ do+        threadDelay toWaitFurther+        writeChanRW (allowWriting outputChan) x++      mInvoker <- tryTakeMVar writingThread+      case mInvoker of+        Nothing -> pure ()+        Just i -> cancel i+      putMVar writingThread newWriter++    pure (writeOnly presentedChan, writer)++  throttleStatic toWaitFurther outputChan = do+    presentedChan <- newChanRW+    writingThread <- newEmptyMVar++    writer <- async $ forever $ do+      x <- readChanRW presentedChan++      mInvoker <- tryTakeMVar writingThread+      case mInvoker of+        Nothing -> pure ()+        Just i -> wait i+      newWriter <- async $ do+        threadDelay toWaitFurther+        writeChanRW (allowWriting outputChan) x+      putMVar writingThread newWriter++    pure (writeOnly presentedChan, writer)++  intersperseStatic timeBetween xM outputChan = do+    presentedChan <- newChanRW+    writingThread <- newEmptyMVar++    writer <- async $ forever $ do+      mInvoker <- tryTakeMVar writingThread+      case mInvoker of+        Nothing -> pure ()+        Just i -> wait i+      newWriter <- async $ do+        threadDelay timeBetween+        x <- xM+        writeChanRW (allowWriting outputChan) x+      putMVar writingThread newWriter++    listener <- async $ forever $ do+      y <- readChanRW presentedChan++      mInvoker <- tryTakeMVar writingThread+      case mInvoker of+        Nothing -> pure ()+        Just i -> cancel i++      writeChanRW (allowWriting outputChan) y++    pure (writeOnly presentedChan, writer, listener)+++ -- debounceDynamic :: (NominalDiffTime -> NominalDiffTime) -> Chan a -> IO (Chan a, Async ()) -- debounceDynamic toWait outputChan = do --   sinceRef <- newTimeSince@@ -50,79 +255,3 @@ --     putMVar writingThread newWriter  --   pure (presentedChan, writer)--debounceStatic :: DiffNanosec -> Chan a -> IO (Chan a, Async ())-debounceStatic toWaitFurther outputChan = do-  presentedChan <- newChan-  writingThread <- newEmptyMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        writeChan outputChan x--  writer <- async $ forever $ do-    x <- readChan presentedChan--    newWriter <- async (invokeWrite x)--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i-    putMVar writingThread newWriter--  pure (presentedChan, writer)----- | Like debounce, but lossless-throttleStatic :: DiffNanosec -> Chan a -> IO (Chan a, Async ())-throttleStatic toWaitFurther outputChan = do-  presentedChan <- newChan-  writingThread <- newEmptyMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        writeChan outputChan x--  writer <- async $ forever $ do-    x <- readChan presentedChan--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async (invokeWrite x)-    putMVar writingThread newWriter--  pure (presentedChan, writer)---intersperseStatic :: DiffNanosec -> IO a -> Chan a -> IO (Chan a, Async (), Async ())-intersperseStatic timeBetween xM outputChan = do-  presentedChan <- newChan-  writingThread <- newEmptyMVar--  let invokeWritePing = do-        threadDelay timeBetween-        x <- xM-        writeChan outputChan x--  writer <- async $ forever $ do-    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async invokeWritePing-    putMVar writingThread newWriter--  listener <- async $ forever $ do-    y <- readChan presentedChan--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i--    writeChan outputChan y--  pure (presentedChan, writer, listener)

@@ -10,14 +10,10 @@   class Readable (a :: Scope) where- instance Readable 'Read where- instance Readable 'ReadWrite where   class Writable (a :: Scope) where- instance Writable 'Write where- instance Writable 'ReadWrite where

@@ -6,24 +6,12 @@ module Control.Concurrent.Chan.Typed where  import Control.Concurrent.Chan.Scope (Scope (..), Writable, Readable)+ import Control.Concurrent.Chan (Chan) import qualified Control.Concurrent.Chan as Chan   newtype ChanRW (scope :: Scope) a = ChanRW (Chan a)---readOnly :: Readable scope => ChanRW scope a -> ChanRW 'Read a-readOnly (ChanRW c) = ChanRW c--writeOnly :: Writable scope => ChanRW scope a -> ChanRW 'Write a-writeOnly (ChanRW c) = ChanRW c--allowReading :: Writable scope => ChanRW scope a -> ChanRW 'ReadWrite a-allowReading (ChanRW c) = ChanRW c--allowWriting :: Readable scope => ChanRW scope a -> ChanRW 'ReadWrite a-allowWriting (ChanRW c) = ChanRW c   newChanRW :: IO (ChanRW 'ReadWrite a)

@@ -1,133 +0,0 @@-{-# LANGUAGE-    DataKinds-  #-}--module Control.Concurrent.Chan.Typed.Extra where--import Data.IORef (newIORef, readIORef, writeIORef)--- import Data.Time.Since (timeSince, newTimeSince)-import Control.Monad (forever)-import Control.Concurrent (threadDelay)-import Control.Concurrent.MVar (newEmptyMVar, tryTakeMVar, putMVar)-import Control.Concurrent.Chan.Scope (Scope (..))-import Control.Concurrent.Chan.Typed (ChanRW, readChanRW, writeChanRW, newChanRW, writeOnly, allowWriting)-import Control.Concurrent.Async (Async, async, cancel, wait)----type DiffNanosec = Int---- type TotalNanosec = Int----- debounceDynamic :: (NominalDiffTime -> NominalDiffTime) -> Chan a -> IO (Chan a, Async ())--- debounceDynamic toWait outputChan = do---   sinceRef <- newTimeSince---   totalWaited <- newIORef 0---   presentedChan <- newChan---   writingThread <- newEmptyMVar----   let invokeWrite x = do---         putStrLn "Being run.."----         waited <- readIORef totalWaited---         let toWaitFurther = toWait waited---         writeIORef totalWaited (waited + toWaitFurther)----         -- FIXME must use clocktime http://hackage.haskell.org/package/chan- overlayed invocations have---         -- no concept of time spent, only have knoweldge of invocations---         -- made----         threadDelay toWaitFurther---         putStrLn "waited done"---         writeChan outputChan x----   writer <- async $ forever $ do---     x <- readChan presentedChan----     newWriter <- async (invokeWrite x)----     mInvoker <- tryTakeMVar writingThread---     case mInvoker of---       Nothing -> pure ()---       Just i -> cancel i---     print "killed"---     putMVar writingThread newWriter----   pure (presentedChan, writer)--debounceStatic :: DiffNanosec -> ChanRW 'Read a -> IO (ChanRW 'Write a, Async ())-debounceStatic toWaitFurther outputChan = do-  presentedChan <- newChanRW-  writingThread <- newEmptyMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        writeChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    x <- readChanRW presentedChan--    newWriter <- async (invokeWrite x)--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i-    putMVar writingThread newWriter--  pure (writeOnly presentedChan, writer)----- | Like debounce, but lossless-throttleStatic :: DiffNanosec -> ChanRW 'Read a -> IO (ChanRW 'Write a, Async ())-throttleStatic toWaitFurther outputChan = do-  presentedChan <- newChanRW-  writingThread <- newEmptyMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        writeChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    x <- readChanRW presentedChan--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async (invokeWrite x)-    putMVar writingThread newWriter--  pure (writeOnly presentedChan, writer)---intersperseStatic :: DiffNanosec -> IO a -> ChanRW 'Read a -> IO (ChanRW 'Write a, Async (), Async ())-intersperseStatic timeBetween xM outputChan = do-  presentedChan <- newChanRW-  writingThread <- newEmptyMVar--  let invokeWritePing = do-        threadDelay timeBetween-        x <- xM-        writeChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async invokeWritePing-    putMVar writingThread newWriter--  listener <- async $ forever $ do-    y <- readChanRW presentedChan--    mInvoker <- tryTakeMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i--    writeChanRW (allowWriting outputChan) y--  pure (writeOnly presentedChan, writer, listener)

@@ -1,99 +0,0 @@-module Control.Concurrent.STM.TChan.Extra where--import Control.Monad (forever)-import Control.Concurrent (threadDelay)-import Control.Concurrent.STM (STM, atomically)--- import Control.Concurrent.STM.TVar (newTVar, readIORef, writeIORef)-import Control.Concurrent.STM.TMVar (newEmptyTMVar, tryTakeTMVar, putTMVar)-import Control.Concurrent.STM.TChan (TChan, readTChan, writeTChan, newTChan)-import Control.Concurrent.Async (Async, async, cancel, wait)----type DiffNanosec = Int----- | Note: In this model, even though we are using STM, a write to the--- outgoing channel does not imply a transactional write to the output--- channel; they are separated between a run IO layer, which means--- we cannot atomically debounce or interleave the system (because--- that depends on real-world time).-debounceStatic :: DiffNanosec -> TChan a -> IO (TChan a, Async ())-debounceStatic toWaitFurther outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChan-                                             <*> newEmptyTMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        atomically $ writeTChan outputChan x--  writer <- async $ forever $ do-    x <- atomically $ readTChan presentedChan--    newWriter <- async (invokeWrite x)--    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i-    atomically $ putTMVar writingThread newWriter--  pure (presentedChan, writer)---throttleStatic :: DiffNanosec -> TChan a -> IO (TChan a, Async ())-throttleStatic toWaitFurther outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChan-                                             <*> newEmptyTMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        atomically $ writeTChan outputChan x--  writer <- async $ forever $ do-    x <- atomically $ readTChan presentedChan--    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async (invokeWrite x)-    atomically $ putTMVar writingThread newWriter--  pure (presentedChan, writer)---intersperseStatic :: DiffNanosec -> IO a -> TChan a -> IO (TChan a, Async (), Async ())-intersperseStatic timeBetween xM outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChan-                                             <*> newEmptyTMVar--  let invokeWritePing = do-        threadDelay timeBetween-        x <- xM-        atomically $ writeTChan outputChan x--  writer <- async $ forever $ do-    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async invokeWritePing-    atomically $ putTMVar writingThread newWriter--  listener <- async $ forever $ do-    (y,mInvoker) <- atomically $ do-      y' <- readTChan presentedChan--      (\q -> (y',q)) <$> tryTakeTMVar writingThread--    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i--    atomically $ writeTChan outputChan y--  pure (presentedChan, writer, listener)

@@ -6,25 +6,13 @@ module Control.Concurrent.STM.TChan.Typed where  import Control.Concurrent.Chan.Scope (Scope (..), Writable, Readable)+ import Control.Concurrent.STM.TChan (TChan) import qualified Control.Concurrent.STM.TChan as TChan import Control.Concurrent.STM (STM)   newtype TChanRW (scope :: Scope) a = TChanRW (TChan a)---readOnly :: Readable scope => TChanRW scope a -> TChanRW 'Read a-readOnly (TChanRW c) = TChanRW c--writeOnly :: Writable scope => TChanRW scope a -> TChanRW 'Write a-writeOnly (TChanRW c) = TChanRW c--allowReading :: Writable scope => TChanRW scope a -> TChanRW 'ReadWrite a-allowReading (TChanRW c) = TChanRW c--allowWriting :: Readable scope => TChanRW scope a -> TChanRW 'ReadWrite a-allowWriting (TChanRW c) = TChanRW c   newTChanRW :: STM (TChanRW 'ReadWrite a)

@@ -1,104 +0,0 @@-{-# LANGUAGE-    DataKinds-  #-}--module Control.Concurrent.STM.TChan.Typed.Extra where--import Control.Monad (forever)-import Control.Concurrent (threadDelay)-import Control.Concurrent.STM (STM, atomically)--- import Control.Concurrent.STM.TVar (newTVar, readIORef, writeIORef)-import Control.Concurrent.STM.TMVar (newEmptyTMVar, tryTakeTMVar, putTMVar)-import Control.Concurrent.Chan.Scope (Scope (..))-import Control.Concurrent.STM.TChan.Typed (TChanRW, readTChanRW, writeTChanRW, newTChanRW, allowWriting, writeOnly)-import Control.Concurrent.Async (Async, async, cancel, wait)----type DiffNanosec = Int----- | Note: In this model, even though we are using STM, a write to the--- outgoing channel does not imply a transactional write to the output--- channel; they are separated between a run IO layer, which means--- we cannot atomically debounce or interleave the system (because--- that depends on real-world time).-debounceStatic :: DiffNanosec -> TChanRW 'Read a -> IO (TChanRW 'Write a, Async ())-debounceStatic toWaitFurther outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChanRW-                                             <*> newEmptyTMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        atomically $ writeTChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    x <- atomically $ readTChanRW presentedChan--    newWriter <- async (invokeWrite x)--    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i-    atomically $ putTMVar writingThread newWriter--  pure (writeOnly presentedChan, writer)---throttleStatic :: DiffNanosec -> TChanRW 'Read a -> IO (TChanRW 'Write a, Async ())-throttleStatic toWaitFurther outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChanRW-                                             <*> newEmptyTMVar--  let invokeWrite x = do-        threadDelay toWaitFurther-        atomically $ writeTChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    x <- atomically $ readTChanRW presentedChan--    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async (invokeWrite x)-    atomically $ putTMVar writingThread newWriter--  pure (writeOnly presentedChan, writer)---intersperseStatic :: DiffNanosec -> IO a -> TChanRW 'Read a -> IO (TChanRW 'Write a, Async (), Async ())-intersperseStatic timeBetween xM outputChan = do-  (presentedChan,writingThread) <- atomically $ (,)-                                             <$> newTChanRW-                                             <*> newEmptyTMVar--  let invokeWritePing = do-        threadDelay timeBetween-        x <- xM-        atomically $ writeTChanRW (allowWriting outputChan) x--  writer <- async $ forever $ do-    mInvoker <- atomically $ tryTakeTMVar writingThread-    case mInvoker of-      Nothing -> pure ()-      Just i -> wait i-    newWriter <- async invokeWritePing-    atomically $ putTMVar writingThread newWriter--  listener <- async $ forever $ do-    (y,mInvoker) <- atomically $ do-      y' <- readTChanRW presentedChan--      (\q -> (y',q)) <$> tryTakeTMVar writingThread--    case mInvoker of-      Nothing -> pure ()-      Just i -> cancel i--    atomically $ writeTChanRW (allowWriting outputChan) y--  pure (writeOnly presentedChan, writer, listener)