diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# Revision history for equeue + +## 0 -- YYYY-mm-dd + +* First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/equeue.cabal b/equeue.cabal new file mode 100644 --- /dev/null +++ b/equeue.cabal @@ -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 diff --git a/src/Control/Concurrent/EQueue.hs b/src/Control/Concurrent/EQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/EQueue.hs @@ -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 diff --git a/src/Control/Concurrent/EQueue/Class.hs b/src/Control/Concurrent/EQueue/Class.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/EQueue/Class.hs @@ -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] diff --git a/src/Control/Concurrent/EQueue/STMEQueue.hs b/src/Control/Concurrent/EQueue/STMEQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/EQueue/STMEQueue.hs @@ -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 diff --git a/src/Control/Concurrent/EQueue/Simple.hs b/src/Control/Concurrent/EQueue/Simple.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/EQueue/Simple.hs @@ -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 ()) diff --git a/test/test.hs b/test/test.hs new file mode 100644 --- /dev/null +++ b/test/test.hs @@ -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