Factor out some new libraries

	word64-map:
		Data.Word64Map
	network-addr:
		Network.Address
	tox-crypto:
		Crypto.Tox
	lifted-concurrent:
		Control.Concurrent.Lifted.Instrument
		Control.Concurrent.Async.Lifted.Instrument
	psq-wrap:
		Data.Wrapper.PSQInt
		Data.Wrapper.PSQ
	minmax-psq:
		Data.MinMaxPSQ
        tasks:
                Control.Concurrent.Tasks
	kad:
		Network.Kademlia
		Network.Kademlia.Bootstrap
		Network.Kademlia.Routing
		Network.Kademlia.CommonAPI
		Network.Kademlia.Persistence
		Network.Kademlia.Search

1 files changed, 206 insertions, 0 deletions

diff --git a/dht/Announcer/Tox.hs b/dht/Announcer/Tox.hs
new file mode 100644
index 00000000..f8343f8d
--- /dev/null
+++ b/dht/Announcer/Tox.hs
@@ -0,0 +1,206 @@
+{-# LANGUAGE DeriveDataTypeable         #-}
+{-# LANGUAGE DeriveGeneric              #-}
+{-# LANGUAGE ExistentialQuantification  #-}
+{-# LANGUAGE FlexibleContexts           #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE LambdaCase                 #-}
+{-# LANGUAGE NamedFieldPuns             #-}
+{-# LANGUAGE NondecreasingIndentation   #-}
+module Announcer.Tox where
+    -- , AnnounceMethod(..)
+    -- , schedule
+
+import Announcer
+import qualified Data.MinMaxPSQ as MM
+import Data.Wrapper.PSQ         as PSQ
+import Network.Kademlia.Search
+
+import Control.Concurrent.Lifted.Instrument
+import Control.Concurrent.STM
+import Control.Monad
+import Data.Hashable
+import Data.Maybe
+import Data.Ord
+import Data.Time.Clock.POSIX
+
+
+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))
+    }
+
+-- | This type specifies an item that can be announced on appropriate nodes in
+-- a Kademlia network.
+data AnnounceMethod r = forall nid ni sr addr tok a.
+                        ( Show nid
+                        , Hashable nid
+                        , Hashable ni
+                        , Ord addr
+                        , Ord nid
+                        , Ord ni
+                        ) => AnnounceMethod
+    { aSearch   :: Search nid addr tok ni sr
+                -- ^ This is the Kademlia search to run repeatedly to find the
+                -- nearby nodes.  A new search is started whenever one is not
+                -- already in progress at announce time.  Repeated searches are
+                -- likely to finish faster than the first since nearby nodes
+                -- are not discarded.
+    , aPublish  :: r -> tok -> Maybe ni -> IO (Maybe a)
+                -- ^ The action to perform when we find nearby nodes.  The
+                -- destination node is given as a Maybe so that methods that
+                -- treat 'Nothing' as loop-back address can be passed here,
+                -- however 'Nothing' will not be passed by the announcer
+                -- thread.
+                --
+                -- The action requires a "token" from the destination
+                -- node.  This is the more typical "announce" semantics for
+                -- Kademlia.
+    , aNearestNodes :: nid -> STM [ni]
+                -- ^ Method to obtain starting nodes from an iterative Kademlia search.
+    , aTarget   :: nid
+                -- ^ This is the Kademlia node-id of the item being announced.
+    , aInterval :: POSIXTime
+                -- ^ Assuming we have nearby nodes from the search, the item
+                -- will be announced at this interval.
+                --
+                -- Current implementation is to make the scheduled
+                -- announcements even if the search hasn't finished.  It will
+                -- use the closest nodes found so far.
+    }
+
+-- | This type specifies a Kademlia search and an action to perform upon the result.
+data SearchMethod r = forall nid ni sr addr tok a.
+                        ( Show nid
+                        , Hashable nid
+                        , Hashable ni
+                        , Ord addr
+                        , Ord nid
+                        , Ord ni
+                        ) => SearchMethod
+    { sSearch   :: Search nid addr tok ni sr
+                -- ^ This is the Kademlia search to run repeatedly to find the
+                -- nearby nodes.  A new search is started whenever one is not
+                -- already in progress at announce time.  Repeated searches are
+                -- likely to finish faster than the first since nearby nodes
+                -- are not discarded.
+                --
+                -- XXX: Currently, "repeatedly" is wrong.
+    , sWithResult :: r -> sr  -> IO ()
+                -- ^
+                -- The action to perform upon a search result.  This was
+                -- implemented to support Tox's DHTKey and Friend-Request
+                -- messages.
+    , sNearestNodes :: nid -> STM [ni]
+                -- ^ Method to obtain starting nodes from an iterative Kademlia search.
+    , sTarget   :: nid
+                -- ^ This is the Kademlia node-id of the item being announced.
+    , sInterval :: POSIXTime
+                -- ^ The time between searches.
+                --
+                -- XXX: Currently, search results will stop any repetition.
+    }
+
+
+-- announcement started:
+newAnnouncement :: STM (IO a)
+                         -> IO ()
+                         -> IO ()
+                         -> POSIXTime
+                         -> Announcer
+                         -> AnnounceKey
+                         -> POSIXTime
+                         -> STM (IO ())
+newAnnouncement checkFin search announce interval = \announcer k now -> do
+        scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval)
+        return $ void $ fork search
+
+-- time for periodic announce:
+--   (re-)announce to the current known set of storing-nodes.
+--   TODO: If the search is finished, restart the search.
+reAnnounce :: STM (IO a)
+                    -> IO ()
+                    -> POSIXTime
+                    -> Announcer
+                    -> AnnounceKey
+                    -> POSIXTime
+                    -> STM (IO ())
+reAnnounce checkFin announce interval = \announcer k now -> do
+        isfin <- checkFin
+        scheduleAbs announcer k (ScheduledItem $ reAnnounce checkFin announce interval) (now + interval)
+        return $ do
+            isfin
+            announce
+
+-- | Schedule a recurring Search + Announce sequence.
+scheduleAnnounce :: Announcer -> AnnounceKey -> AnnounceMethod r -> r -> IO ()
+scheduleAnnounce  announcer k AnnounceMethod{aSearch,aPublish,aNearestNodes,aTarget,aInterval} r = do
+    st <- atomically $ newSearch aSearch aTarget []
+    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
+            publishToNodes is
+        onResult sr = return True
+        searchAgain = do
+            -- Canceling a pending search here seems to make announcements more reliable.
+            searchCancel st
+            isfin <- searchIsFinished st -- Always True, since we canceled.
+            return $ when isfin $ void $ fork search
+        search  = do -- thread to fork
+            atomically $ reset aNearestNodes 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 $ ScheduledItem (newAnnouncement searchAgain search announce aInterval)
+
+-- | Schedule a recurring Search + Publish sequence.
+scheduleSearch :: Announcer -> AnnounceKey -> SearchMethod r -> r -> IO ()
+scheduleSearch announcer k SearchMethod{sSearch,sWithResult,sNearestNodes,sTarget,sInterval} r = do
+    st <- atomically $ newSearch sSearch sTarget []
+    ns <- atomically $ newTVar MM.empty
+    let astate = AnnounceState st ns
+        onResult sr = do
+                runAction announcer $ do
+                    got <- sWithResult r sr
+                    -- If we had a way to get the source of a search result, we might want to
+                    -- treat it similarly to an announcing node and remember it in the 'aStoringNodes'
+                    -- MinMaxPSQ.  For now, I'm just letting the nodes for which we've already sent
+                    -- a message be forgotten.
+                    --
+                    -- forM_ got $ \_ -> do
+                    --    atomically $ modifyTVar ns $ MM.insertTake announceK ni (Down now)
+                    return ()
+                return True -- True to keep searching.
+        searchAgain = do
+            -- Canceling a pending search here seems to make announcements more reliable.
+            searchCancel st
+            isfin <- searchIsFinished st -- Always True, since we canceled.
+            return $ when isfin $ void $ fork search
+        search  = do -- thread to fork
+            atomically $ reset sNearestNodes sSearch sTarget st
+            searchLoop sSearch sTarget onResult st
+    atomically $ scheduleImmediately announcer k $ ScheduledItem (newAnnouncement searchAgain search (return ()) sInterval)
+