Combination of QueueManagers

1 files changed, 95 insertions, 19 deletions

diff --git a/server/src/Thermoprint/Server/Queue.hs b/server/src/Thermoprint/Server/Queue.hs
index 68dc7a9..69295bb 100644
--- a/server/src/Thermoprint/Server/Queue.hs
+++ b/server/src/Thermoprint/Server/Queue.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FlexibleInstances, FlexibleContexts #-}
 {-# LANGUAGE ViewPatterns #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE DeriveGeneric, DeriveAnyClass #-}
@@ -7,7 +7,9 @@
 module Thermoprint.Server.Queue
        ( Queue(..), QueueEntry(..)
        , HasQueue(..)
-       , QueueManager(..), runQM
+       , QueueManager, QueueManagerM, runQM
+       , intersection, idQM
+       , union, nullQM
        ) where
 
 import           Thermoprint.API (PrintingError(..), Printout)
@@ -15,11 +17,14 @@ import qualified Thermoprint.API as API (JobStatus(..))
 
 import           Thermoprint.Server.Database
 
-import           Data.Sequence (Seq, ViewL(..), viewl)
+import           Data.Sequence (Seq, ViewL(..), viewl, (|>), (<|))
 import qualified Data.Sequence as Seq
+import           Data.Set (Set)
+import qualified Data.Set as Set
 
 import           Data.Time
-import           Data.Time.Clock
+import           Data.ExtendedReal
+import           Data.Fixed
 
 import           Control.DeepSeq (NFData)
 import           Data.Typeable (Typeable)
@@ -31,10 +36,14 @@ import           Control.Monad.State
 
 import           Data.Default.Class
 
+import           Control.Monad
 import           Control.Monad.Morph
 import           Control.Monad.Trans.Compose
+import           Data.Foldable
+import           Data.Function
 
 import           Data.Monoid
+import           Data.Ord
 
 -- | Zipper for 'Seq QueueEntry' with additional support for 'PrintingError' in the section after point
 data Queue = Queue
@@ -44,17 +53,6 @@ data Queue = Queue
   }
   deriving (Typeable, Generic, NFData)
 
-toSeq :: Queue -> Seq (Bool, QueueEntry, Maybe PrintingError)
-toSeq Queue{..} = fmap (\x -> (False, x, Nothing)) pending <> maybe Seq.empty (\c -> Seq.singleton (True, c, Nothing)) current <> fmap (\(x, p) -> (False, x, p)) history
-
-fromSeq :: Seq (Bool, QueueEntry, Maybe PrintingError) -> Queue
-fromSeq s = Queue pending' current' history'
-  where
-    (fmap (\(_, x, _) -> x) -> pending', pending'') = Seq.breakl (\(c, _, _) -> c) s
-    (current', history') = case viewl pending'' of
-      EmptyL -> (Nothing, Seq.empty)
-      (_, a, _) :< as -> (Just a, fmap (\(_, x, p) -> (x, p)) as)
-
 class HasQueue a where
   extractQueue :: a -> TVar Queue
 
@@ -72,10 +70,59 @@ data QueueEntry = QueueEntry
   { jobId :: JobId
   , created :: UTCTime
   }
-  deriving (Typeable, Generic, NFData)
+  deriving (Typeable, Generic, NFData, Eq, Ord)
+
+data QueueItem = Pending Int QueueEntry | Current QueueEntry | History Int QueueEntry (Maybe PrintingError)
+
+instance Eq QueueItem where
+  (Pending i a  ) == (Pending j b  ) = i == j && a == b
+  (Current   a  ) == (Current   b  ) =           a == b
+  (History i a _) == (History j b _) = i == j && a == b
+  _ == _ = False
+
+instance Ord QueueItem where
+  (Pending i a  ) `compare` (Pending j b  ) = compare i j <> compare a b
+  (Current   a  ) `compare` (Current   b  ) = compare a b
+  (History i a _) `compare` (History j b _) = compare i j <> compare a b
+  (Pending _ _  ) `compare` _               = LT
+  (Current   _  ) `compare` (Pending _ _  ) = GT
+  (Current   _  ) `compare` _               = LT
+  (History _ _ _) `compare` _               = GT
+
+newtype PlainQueueItem = Plain { unPlain :: QueueItem }
+
+instance Eq PlainQueueItem where
+  (unPlain -> Pending _ a  ) == (unPlain -> Pending _ b  ) = a == b
+  (unPlain -> Current   a  ) == (unPlain -> Current   b  ) = a == b
+  (unPlain -> History _ a _) == (unPlain -> History _ b _) = a == b
+  _ == _ = False
+
+instance Ord PlainQueueItem where
+  (unPlain -> Pending _ a  ) <= (unPlain -> Pending _ b  ) = a <= b
+  (unPlain -> Current   a  ) <= (unPlain -> Current   b  ) = a <= b
+  (unPlain -> History _ a _) <= (unPlain -> History _ b _) = a <= b
+  (unPlain -> Current _    ) <= (unPlain -> Pending _ _  ) = False
+  (unPlain -> History _ _ _) <= (unPlain -> Pending _ _  ) = False
+  (unPlain -> History _ _ _) <= (unPlain -> Current _    ) = False
+  (unPlain -> Pending _ _  ) <= _                          = True
+  (unPlain -> Current _    ) <= _                          = True
+
+fromZipper :: Queue -> Set QueueItem
+fromZipper Queue{..} = Set.fromList . toList $ mconcat [ Seq.mapWithIndex Pending pending
+                                                       , maybe Seq.empty (Seq.singleton . Current) current
+                                                       , Seq.mapWithIndex (\i (a, e) -> History i a e) history
+                                                       ]
+
+toZipper :: Set QueueItem -> Queue
+toZipper = Set.foldr' insert def
+  where
+    insert (Pending _ a) q@(Queue{..}) = q { pending = pending |> a }
+    insert (Current a) q = q { current = Just a }
+    insert (History _ a e) q@(Queue{..}) = q { history = history |> (a, e) }
 
 -- | A queue manager periodically modifies a 'Queue', e.g. for cleanup of old jobs
-type QueueManager t = ComposeT (StateT Queue) t STM DiffTime
+type QueueManager t = QueueManagerM t (Extended Micro)
+type QueueManagerM t = ComposeT (StateT Queue) t STM
 
 runQM :: ( HasQueue q
          , MFunctor t
@@ -84,7 +131,36 @@ runQM :: ( HasQueue q
          , Monad (t STM)
          ) => QueueManager t -> q -> t IO ()
 -- ^ Periodically modify a 'Queue'
-runQM qm (extractQueue -> q) = forever $ liftIO . threadDelay . toMicro =<< qm'
+runQM qm (extractQueue -> q) = sleep =<< qm'
   where
     qm' = hoist atomically $ (\(a, s) -> lift (writeTVar q s) >> return a) =<< runStateT (getComposeT qm) =<< lift (readTVar q)
-    toMicro = (`div` 10^6) . fromEnum
+    sleep (abs -> delay)
+      | (Finite d) <- delay = liftIO (threadDelay $ fromEnum d) >> runQM qm q
+      | otherwise = return ()
+
+intersection :: (Foldable f, MonadState Queue (QueueManagerM t)) => f (QueueManager t) -> QueueManager t
+-- ^ Combine two 'QueueManager's keeping only 'QueueEntry's both managers decide to keep
+intersection = foldr' (qmCombine Set.intersection) idQM
+
+idQM :: Monad (QueueManagerM t) => QueueManager t
+-- ^ Identity of 'intersect'
+idQM = return PosInf
+
+union :: (Foldable f, MonadState Queue (QueueManagerM t)) => f (QueueManager t) -> QueueManager t
+-- ^ Combine two 'QueueManager's keeping all 'QueueEntry's either of the managers decides to keep
+union = foldr' (qmCombine Set.union) nullQM
+
+nullQM :: MonadState Queue (QueueManagerM t) => QueueManager t
+-- ^ Identity of 'union'
+nullQM = put def >> return PosInf
+
+qmCombine :: MonadState Queue (QueueManagerM t)
+             => (Set PlainQueueItem -> Set PlainQueueItem -> Set PlainQueueItem)
+             -> (QueueManager t -> QueueManager t -> QueueManager t)
+qmCombine setCombine a b = do
+  (d1, s1) <- local a
+  (d2, s2) <- local b
+  put . toZipper . Set.map unPlain $ on setCombine (Set.map Plain . fromZipper) s1 s2
+  return $ min d1 d2
+  where
+    local x = ((,) <$> get <*> ((,) <$> x <*> get)) >>= (\(oldS, r) -> r <$ put oldS)