equeue (empty) → 0
raw patch · 9 files changed
+736/−0 lines, 9 filesdep +basedep +containersdep +contravariantsetup-changed
Dependencies added: base, containers, contravariant, delay, equeue, mtl, semigroups, stm, tasty, tasty-hunit, time
Files
- CHANGELOG.md +5/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- equeue.cabal +65/−0
- src/Control/Concurrent/EQueue.hs +45/−0
- src/Control/Concurrent/EQueue/Class.hs +30/−0
- src/Control/Concurrent/EQueue/STMEQueue.hs +76/−0
- src/Control/Concurrent/EQueue/Simple.hs +62/−0
- test/test.hs +421/−0
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for equeue++## 0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2018, davean++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of davean nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ equeue.cabal view
@@ -0,0 +1,65 @@+cabal-version: 2.2++name: equeue+version: 0+synopsis: Application level triggered, and edge triggered event multiqueues.+description:+ A system for providing late binding for how different types of events are handled.+ .+ It is often important for an application to control how it consumes updates for optimal+ processing, but event sources are typically in control of delivery. The EQueue abstraction+ allows the consumer to provide an implimentation that balances delivery for its needs.+ To do this, it distinguishes events into two types, level and edge triggered.+ .+ Level triggered events are used where the resulting state is cared about and a pure model+ of how events combine is available as a Semigroup instance. The transitions it took to get+ to the new state between dequeues is not of interest.+ .+ Edge triggered events are where the sequence of occurences are of importance, or a+ pure model is not available.+homepage: https://oss.xkcd.com/+bug-reports: https://code.xkrd.net/AlON/equeue/issues+license: BSD-3-Clause+license-file: LICENSE+author: davean+maintainer: oss@xkcd.com+copyright: Copyright (C) 2018 davean+category: Concurrency+extra-source-files: CHANGELOG.md++source-repository head+ type: git+ location: https://code.xkrd.net/AlON/equeue.git++library+ hs-source-dirs: src+ default-language: Haskell2010+ exposed-modules:+ Control.Concurrent.EQueue+ , Control.Concurrent.EQueue.Class+ , Control.Concurrent.EQueue.Simple+ , Control.Concurrent.EQueue.STMEQueue+ build-depends:+ base >= 4.8 && < 4.13+ , contravariant ^>= 1.5+ , containers ^>= 0.6+ , mtl ^>= 2.2+ , semigroups ^>= 0.18+ , stm ^>= 2.5++test-suite test-equeue+ type: exitcode-stdio-1.0+ default-language: Haskell2010+ hs-source-dirs: test+ main-is: test.hs+ build-depends:+ base >= 4.8 && < 4.13+ , containers+ , contravariant+ , delay ^>= 0+ , equeue+ , semigroups+ , stm+ , tasty >= 1.1 && < 1.3+ , tasty-hunit ^>= 0.10+ , time >= 1.8 && < 1.10
+ src/Control/Concurrent/EQueue.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeFamilies #-}+module Control.Concurrent.EQueue+ ( EQueue(registerSemi, registerQueued), EQueueW(waitEQ)+ , AnyEQueue(AEQ), ForceEdge(EEQ), MappedEQueue(MEQ), meqEQ+ , STMEQueue, STMEQueueWait(..), newSTMEQueue+ ) where++import Control.Concurrent.EQueue.Class+import Control.Concurrent.EQueue.STMEQueue (STMEQueue, STMEQueueWait(..), newSTMEQueue)+import Data.Bifunctor+import Data.Functor.Contravariant++-- | Allows us to return an unknown instance of EQueue, getting around Haskells lack of+-- existential qualification.+data AnyEQueue a where+ AEQ :: EQueue eq => eq a -> AnyEQueue a++instance EQueue AnyEQueue where+ registerSemi (AEQ eq) = registerSemi eq+ registerQueued (AEQ eq) = registerQueued eq++-- | A wrapper that translates level triggered events into events that observe the edges.+data ForceEdge a where+ EEQ :: EQueue eq => eq a -> ForceEdge a++instance EQueue ForceEdge where+ registerSemi (EEQ eq) f = (first (. f)) <$> registerQueued eq+ registerQueued (EEQ eq) = registerQueued eq++-- | A wrapper that allows us to pretend a queue of one type is of another.+data MappedEQueue eq b a where+ MEQ :: (a -> b) -> eq b -> MappedEQueue eq b a++-- | Retrieve the EQueue we're mapping to from the MappedEQueue.+meqEQ :: MappedEQueue eq b a -> eq b+meqEQ (MEQ _ eq) = eq++instance Contravariant (MappedEQueue eq b) where+ contramap f (MEQ g eq) = MEQ (g . f) eq++instance EQueue eq => EQueue (MappedEQueue eq b) where+ registerSemi (MEQ g eq) f = registerSemi eq (g . f)+ registerQueued (MEQ g eq) = (first (. g)) <$> registerQueued eq
+ src/Control/Concurrent/EQueue/Class.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE TypeFamilies #-}++-- | This module provides the core abstractions that represent managers for queueable events.++module Control.Concurrent.EQueue.Class where++import Control.Monad.Trans+import Data.Semigroup++-- | An EQueue is a way of managing edge and level triggered events.+-- The choice of EQueue implementation allows late binding of the policy+-- by which the application processes events.+class EQueue eq where+ -- | Registers a level triggered event.+ -- These are the events that accumulate a combined change or resulting state.+ -- Returns a function to enqueue updates and unregister this event.+ registerSemi :: (MonadIO m, Semigroup b) => eq a -> (b -> a) -> m (b -> IO (), IO ())+ -- | Registers an edge triggered event.+ -- Returns a function to enqueue updates and unregister this event.+ registerQueued :: MonadIO m => eq a -> m (a -> IO (), IO ())++-- | An EQueueW is the interface for waiting on events in a queue.+class EQueueW eq where+ -- | The WaitPolicy allows control per-call of waitEQ as to which policy is followed.+ -- For example, if it should return immediately if there are no events to dequeue+ -- or if it should wait for at least one event to be available.+ type WaitPolicy eq :: *+ -- | The dequeue operation, collecting some set of available events, depending on the+ -- particular policy the given EQueue impliments.+ waitEQ :: MonadIO m => eq a -> WaitPolicy eq -> m [a]
+ src/Control/Concurrent/EQueue/STMEQueue.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeFamilies #-}++-- | This module provides an EQueue implimentation that uses STM to wait on the first+-- available event among a set of possible events. When Waited on it can provide up to one+-- event from every available event source when it provides one event, allowing coalescing.++module Control.Concurrent.EQueue.STMEQueue where++import Control.Concurrent.EQueue.Class+import Control.Concurrent.STM+import Control.Monad.Trans+import Data.Foldable+import Data.Map (Map)+import qualified Data.Map as Map+import Data.Maybe (catMaybes)+import Data.Semigroup+import Data.Unique++-- | A basic example implimentation of an EQueue using STM.+-- This implimentation must look at every registered event source+-- leading to inefficiency in systems with a very large number of+-- sources. For most systems it should be a sufficient implimentation+-- though.+data STMEQueue a =+ STMEQueue+ { _eqActiveSources :: TVar (Map Unique (STM (Maybe a)))+ }++-- | Passed an STM dequeueing the current value of this signal.+-- Returns an action to unregister said.+register :: (MonadIO m) => STMEQueue a -> STM (Maybe a) -> m (IO ())+register (STMEQueue tqm) g = liftIO $ do+ u <- newUnique+ atomically $ do+ modifyTVar tqm (Map.insert u g)+ return . atomically $ modifyTVar tqm (Map.delete u)++-- | Create a new STMEQueue which initally has no event sources registered.+newSTMEQueue :: MonadIO m => m (STMEQueue a)+newSTMEQueue = liftIO $ STMEQueue <$> newTVarIO mempty++instance EQueue STMEQueue where+ registerSemi eq f = liftIO $ do+ t <- newEmptyTMVarIO+ (mappendTMVar t,) <$> register eq ((fmap f) <$> tryTakeTMVar t)+ where+ mappendTMVar :: Semigroup a => TMVar a -> a -> IO ()+ mappendTMVar t a = atomically $ do+ mv <- tryTakeTMVar t+ case mv of+ Nothing -> putTMVar t a+ Just v -> putTMVar t (v <> a)++ registerQueued eq = liftIO $ do+ t <- newTQueueIO+ (atomically . writeTQueue t,) <$> register eq (tryReadTQueue t)++-- | The policy for waiting on an STMEQueue.+data STMEQueueWait =+ ReturnImmediate+ -- ^ Immediately return, even if no events are available.+ | RequireEvent+ -- ^ Wait for at least one event to be available before returning.+ deriving (Eq)++instance EQueueW STMEQueue where+ type WaitPolicy STMEQueue = STMEQueueWait++ waitEQ (STMEQueue tqm) wp = liftIO . atomically $ do+ qm <- readTVar tqm+ es <- catMaybes <$> (sequenceA . toList $ qm)+ if (null es && wp == RequireEvent)+ then retry+ else return es
+ src/Control/Concurrent/EQueue/Simple.hs view
@@ -0,0 +1,62 @@+-- | Simple compatability newtype wrappers to standard queues used in the Haskell wild.+{-# LANGUAGE TypeFamilies #-}+module Control.Concurrent.EQueue.Simple (+ JustOneEventually(..)+ , ChanEQueue(..)+ , TChanEQueue(..)+ , TQueueEQueue(..)+ , IOEQueue(..)+ ) where++import Control.Concurrent.Chan+import Control.Concurrent.EQueue.Class+import Control.Concurrent.STM+import Control.Monad.Trans++-- | A policy for waiting until a single event can be gotten and returning just one.+-- Since these standard queing options are fairly simple the don't allows more+-- detailed policies abiding by EQueue's laws.+data JustOneEventually = JustOneEventually++-- | A wrapper for EQueueing events into a Chan.+newtype ChanEQueue a = CEQ (Chan a)++{- | Can not unregister events -}+instance EQueue ChanEQueue where+ registerSemi (CEQ c) f = return (writeChan c . f, return ())+ registerQueued (CEQ c) = return (writeChan c, return ())++instance EQueueW ChanEQueue where+ type WaitPolicy ChanEQueue = JustOneEventually+ waitEQ (CEQ c) JustOneEventually = fmap pure . liftIO . readChan $ c++-- | A wrapper for EQueueing events into a TChan.+newtype TChanEQueue a = TCEQ (TChan a)++{- | Can not unregister events -}+instance EQueue TChanEQueue where+ registerSemi (TCEQ c) f = return (atomically . writeTChan c . f, return ())+ registerQueued (TCEQ c) = return (atomically . writeTChan c, return ())++instance EQueueW TChanEQueue where+ type WaitPolicy TChanEQueue = JustOneEventually+ waitEQ (TCEQ c) JustOneEventually = fmap pure . liftIO . atomically . readTChan $ c++-- | A wrapper for EQueueing events into a TQueue.+newtype TQueueEQueue a = TQEQ (TQueue a)++{- | Can not unregister events -}+instance EQueue TQueueEQueue where+ registerSemi (TQEQ c) f = return (atomically . writeTQueue c . f, return ())+ registerQueued (TQEQ c) = return (atomically . writeTQueue c, return ())++instance EQueueW TQueueEQueue where+ type WaitPolicy TQueueEQueue = JustOneEventually+ waitEQ (TQEQ c) JustOneEventually = fmap pure . liftIO . atomically . readTQueue $ c++-- | A wrapper for EQueueing events that we have an IO action to submit.+newtype IOEQueue a = IOEQ (a -> IO ())++instance EQueue IOEQueue where+ registerSemi (IOEQ act) f = return (act . f, return ())+ registerQueued (IOEQ act) = return (act, return ())
+ test/test.hs view
@@ -0,0 +1,421 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeFamilies #-}+module Main (main) where++import Control.Arrow ((&&&))+import Control.Concurrent+import Control.Concurrent.EQueue+import Control.Concurrent.EQueue.Class+import Control.Concurrent.EQueue.Simple+import Control.Concurrent.EQueue.STMEQueue+import Control.Concurrent.STM+import Control.Monad+import Control.Time+import Data.Functor.Contravariant+import qualified Data.Map as Map+import Data.Semigroup+import Data.Time+import Test.Tasty+import Test.Tasty.HUnit++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "EQueue"+ [ stmeqTests+ , anyTests+ , forceEdgeTests+ , mappedTests+ , simpleTests "ChanEQueue Tests" (CEQ <$> newChan)+ , simpleTests "TChanEQueue Tests" (TCEQ <$> newTChanIO)+ , simpleTests "TQueueEQueue Tests" (TQEQ <$> newTQueueIO)+ , ioTests+ ]++numSources :: STMEQueue a -> IO Int+numSources = fmap Map.size . atomically . readTVar . _eqActiveSources++trivialSource :: STM (Maybe ())+trivialSource = return (Just ())++waitEQBlocks :: (Eq a, Show a) => STMEQueue a -> IO ()+waitEQBlocks eq = do+ r <- timeout (0.05::Double) $ waitEQ eq RequireEvent+ r @=? Nothing++waitEQDelaysForAdd :: NominalDiffTime -> TestTree+waitEQDelaysForAdd d = testCase ("waitEQ delay for register ("++show d++")") $ do+ eq <- newSTMEQueue+ st <- getCurrentTime+ void $ forkIO (delay ((fromRational.toRational $ d)::Double) >> (void $ register eq trivialSource))+ void $ waitEQ eq RequireEvent+ t <- (`diffUTCTime` st) <$> getCurrentTime+ assertBool "In time window" (t >= d && t < 2*d)++stmeqTests :: TestTree+stmeqTests = testGroup "STMEQueue"+ [ testCase "No initial Sources" $ do+ (0 @?=) =<< numSources =<< newSTMEQueue+ , testCase "register adds" $ do+ eq <- newSTMEQueue+ (0 @?=) =<< numSources eq+ void $ register eq trivialSource+ (1 @?=) =<< numSources eq+ , testCase "Empty waitEQ blocks" $ do+ eq <- newSTMEQueue::IO (STMEQueue ())+ waitEQBlocks eq+ , testCase "Empty waitEQ returns immediate" $ do+ eq <- newSTMEQueue::IO (STMEQueue ())+ ([] @=?) =<< waitEQ eq ReturnImmediate+ , testCase "waitEQ retrieves" $ do+ eq <- newSTMEQueue+ void $ register eq trivialSource+ ([()] @=?) =<< waitEQ eq RequireEvent+ ([()] @=?) =<< waitEQ eq RequireEvent+ , waitEQDelaysForAdd 0.01+ , waitEQDelaysForAdd 0.05+ , testCase "register killer removes" $ do+ eq <- newSTMEQueue+ k <- register eq trivialSource+ ([()] @=?) =<< waitEQ eq RequireEvent+ k+ waitEQBlocks eq+ , testGroup "registerSemi"+ [ testCase "adds" $ do+ eq <- newSTMEQueue+ (add, _) <- registerSemi eq id+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue [Int] <- newSTMEQueue+ (add, _) <- registerSemi eq id+ add [0]+ add [1]+ ([[0,1]] @=?) =<< waitEQ eq RequireEvent+ add [2]+ ([[2]] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq <- newSTMEQueue+ (add, k) <- registerSemi eq id+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ , testCase "maps" $ do+ eq::STMEQueue (Char, Int) <- newSTMEQueue+ (add, _) <- registerSemi eq (('a',) . getMax::Max Int -> (Char, Int))+ add (Max 0)+ add (Max 1)+ ([('a', 1)] @=?) =<< waitEQ eq RequireEvent+ ]+ , testGroup "registerQueued"+ [ testCase "adds" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, _) <- registerQueued eq+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue Int <- newSTMEQueue+ (add, _) <- registerQueued eq+ add 0+ add 1+ ([0] @=?) =<< waitEQ eq RequireEvent+ ([1] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, k) <- registerQueued eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ ]+ , testCase "Gets all latest" $ do+ eq::STMEQueue (Either String Int) <- newSTMEQueue+ (addQ, _) <- registerQueued eq+ (addS, _) <- registerSemi eq id+ addQ (Left "a")+ addQ (Left "b")+ addQ (Left "c")+ ([Left "a"] @=?) =<< waitEQ eq RequireEvent+ addS (Right 2)+ addS (Right 1)+ ([Left "b", Right 2] @=?) =<< waitEQ eq RequireEvent+ ([Left "c"] @=?) =<< waitEQ eq RequireEvent+ ]++anyTests :: TestTree+anyTests = testGroup "AnyEQueue"+ [ testGroup "registerSemi"+ [ testCase "adds" $ do+ eq <- newSTMEQueue+ (add, _) <- registerSemi (AEQ eq) id+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue [Int] <- newSTMEQueue+ (add, _) <- registerSemi (AEQ eq) id+ add [0]+ add [1]+ ([[0,1]] @=?) =<< waitEQ eq RequireEvent+ add [2]+ ([[2]] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq <- newSTMEQueue+ (add, k) <- registerSemi (AEQ eq) id+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ , testCase "maps" $ do+ eq::STMEQueue (Char, Int) <- newSTMEQueue+ (add, _) <- registerSemi (AEQ eq) (('a',) . getMax::Max Int -> (Char, Int))+ add (Max 0)+ add (Max 1)+ ([('a', 1)] @=?) =<< waitEQ eq RequireEvent+ ]+ , testGroup "registerQueued"+ [ testCase "adds" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, _) <- registerQueued (AEQ eq)+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue Int <- newSTMEQueue+ (add, _) <- registerQueued (AEQ eq)+ add 0+ add 1+ ([0] @=?) =<< waitEQ eq RequireEvent+ ([1] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, k) <- registerQueued (AEQ eq)+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ ]+ , testCase "Gets all latest" $ do+ eq::STMEQueue (Either String Int) <- newSTMEQueue+ (addQ, _) <- registerQueued (AEQ eq)+ (addS, _) <- registerSemi (AEQ eq) id+ addQ (Left "a")+ addQ (Left "b")+ addQ (Left "c")+ ([Left "a"] @=?) =<< waitEQ eq RequireEvent+ addS (Right 2)+ addS (Right 1)+ ([Left "b", Right 2] @=?) =<< waitEQ eq RequireEvent+ ([Left "c"] @=?) =<< waitEQ eq RequireEvent+ ]++forceEdgeTests :: TestTree+forceEdgeTests = testGroup "AnyEQueue"+ [ testGroup "registerSemi"+ [ testCase "adds" $ do+ eq <- newSTMEQueue+ (add, _) <- registerSemi (EEQ eq) id+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue [Int] <- newSTMEQueue+ (add, _) <- registerSemi (EEQ eq) id+ add [0]+ add [1]+ ([[0]] @=?) =<< waitEQ eq RequireEvent+ ([[1]] @=?) =<< waitEQ eq RequireEvent+ add [2]+ ([[2]] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq <- newSTMEQueue+ (add, k) <- registerSemi (EEQ eq) id+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ , testCase "maps" $ do+ eq::STMEQueue (Char, Int) <- newSTMEQueue+ (add, _) <- registerSemi (EEQ eq) (('a',) . getMax::Max Int -> (Char, Int))+ add (Max 0)+ add (Max 1)+ ([('a', 0)] @=?) =<< waitEQ eq RequireEvent+ ([('a', 1)] @=?) =<< waitEQ eq RequireEvent+ ]+ , testGroup "registerQueued"+ [ testCase "adds" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, _) <- registerQueued (EEQ eq)+ (1 @?=) =<< numSources eq+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ , testCase "appends" $ do+ eq::STMEQueue Int <- newSTMEQueue+ (add, _) <- registerQueued (EEQ eq)+ add 0+ add 1+ ([0] @=?) =<< waitEQ eq RequireEvent+ ([1] @=?) =<< waitEQ eq RequireEvent+ , testCase "killer removes" $ do+ eq::STMEQueue () <- newSTMEQueue+ (add, k) <- registerQueued (EEQ eq)+ add ()+ ([()] @=?) =<< waitEQ eq RequireEvent+ add ()+ k+ waitEQBlocks eq+ ]+ , testCase "Gets all latest" $ do+ eq::STMEQueue (Either String Int) <- newSTMEQueue+ (addQ, _) <- registerQueued (EEQ eq)+ (addS, _) <- registerSemi (EEQ eq) id+ addQ (Left "a")+ addQ (Left "b")+ addQ (Left "c")+ ([Left "a"] @=?) =<< waitEQ eq RequireEvent+ addS (Right 2)+ addS (Right 1)+ ([Left "b", Right 2] @=?) =<< waitEQ eq RequireEvent+ ([Left "c", Right 1] @=?) =<< waitEQ eq RequireEvent+ ]++mappedTests :: TestTree+mappedTests = testGroup "MappedEQueue Tests"+ [ testGroup "non-mapped"+ [ testCase "Single Edge" $ do+ eq <- (MEQ id) <$> newSTMEQueue+ (add, k) <- registerQueued eq+ add 'a'+ (['a'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Double Edge" $ do+ eq <- (MEQ id) <$> newSTMEQueue+ (add, k) <- registerQueued eq+ add 'a'+ add 'b'+ (['a'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (['b'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Single Level" $ do+ eq <- (MEQ id) <$> newSTMEQueue+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ ([[fromEnum 'a']] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Double Level" $ do+ eq <- (MEQ id) <$> newSTMEQueue+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ add "b"+ ([[fromEnum 'a', fromEnum 'b']] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ ]+ , testGroup "mapped"+ [ testCase "Single Edge" $ do+ eq <- (contramap fromEnum . MEQ id) <$> newSTMEQueue+ (add, k) <- registerQueued eq+ add 'a'+ ([fromEnum 'a'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Double Edge" $ do+ eq <- (contramap fromEnum . MEQ id) <$> newSTMEQueue+ (add, k) <- registerQueued eq+ add 'a'+ add 'b'+ ([fromEnum 'a'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ ([fromEnum 'b'] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Single Level" $ do+ eq <- (contramap (fmap fromEnum) . MEQ id) <$> newSTMEQueue+ (add, k) <- registerSemi eq ("+"++)+ add "a"+ ([fromEnum <$> "+a"] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ , testCase "Double Level" $ do+ eq <- (contramap (fmap fromEnum) . MEQ id) <$> newSTMEQueue+ (add, k) <- registerSemi eq ("+"++)+ add "a"+ add "b"+ ([fromEnum <$> "+ab"] @=?) =<< waitEQ (meqEQ eq) ReturnImmediate+ (() @=?) =<< k+ ]+ ]++simpleTests :: (EQueue eq, EQueueW eq, JustOneEventually ~ WaitPolicy eq)+ => TestName -> (forall a . IO (eq a)) -> TestTree+simpleTests nm new = testGroup nm+ [ testCase "Single Edge" $ do+ eq <- new+ (add, k) <- registerQueued eq+ add 'a'+ (['a'] @=?) =<< waitEQ eq JustOneEventually+ (() @=?) =<< k+ , testCase "Double Edge" $ do+ eq <- new+ (add, k) <- registerQueued eq+ add 'a'+ add 'b'+ (['a'] @=?) =<< waitEQ eq JustOneEventually+ (['b'] @=?) =<< waitEQ eq JustOneEventually+ (() @=?) =<< k+ , testCase "Single Level" $ do+ eq <- new+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ ([[fromEnum 'a']] @=?) =<< waitEQ eq JustOneEventually+ (() @=?) =<< k+ , testCase "Double Level" $ do+ eq <- new+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ add "b"+ ([[fromEnum 'a']] @=?) =<< waitEQ eq JustOneEventually+ ([[fromEnum 'b']] @=?) =<< waitEQ eq JustOneEventually+ (() @=?) =<< k+ ]++ioTests :: TestTree+ioTests = testGroup "IOEQueue Tests"+ [ testCase "Single Edge" $ do+ (eq, w) <- newIOEQ+ (add, k) <- registerQueued eq+ add 'a'+ (['a'] @=?) =<< w+ (() @=?) =<< k+ , testCase "Double Edge" $ do+ (eq, w) <- newIOEQ+ (add, k) <- registerQueued eq+ add 'a'+ add 'b'+ (['a'] @=?) =<< w+ (['b'] @=?) =<< w+ (() @=?) =<< k+ , testCase "Single Level" $ do+ (eq, w) <- newIOEQ+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ ([[fromEnum 'a']] @=?) =<< w+ (() @=?) =<< k+ , testCase "Double Level" $ do+ (eq, w) <- newIOEQ+ (add, k) <- registerSemi eq (fmap fromEnum)+ add "a"+ add "b"+ ([[fromEnum 'a']] @=?) =<< w+ ([[fromEnum 'b']] @=?) =<< w+ (() @=?) =<< k+ ]+ where+ newIOEQ :: IO (IOEQueue a, IO [a])+ newIOEQ = ((IOEQ . writeChan) &&& (fmap pure . readChan)) <$> newChan