chan 0.0.2 → 0.0.3

raw patch · 7 files changed

+425/−1 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +40/−0
• chan.cabal +6/−1
• src/Control/Concurrent/Chan/Scope.hs +23/−0
• src/Control/Concurrent/Chan/Typed.hs +44/−0
• src/Control/Concurrent/Chan/Typed/Extra.hs +133/−0
• src/Control/Concurrent/STM/TChan/Typed.hs +75/−0
• src/Control/Concurrent/STM/TChan/Typed/Extra.hs +104/−0

@@ -1,1 +1,41 @@ # chan++This is just some extra `Chan` and `TChan` kit that might help the average user. It relies on+spawning threads with `async` and either canceling (debouncing) or waiting (throttling) messages.++Unfortunately, the current design is untyped in the sense that the channel which you supply is the+_output_ channel, and the returned channel is the one you would write to. I'm not sure how this should+be fixed.++An example might be the following:++```haskell+import Control.Concurrent.Chan (readChan)+import Control.Concurrent.Chan.Extra (throttleStatic, intersperseStatic)++++-- For example, some websockets:++data SomeMessage+  = Ping+  -- | ...++throttleLayer :: Chan SomeMessage -> IO (Chan SomeMessage)+throttleLayer output = do+  (x,_) <- throttleStatic output 1000000 -- nanoseconds, = 1 second+  pure x++pingLayer :: Chan SomeMessage -> IO (Chan SomeMessage)+pingLayer output = do+  (x,_,_) <- intersperseStatic output (pure Ping) 1000000+  pure x++performWebsocket :: Chan SomeMessage -> IO ()+performWebsocket output = do+  output' <- pingLayer =<< throttleLayer output+  _ <- async $ forever $ do+    msg <- readChan output'+    send msg -- something like that - it'll include Ping messages for us,+             -- and throttle the outgoing messages at the same time.+```

@@ -1,5 +1,5 @@ name:                chan-version:             0.0.2+version:             0.0.3 synopsis:            Some extra kit for Chans -- description: homepage:            https://github.com/athanclark/chan#readme@@ -16,7 +16,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

@@ -0,0 +1,23 @@+{-# LANGUAGE+    DataKinds+  , KindSignatures+  #-}++module Control.Concurrent.Chan.Scope where+++data Scope = Read | Write | ReadWrite+++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

@@ -0,0 +1,44 @@+{-# LANGUAGE+    DataKinds+  , KindSignatures+  #-}++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)+newChanRW = ChanRW <$> Chan.newChan+++writeChanRW :: Writable scope => ChanRW scope a -> a -> IO ()+writeChanRW (ChanRW c) x = Chan.writeChan c x+++readChanRW :: Readable scope => ChanRW scope a -> IO a+readChanRW (ChanRW c) = Chan.readChan c+++dupChanRW :: Writable scopeIn+          => Readable scopeOut+          => ChanRW scopeIn a -> IO (ChanRW scopeOut a)+dupChanRW (ChanRW c) = ChanRW <$> Chan.dupChan c

@@ -0,0 +1,133 @@+{-# 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)

@@ -0,0 +1,75 @@+{-# LANGUAGE+    DataKinds+  , KindSignatures+  #-}++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)+newTChanRW = TChanRW <$> TChan.newTChan+++writeTChanRW :: Writable scope => TChanRW scope a -> a -> STM ()+writeTChanRW (TChanRW c) x = TChan.writeTChan c x+++unGetTChanRW :: Writable scope => TChanRW scope a -> a -> STM ()+unGetTChanRW (TChanRW c) x = TChan.unGetTChan c x+++isEmptyTChanRW :: Readable scope => TChanRW scope a -> STM Bool+isEmptyTChanRW (TChanRW c) = TChan.isEmptyTChan c+++readTChanRW :: Readable scope => TChanRW scope a -> STM a+readTChanRW (TChanRW c) = TChan.readTChan c+++tryReadTChanRW :: Readable scope => TChanRW scope a -> STM (Maybe a)+tryReadTChanRW (TChanRW c) = TChan.tryReadTChan c+++peekTChanRW :: Readable scope => TChanRW scope a -> STM a+peekTChanRW (TChanRW c) = TChan.peekTChan c+++tryPeekTChanRW :: Readable scope => TChanRW scope a -> STM (Maybe a)+tryPeekTChanRW (TChanRW c) = TChan.tryPeekTChan c+++newBroadcastTChanRW :: STM (TChanRW 'Write a)+newBroadcastTChanRW = TChanRW <$> TChan.newBroadcastTChan+++dupTChanRW :: Writable scopeIn+           => Readable scopeOut+           => TChanRW scopeIn a -> STM (TChanRW scopeOut a)+dupTChanRW (TChanRW c) = TChanRW <$> TChan.dupTChan c+++cloneTChanRW :: Writable scopeIn+           => Readable scopeOut+           => TChanRW scopeIn a -> STM (TChanRW scopeOut a)+cloneTChanRW (TChanRW c) = TChanRW <$> TChan.cloneTChan c

@@ -0,0 +1,104 @@+{-# 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)