Bug fix to Announcer.

1 files changed, 31 insertions, 20 deletions

diff --git a/Announcer.hs b/Announcer.hs
index a9b30940..4f4eba8f 100644
--- a/Announcer.hs
+++ b/Announcer.hs
@@ -16,7 +16,7 @@ module Announcer
 
 import Data.Wrapper.PSQ as PSQ
 import Network.Kademlia.Search
-import Ticker
+import InterruptibleDelay
 
 import Control.Concurrent.Lifted.Instrument
 import Control.Concurrent.STM
@@ -42,10 +42,9 @@ data ScheduledItem
     | StopAnnouncer
 
 data Announcer = Announcer
-    { scheduled :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem)
+    { scheduled       :: TVar (PSQ' AnnounceKey POSIXTime ScheduledItem)
     , announcerActive :: TVar Bool
-    , lastTick :: TVar POSIXTime
-    , announceTicker :: Ticker
+    , interrutible    :: InterruptibleDelay
     }
 
 scheduleImmediately :: Announcer -> ScheduledItem -> STM ()
@@ -62,25 +61,27 @@ data AnnounceMethod r = forall nid ni addr r tok a. AnnounceMethod
     , aPublish :: r -> tok -> Maybe ni -> IO (Maybe a)
     }
 
+-- startDelay :: InterruptibleDelay -> Microseconds -> IO Bool
+-- interruptDelay :: InterruptibleDelay -> IO ()
+
 schedule :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
-schedule _ _ _ _ = do
+schedule announcer _ _ _ = do
     -- fork the search
     -- add it to the priority queue of announce methods.
-    -- update ticker
-    return ()
+    interruptDelay (interrutible announcer)
 
 cancel :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
-cancel _ _ _ _ = return ()
+cancel announcer _ _ _ = do
+    -- cancel search/announce
+    interruptDelay (interrutible announcer)
 
 
 forkAnnouncer :: IO Announcer
 forkAnnouncer = do
-    tickvar <- atomically $ newTVar 0
-    ticker <- forkTicker $ writeTVar tickvar
+    delay <- interruptibleDelay
     announcer <- atomically $ Announcer <$> newTVar PSQ.empty
                                         <*> newTVar True
-                                        <*> pure tickvar
-                                        <*> pure ticker
+                                        <*> pure delay
     fork $ announceThread announcer
     return announcer
 
@@ -89,14 +90,24 @@ announceThread :: Announcer -> IO ()
 announceThread announcer = do
     myThreadId >>= flip labelThread "announcer"
     fix $ \loop -> do
-    action <- atomically $ do
-        now <- readTVar $ lastTick announcer
-        (item,q) <- readTVar (scheduled announcer)
-                    >>= maybe retry return . PSQ.minView
-        when (prio item > now) retry        -- Is it time to do something?
-        writeTVar (scheduled announcer) q   -- Remove the event from the queue.
-        performScheduledItem announcer item -- Go for it!
-    mapM_ (>> loop) action
+    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 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
 
 performScheduledItem :: Announcer -> Binding' AnnounceKey POSIXTime ScheduledItem -> STM (Maybe (IO ()))