chan 0.0.1 → 0.0.2

raw patch · 4 files changed

+124/−5 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• chan.cabal +1/−1
• src/Control/Concurrent/Chan/Extra.hs +23/−1
• src/Control/Concurrent/STM/TChan/Extra.hs +22/−0
• test/Spec.hs +78/−3

@@ -1,5 +1,5 @@ name:                chan-version:             0.0.1+version:             0.0.2 synopsis:            Some extra kit for Chans -- description: homepage:            https://github.com/athanclark/chan#readme

@@ -29,7 +29,7 @@ --         let toWaitFurther = toWait waited --         writeIORef totalWaited (waited + toWaitFurther) ---         -- FIXME must use clocktime - overlayed invocations have+--         -- FIXME must use clocktime http://hackage.haskell.org/package/chan- overlayed invocations have --         -- no concept of time spent, only have knoweldge of invocations --         -- made @@ -73,6 +73,28 @@    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 ())

@@ -42,6 +42,28 @@   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

@@ -1,7 +1,7 @@ import Control.Monad (forever) import Control.Concurrent (threadDelay) import Control.Concurrent.Chan (newChan, writeChan, readChan)-import Control.Concurrent.Chan.Extra (debounceStatic, intersperseStatic)+import Control.Concurrent.Chan.Extra (debounceStatic, throttleStatic, intersperseStatic) import Control.Concurrent.Async (async) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM (atomically)@@ -9,9 +9,9 @@ import qualified Control.Concurrent.STM.TChan.Extra as TChan import Control.Concurrent.STM.TMVar (newEmptyTMVar, putTMVar, takeTMVar) -main :: IO ()-main = do +chanDebounceTest = do+  putStrLn "Debounce ---------"   outgoing <- newChan   (incoming, _) <- debounceStatic 1000000 outgoing @@ -33,8 +33,35 @@    _ <- takeMVar lock +  pure () +chanThrottleTest = do+  putStrLn "Throttle ---------"+  outgoing <- newChan+  (incoming, _) <- throttleStatic 1000000 outgoing +  lock <- newChan++  _ <- async $ forever $ do+    x <- readChan outgoing+    print x+    writeChan lock ()++  putStrLn "writing 1..."+  writeChan incoming 1+  putStrLn "writing 2..."+  writeChan incoming 2+  putStrLn "writing 3..."+  writeChan incoming 3++  _ <- readChan lock+  _ <- readChan lock+  _ <- readChan lock++  pure ()++chanIntersperseTest = do+  putStrLn "Intersperse ---------"   outgoing <- newChan   (incoming, _, _) <- intersperseStatic 1000000 (pure 0) outgoing @@ -57,6 +84,13 @@   pure ()  +chanTests = do+  chanDebounceTest+  chanThrottleTest+  chanIntersperseTest++tchanDebounceTests = do+  putStrLn "Debounce STM ---------"   outgoing <- atomically newTChan   (incoming, _) <- TChan.debounceStatic 1000000 outgoing @@ -78,8 +112,36 @@    _ <- atomically $ takeTMVar lock +  pure () +tchanThrottleTests = do+  putStrLn "Throttle STM ---------"+  outgoing <- atomically newTChan+  (incoming, _) <- TChan.throttleStatic 1000000 outgoing +  lock <- atomically newTChan++  _ <- async $ forever $ do+    x <- atomically $ readTChan outgoing+    print x+    atomically $ writeTChan lock ()++  putStrLn "writing 1..."+  atomically $ writeTChan incoming 1+  putStrLn "writing 2..."+  atomically $ writeTChan incoming 2+  putStrLn "writing 3..."+  atomically $ writeTChan incoming 3++  _ <- atomically $ readTChan lock+  _ <- atomically $ readTChan lock+  _ <- atomically $ readTChan lock++  pure ()+++tchanIntersperseTests = do+  putStrLn "Intersperse STM ---------"   outgoing <- atomically newTChan   (incoming, _, _) <- TChan.intersperseStatic 1000000 (pure 0) outgoing @@ -100,3 +162,16 @@   _ <- atomically $ takeTMVar lock    pure ()+++tchanTests = do+  tchanDebounceTests+  tchanThrottleTests+  tchanIntersperseTests+++main :: IO ()+main = do+  chanTests+  threadDelay 2000000+  tchanTests