Announce command now includes a recurring search.

1 files changed, 44 insertions, 18 deletions

diff --git a/Announcer.hs b/Announcer.hs
index 2f0eca10..668e00c2 100644
--- a/Announcer.hs
+++ b/Announcer.hs
@@ -29,6 +29,7 @@ import qualified Data.ByteString.Char8      as Char8
 import Data.Function
 import Data.Hashable
 import Data.Maybe
+import Data.Ord
 import Data.Time.Clock.POSIX
 
 newtype AnnounceKey = AnnounceKey ByteString
@@ -41,11 +42,10 @@ unpackAnnounceKey :: AnnounceKey -> AnnounceKey -> STM String
 unpackAnnounceKey _ (AnnounceKey bs) = return $ Char8.unpack bs
 
 data ScheduledItem
-    = forall r. ScheduledItem (AnnounceMethod r)
-    | StopAnnouncer
-    | NewAnnouncement (IO ()) (IO ()) POSIXTime
+    = StopAnnouncer
+    | NewAnnouncement (STM (IO ())) (IO ()) (IO ()) POSIXTime
     | SearchFinished (IO ()) (IO ()) POSIXTime
-    | Announce (IO ()) POSIXTime
+    | Announce (STM (IO ())) (IO ()) POSIXTime
     | DeleteAnnouncement
 
 data Announcer = Announcer
@@ -54,6 +54,14 @@ data Announcer = Announcer
     , interrutible    :: InterruptibleDelay
     }
 
+announceK :: Int
+announceK = 8
+
+data AnnounceState = forall nid addr tok ni r. AnnounceState
+    { aState        :: SearchState nid addr tok ni r
+    , aStoringNodes :: TVar (MM.MinMaxPSQ ni (Down POSIXTime))
+    }
+
 scheduleImmediately :: Announcer -> AnnounceKey -> ScheduledItem -> STM ()
 scheduleImmediately announcer k item
     = modifyTVar' (scheduled announcer) (PSQ.insert' k item 0)
@@ -82,22 +90,39 @@ data AnnounceMethod r = forall nid ni sr addr tok a.
 schedule :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
 schedule announcer k AnnounceMethod{aSearch,aPublish,aBuckets,aTarget,aInterval} r = do
     st <- atomically $ newSearch aSearch aTarget []
-    let announce = do -- publish to current search results
+    ns <- atomically $ newTVar MM.empty
+    let astate = AnnounceState st ns
+        publishToNodes is = do
+            forM_ is $ \(Binding ni mtok _) -> do
+            forM_ mtok $ \tok -> do
+                got <- aPublish r tok (Just ni)
+                now <- getPOSIXTime
+                forM_ got $ \_ -> do
+                    atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now)
+        announce = do -- publish to current search results
             is <- atomically $ do
                 bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -})
                 return $ MM.toList bs
-            forM_ is $ \(Binding ni mtok _) -> do
-            forM_ mtok $ \tok -> do
-            aPublish r tok (Just ni)
-            return ()
+            publishToNodes is
         onResult _ = return True -- action for each search-hit (True = keep searching)
+        searchAgain = searchIsFinished st >>= \isfin -> return $ when isfin $ void $ fork search
         search  = do -- thread to fork
             atomically $ reset aBuckets aSearch aTarget st
             searchLoop aSearch aTarget onResult st
+            fork $ do -- Announce to any nodes we haven't already announced to.
+                is <- atomically $ do
+                    bs <- readTVar (searchInformant st {- :: TVar (MinMaxPSQ' ni nid tok -})
+                    nq <- readTVar ns
+                    return $ filter (\(Binding ni _ _) -> not $ isJust $ MM.lookup' ni nq)
+                           $ MM.toList bs
+                publishToNodes is
+            return ()
+            {-
             atomically $ scheduleImmediately announcer k
                          $ SearchFinished {- st -} search announce aInterval
             interruptDelay (interrutible announcer)
-    atomically $ scheduleImmediately announcer k $ NewAnnouncement search announce aInterval
+            -}
+    atomically $ scheduleImmediately announcer k $ NewAnnouncement searchAgain search announce aInterval
     interruptDelay (interrutible announcer)
 
 cancel :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
@@ -145,12 +170,10 @@ performScheduledItem announcer now = \case
     (Binding _ StopAnnouncer _) ->  return Nothing
 
     -- announcement started:
-    (Binding k (NewAnnouncement search announce interval) _) -> do
+    (Binding k (NewAnnouncement checkFin search announce interval) _) -> do
         modifyTVar (scheduled announcer)
-                   (PSQ.insert' k (Announce announce interval) (now + interval))
-        return $ Just $ do
-            fork search
-            return ()
+                   (PSQ.insert' k (Announce checkFin announce interval) (now + interval))
+        return $ Just $ void $ fork search
 
     -- announcement removed:
     (Binding k DeleteAnnouncement _) -> return $ Just $ return ()
@@ -158,10 +181,13 @@ performScheduledItem announcer now = \case
     -- time for periodic announce:
     --   (re-)announce to the current known set of storing-nodes.
     --   TODO: If the search is finished, restart the search.
-    (Binding k (Announce announce interval) _) -> do
+    (Binding k (Announce checkFin announce interval) _) -> do
+        isfin <- checkFin
         modifyTVar (scheduled announcer)
-                   (PSQ.insert' k (Announce announce interval) (now + interval))
-        return $ Just announce
+                   (PSQ.insert' k (Announce checkFin announce interval) (now + interval))
+        return $ Just $ do
+            isfin
+            announce
 
     -- search finished:
     --   if any of the current storing-nodes set have not been