3 files changed, 44 insertions, 28 deletions

diff --git a/Announcer.hs b/Announcer.hs
index f0d65656..ad121f13 100644
--- a/Announcer.hs
+++ b/Announcer.hs
@@ -7,8 +7,10 @@
 {-# LANGUAGE NamedFieldPuns             #-}
 {-# LANGUAGE NondecreasingIndentation   #-}
 module Announcer
-    ( Announcer(scheduled)
+    ( Announcer()
     , AnnounceKey
+    , scheduleAbs
+    , scheduleToList
     , packAnnounceKey
     , unpackAnnounceKey
     , forkAnnouncer
@@ -77,10 +79,24 @@ itemStatusNum (ScheduledItem      {}) = 1
 itemStatusNum (StopAnnouncer        ) = 2
 itemStatusNum _                       = error "itemStatusNum not in sync with ScheduledItem declaration."
 
+newtype Schedule = Schedule { unSchedule :: PSQ' AnnounceKey POSIXTime ScheduledItem }
+
+emptySchedule :: Schedule
+emptySchedule = Schedule PSQ.empty
+
+findNextScheduled :: Announcer -> STM (AnnounceKey, POSIXTime, ScheduledItem)
+findNextScheduled announcer = maybe retry return =<< (findMin . unSchedule) <$> readTVar (scheduled announcer)
+
+removeFromSchedule :: Announcer -> (AnnounceKey, POSIXTime, ScheduledItem) -> STM ()
+removeFromSchedule announcer item = modifyTVar' (scheduled announcer) (Schedule . PSQ.delete (key item) . unSchedule)
+
+scheduleToList :: Announcer -> STM [(AnnounceKey, POSIXTime, ScheduledItem)]
+scheduleToList announcer = PSQ.toList . unSchedule <$> readTVar (scheduled announcer)
+
 data Announcer = Announcer
     { -- | The queue of scheduled events.  The priority is the time at which an
       -- event is supposed to occur.
-      scheduled       :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem)
+      scheduled       :: TVar Schedule
       -- | This TVar is False when the Announcer thread has finished.
     , announcerActive :: TVar Bool
       -- | This delay is used to wait until it's time to act on the earliest
@@ -95,7 +111,12 @@ data Announcer = Announcer
 -- immediately.
 scheduleImmediately :: Announcer -> AnnounceKey -> ScheduledItem -> STM ()
 scheduleImmediately announcer k item
-    = modifyTVar' (scheduled announcer) (PSQ.insert' k item 0)
+    = modifyTVar' (scheduled announcer) (Schedule . PSQ.insert' k item 0 . unSchedule)
+
+scheduleAbs :: Announcer -> AnnounceKey -> ScheduledItem -> POSIXTime -> STM ()
+scheduleAbs announcer k item absTime =
+        modifyTVar (scheduled announcer)
+                   (Schedule . PSQ.insert' k item absTime . unSchedule)
 
 -- | Terminate the 'Announcer' thread.  Don't use the Announcer after this.
 stopAnnouncer :: Announcer -> IO ()
@@ -114,34 +135,31 @@ cancel announcer k = do
 forkAnnouncer :: IO Announcer
 forkAnnouncer = do
     delay <- interruptibleDelay
-    announcer <- atomically $ Announcer <$> newTVar PSQ.empty
+    announcer <- atomically $ Announcer <$> newTVar emptySchedule
                                         <*> newTVar True
                                         <*> pure delay
     fork $ announceThread announcer
     return announcer
 
-
 announceThread :: Announcer -> IO ()
 announceThread announcer = do
     myThreadId >>= flip labelThread "announcer"
     fix $ \loop -> do
-    join $ atomically $ do
-        item <- maybe retry return =<< findMin <$> readTVar (scheduled announcer)
-        return $ do
-        now <- getPOSIXTime
-        -- Is it time to do something?
-        if (prio item <= now)
-            then do -- Yes.  Dequeue and handle this event.
-                 action <- atomically $ do
-                     modifyTVar' (scheduled announcer)
-                                 (PSQ.delete (key item))
-                     performScheduledItem announcer now item
-                 -- Are we finished?
-                 mapM_ (>> loop)  -- No? Okay, perform scheduled op and loop.
-                       action
-            else do -- No.  Wait a bit.
-                 startDelay (interrutible announcer) (microseconds $ prio item - now)
-                 loop
+    item <- atomically $ findNextScheduled announcer
+
+    now <- getPOSIXTime
+    -- Is it time to do something?
+    if (prio item <= now)
+        then do -- Yes.  Dequeue and handle this event.
+              action <- atomically $ do
+                  removeFromSchedule announcer item
+                  performScheduledItem announcer now item
+              -- Are we finished?
+              mapM_ (>> loop)  -- No? Okay, perform scheduled op and loop.
+                    action
+        else do -- No.  Wait a bit.
+              startDelay (interrutible announcer) (microseconds $ prio item - now)
+              loop
     -- We're done.  Let 'stopAnnouncer' know that it can stop blocking.
     atomically $ writeTVar (announcerActive announcer) False
 
diff --git a/Announcer/Tox.hs b/Announcer/Tox.hs
index f79f8ee7..60a5a8b1 100644
--- a/Announcer/Tox.hs
+++ b/Announcer/Tox.hs
@@ -116,8 +116,7 @@ newAnnouncement :: STM (IO a)
                          -> POSIXTime
                          -> STM (IO ())
 newAnnouncement checkFin search announce interval = \announcer k now -> do
-        modifyTVar (scheduled announcer)
-                   (PSQ.insert' k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval))
+        scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval)
         return $ void $ fork search
 
 -- time for periodic announce:
@@ -132,8 +131,7 @@ reAnnounce :: STM (IO a)
                     -> STM (IO ())
 reAnnounce checkFin announce interval = \announcer k now -> do
         isfin <- checkFin
-        modifyTVar (scheduled announcer)
-                   (PSQ.insert' k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval))
+        scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval)
         return $ do
             isfin
             hPutStrLn stderr $ "This print avoids negative-time future scheduling. Weird bug. TODO: fix it. "++show now
diff --git a/examples/dhtd.hs b/examples/dhtd.hs
index 7cf3393c..35035372 100644
--- a/examples/dhtd.hs
+++ b/examples/dhtd.hs
@@ -988,8 +988,8 @@ clientSession s@Session{..} sock cnum h = do
         ("a", "") -> cmd0 $ do
                         now <- getPOSIXTime
                         rs <- atomically $ do
-                            as <- readTVar (scheduled $ announcer)
-                            forM (PSQ.toList as) $ \(k,ptm,item) -> do
+                            as <- scheduleToList announcer
+                            forM (as) $ \(k,ptm,item) -> do
                                 kstr <- unpackAnnounceKey announcer k
                                 return [ if ptm==0 then "now"
                                                    else show (ptm - now)