tox: freshenContact WIP.

2 files changed, 40 insertions, 18 deletions

diff --git a/Connection/Tox.hs b/Connection/Tox.hs
index 2b39ef1c..42a64a6d 100644
--- a/Connection/Tox.hs
+++ b/Connection/Tox.hs
@@ -6,19 +6,21 @@ module Connection.Tox
     , ToxProgress(..)
     ) where
 
-import qualified Connection          as G
-         ;import Connection          (Manager (..), Policy (..))
+import qualified Connection           as G
+         ;import Connection           (Manager (..), Policy (..))
 import Connection.Tox.Threads
 import Control.Concurrent.STM
 import Control.Monad
 import Crypto.Tox
-import qualified Data.HashMap.Strict as HashMap
-import qualified Data.Map            as Map
+import qualified Data.HashMap.Strict  as HashMap
+import qualified Data.Map             as Map
 import Data.Maybe
+import Network.Kademlia.Routing       as R
+import Network.Kademlia.Search
 import Network.Tox.ContactInfo
 import Network.Tox.Crypto.Handlers
-import Network.Tox.DHT.Handlers      as DHT
-import Network.Tox.DHT.Transport     as DHT
+import Network.Tox.DHT.Handlers       as DHT
+import Network.Tox.DHT.Transport      as DHT
 import PingMachine
 import Text.Read
 #ifdef THREAD_DEBUG
@@ -147,14 +149,20 @@ setToxPolicy params conmap k@(Key me them) policy = case policy of
                 , longRetryInterval  = _todo
                 , contact            = c
                 }
+            sch = nodeSearch (dhtClient params) (nodesOfInterest $ dhtRouting params)
             freshen_methods = FreshenContactMethods
                 { dhtkeyInterval   = _todo :: Int
                 , sockAddrInterval = _todo :: Int
-                , nodeSch          = nodeSearch (dhtClient params)
-                                                (nodesOfInterest $ dhtRouting params)
+                , nodeSch          = sch
                 , getDHTKey        = retry :: STM (Maybe NodeId)
                 , getSockAddr      = retry -- :: STM (Maybe SockAddr)
-                , getBuckets       = retry -- :: STM (BucketList NodeInfo)
+                , nearestNodes     = \nid -> do
+                    bkts4 <- readTVar $ routing4 $ dhtRouting params
+                    bkts6 <- readTVar $ routing6 $ dhtRouting params
+                    let interweave []     ys = ys
+                        interweave (x:xs) ys = x : interweave ys xs
+                    return $ interweave (R.kclosest (searchSpace sch) searchK nid bkts4)
+                                        (R.kclosest (searchSpace sch) searchK nid bkts6)
                 }
             get_status = do
                 sbk <- readTVar $ netCryptoSessionsByKey (sessions params)
@@ -187,7 +195,7 @@ setToxPolicy params conmap k@(Key me them) policy = case policy of
             return ()
     RefusingToConnect -> do -- disconnect or cancel any pending connection
         mst <- lookupForPolicyChange conmap k policy
-        -- Since the 3 connection threads poll the current policy, they should
+        -- Since the connection threads poll the current policy, they should
         -- all terminate on their own.
         --
         -- Here we block until they finish.
@@ -205,6 +213,8 @@ setToxPolicy params conmap k@(Key me them) policy = case policy of
                     _                     -> retry
     OpenToConnect -> do -- passively accept connections if they initiate.
         mst <- lookupForPolicyChange conmap k policy
+        r <- atomically $ lookupContact k (roster params)
+        forM_ r $ \(sec,c) -> do
         forM_ mst $ \st -> do
             {-
             let getPolicy = readTVar $ connPolicy st
@@ -218,8 +228,12 @@ setToxPolicy params conmap k@(Key me them) policy = case policy of
                 registerNodeCallback routing $ NodeInfoCallback
                     { interestingNodeId = nid
                     , listenerId        = callbackId
-                    , observedAddress   = \ni -> return () -- TODO
-                    , rumoredAddress    = \saddr ni -> return () -- TODO
+                    , observedAddress   = \ni -> writeTVar (contactLastSeenAddr c) (Just $ nodeAddr ni)
+                    , rumoredAddress    = \saddr ni -> do
+                        m <- readTVar (contactLastSeenAddr c)
+                        -- TODO remember information source and handle multiple rumors.
+                        case m of Just _  -> return ()
+                                  Nothing -> writeTVar (contactLastSeenAddr c) (Just $ nodeAddr ni)
                     }
 
 stringToKey_ :: String -> Maybe Key
diff --git a/Connection/Tox/Threads.hs b/Connection/Tox/Threads.hs
index 73c83338..5602fa40 100644
--- a/Connection/Tox/Threads.hs
+++ b/Connection/Tox/Threads.hs
@@ -8,6 +8,7 @@
 --
 --  * 'freshenContact'
 --
+{-# LANGUAGE CPP        #-}
 {-# LANGUAGE LambdaCase #-}
 module Connection.Tox.Threads where
 
@@ -24,6 +25,12 @@ import Network.Tox.DHT.Transport as DHT (dhtpk)
 import Network.Socket
 import Network.Kademlia.Search
 import Network.Kademlia.Routing (BucketList)
+#ifdef THREAD_DEBUG
+import Control.Concurrent.Lifted.Instrument
+#else
+import Control.Concurrent.Lifted
+import GHC.Conc                  (labelThread)
+#endif
 
 import Control.Concurrent.STM
 import Control.Monad
@@ -169,7 +176,7 @@ data FreshenContactMethods = FreshenContactMethods
     , nodeSch          :: NodeSearch
     , getDHTKey        :: STM (Maybe NodeId)
     , getSockAddr      :: STM (Maybe SockAddr)
-    , getBuckets       :: STM (BucketList NodeInfo)
+    , nearestNodes     :: NodeId -> STM [NodeInfo]
     }
 
 -- send my dht key
@@ -206,12 +213,13 @@ freshenContact getPolicy getStatus FreshenContactMethods{..} statusVar
                     Nothing -> do -- AcquiringIPAddress
                            writeTVar statusVar (InProgress AcquiringIPAddress)
                            return $
-                               do bkts <- atomically $ getBuckets
-                                  st <- search nodeSch bkts dk $
-                                            \r -> do -- TODO: store saddr, check for finish
-                                                     return True
+                               do st <- atomically $ do
+                                    ns <- nearestNodes dk
+                                    newSearch nodeSch dk ns
+                                  -- forked simply to avoid relabeling this thread.
+                                  forkIO $ searchLoop nodeSch dk (const $ return True) st
+                                  -- TODO: searchCancel on stop condition
                                   atomically $ searchIsFinished st >>= check
-                                               -- TODO: searchCancel on stop condition
                                   retryAfterTimeout sockAddrInterval
                     Just a  -> do
                             writeTVar statusVar (InProgress AcquiringCookie)