Switched to new session tracker.

1 files changed, 83 insertions, 44 deletions

diff --git a/src/Network/Tox/AggregateSession.hs b/src/Network/Tox/AggregateSession.hs
index edb897e0..1dd10eef 100644
--- a/src/Network/Tox/AggregateSession.hs
+++ b/src/Network/Tox/AggregateSession.hs
@@ -22,7 +22,6 @@ module Network.Tox.AggregateSession
 
 import Control.Concurrent.STM
 import Control.Concurrent.STM.TMChan
-import Control.Concurrent.Supply
 import Control.Monad
 import Data.Function
 import qualified Data.IntMap.Strict  as IntMap
@@ -47,9 +46,7 @@ import Network.Tox.Crypto.Transport (CryptoMessage (..), pattern KillPacket,
                                      pattern PacketRequest)
 import Network.Tox.DHT.Transport    (key2id)
 import Network.Tox.NodeId           (ToxProgress (..))
-import Network.Tox.Crypto.Handlers
-
-type Session = NetCryptoSession
+import Network.Tox.Session
 
 -- | For each component session, we track the current status.
 data SingleCon = SingleCon
@@ -113,47 +110,94 @@ data KeepAliveEvents = DoTimeout        -- ^ A session timed-out, close it.
                      | DoRequestMissing -- ^ Detect and request lost packets.
                      deriving Enum
 
--- | This function forks a thread to read all packets from the provided
--- 'Session' and forward them to 'contactChannel' for a containing
--- 'AggregateSession'
+-- | This call loops until the provided sesison is closed or times out.  It
+-- monitors the provided (non-empty) priority queue for scheduled tasks (see
+-- 'KeepAliveEvents') to perform for the connection.
+keepAlive :: Session -> TVar (PSQ POSIXTime) -> IO ()
+keepAlive s q = do
+    myThreadId >>= flip labelThread
+        (intercalate "." ["beacon"
+                         , take 8 $ show $ key2id $ sTheirUserKey s
+                         , show $ sSessionID s])
+
+    let outPrint e = dput XNetCrypto $ shows (sSessionID s,sTheirAddr s) $ " <-- " ++ e
+
+        doAlive = do
+            -- outPrint $ "Beacon"
+            sendMessage (sTransport s) () (OneByte PING)
+
+        doRequestMissing = do
+            (ns,nmin) <- sMissingInbound s
+            -- outPrint $ "PacketRequest " ++ show (nmin,ns)
+            sendMessage (sTransport s) () (RequestResend PacketRequest ns)
+
+        re tm again e io = do
+            io
+            atomically $ modifyTVar' q $ PSQ.insert (fromEnum e) tm
+            again
+
+        doEvent again now e = case e of
+            DoTimeout        -> do dput XNetCrypto $ "TIMEOUT: " ++ show (sSessionID s)
+                                   sClose s
+            DoAlive          -> re (now + 10) again e doAlive
+            DoRequestMissing -> re (now + 5)  again e doRequestMissing -- tox-core does this at 1 second intervals
+
+    fix $ \again -> do
+
+    now <- getPOSIXTime
+    join $ atomically $ do
+        Just ( k :-> tm ) <- PSQ.findMin <$> readTVar q
+        return $ if now < tm then threadDelay (toMicroseconds $ tm - now) >> again
+                             else doEvent again now (toEnum k)
+
+-- | This function forks two threads: the 'keepAlive' beacon-sending thread and
+-- a thread to read all packets from the provided 'Session' and forward them to
+-- 'contactChannel' for a containing 'AggregateSession'
 forkSession :: AggregateSession -> Session -> (Status ToxProgress -> STM ()) -> IO ThreadId
 forkSession c s setStatus = forkIO $ do
     myThreadId >>= flip labelThread
         (intercalate "." ["s"
-                         , take 8 $ show $ key2id $ ncTheirPublicKey s
+                         , take 8 $ show $ key2id $ sTheirUserKey s
                          , show $ sSessionID s])
-    tmchan <- atomically $ do
-        tmchan <- newTMChan
-        supply <- readTVar (listenerIDSupply $ ncAllSessions s)
-        let (listenerId,supply') = freshId supply
-        writeTVar (listenerIDSupply $ ncAllSessions s) supply'
-        modifyTVar' (ncListeners s) (IntMap.insert listenerId (0,tmchan))
-        return tmchan
+
+    q <- atomically $ newTVar $ fromList
+            [ fromEnum DoAlive :-> 0
+            , fromEnum DoRequestMissing :-> 0
+            ]
 
     let sendPacket :: CryptoMessage -> STM ()
         sendPacket msg = writeTMChan (contactChannel c) (sSessionID s, msg)
 
-        inPrint e = dput XNetCrypto $ shows (sSessionID s,ncSockAddr s) $ " --> " ++ e
+        inPrint e = dput XNetCrypto $ shows (sSessionID s,sTheirAddr s) $ " --> " ++ e
+
+        bump = do
+            -- inPrint $ "BUMP: " ++ show (sSessionID s)
+            now <- getPOSIXTime
+            atomically $ modifyTVar' q $ PSQ.insert (fromEnum DoTimeout) (now + 15)
 
         onPacket body loop Nothing          = return ()
         onPacket body loop (Just (Left e))  = inPrint e >> loop
         onPacket body loop (Just (Right x)) = body loop x
 
-        awaitPacket body = fix $ (.) (fmap Right <$> atomically (readTMChan tmchan) >>=)
-                               $ onPacket body
+        awaitPacket body = fix $ awaitMessage (sTransport s) . onPacket body
 
     atomically $ setStatus $ InProgress AwaitingSessionPacket
-    atomically $ setStatus Established
-    awaitPacket $ \loop x -> do
-        case msgID x of
-            KillPacket -> return ()
-            _          -> atomically (sendPacket x) >> loop
-
-    atomically $ setStatus Dormant
-
-
-sSessionID :: Session -> Int
-sSessionID s = fromIntegral $ ncSessionId s
+    awaitPacket $ \_ (online,()) -> do
+        when (msgID online /= ONLINE) $ do
+            inPrint $ "Unexpected initial packet: " ++ show (msgID online)
+        atomically $ do setStatus Established
+                        sendPacket online
+        bump
+        beacon <- forkIO $ keepAlive s q
+        awaitPacket $ \awaitNext (x,()) -> do
+            bump
+            case msgID x of
+                PING       -> return ()
+                KillPacket -> sClose s
+                _          -> atomically $ sendPacket x
+            awaitNext
+        atomically $ setStatus Dormant
+        killThread beacon
 
 -- | Add a new session (in 'AwaitingSessionPacket' state) to the
 -- 'AggregateSession'.  If the supplied session is not compatible because it is
@@ -166,8 +210,8 @@ sSessionID s = fromIntegral $ ncSessionId s
 addSession :: AggregateSession -> Session -> IO AddResult
 addSession c s = do
     (result,mcon,replaced) <- atomically $ do
-        let them = ncTheirPublicKey s
-            me   = ncMyPublicKey s
+        let them = sTheirUserKey s
+            me   = toPublic $ sOurKey s
         compat <- checkCompatible me them c
         let result = case compat of
                         Nothing      -> FirstSession
@@ -184,7 +228,7 @@ addSession c s = do
                                 writeTVar (contactSession c) imap'
                                 return (result,Just con,s0)
 
-    mapM_ (destroySession . singleSession) replaced
+    mapM_ (sClose . singleSession) replaced
     forM_ mcon $ \con ->
         forkSession c s $ \progress -> do
             writeTVar (singleStatus con) progress
@@ -203,6 +247,7 @@ addSession c s = do
                     return emap'
             writeTVar (contactEstablished c) emap'
     return result
+
 -- | Information returned from 'delSession'.
 data DelResult = NoSession      -- ^ Contact is completely disconnected.
                | DeletedSession -- ^ Connection removed but session remains active.
@@ -230,11 +275,10 @@ delSession c sid = do
         writeTVar (contactSession c) imap'
         writeTVar (contactEstablished c) emap'
         return ( IntMap.lookup sid imap, IntMap.null imap')
-    mapM_ (destroySession . singleSession) con
+    mapM_ (sClose . singleSession) con
     return $ if r then NoSession
                   else DeletedSession
 
-
 -- | Send a packet to one or all of the component sessions in the aggregate.
 dispatchMessage :: AggregateSession -> Maybe Int -- ^ 'Nothing' to broadcast, otherwise SessionID.
                                     -> CryptoMessage -> IO ()
@@ -242,11 +286,7 @@ dispatchMessage c msid msg = join $ atomically $ do
     imap <- readTVar (contactSession c)
     let go = case msid of Nothing  -> forM_ imap
                           Just sid -> forM_ (IntMap.lookup sid imap)
-    return $ go $ \con -> do
-        eResult <- sendLossless (transportCrypto (ncAllSessions (singleSession con))) (singleSession con) msg
-        case eResult of
-            Left msg -> dput XJabber msg
-            Right pkt -> dput XJabber ("sendLossLess SUCCESS: " ++ show pkt)
+    return $ go $ \con -> sendMessage (sTransport $ singleSession con) () msg
 
 -- | Retry until:
 --
@@ -287,7 +327,6 @@ aggregateStatus c = do
           | not (IntMap.null imap) -> InProgress AwaitingSessionPacket
           | otherwise              -> Dormant
 
-
 -- | Query whether the supplied ToxID keys are compatible with this aggregate.
 --
 --   [ Nothing ]    Any keys would be compatible because there is not yet any
@@ -304,8 +343,8 @@ checkCompatible me them c = do
     imap <- readTVar (contactSession c)
     return $ case IntMap.elems imap of
                     _ | isclosed -> Just False -- All keys are incompatible (closed).
-                    con:_        -> Just $ ncTheirPublicKey (singleSession con)   == them
-                                           && (ncMyPublicKey $ singleSession con) == me
+                    con:_        -> Just $ sTheirUserKey (singleSession con)         == them
+                                           && toPublic (sOurKey $ singleSession con) == me
                     []           -> Nothing
 
 -- | Returns the local and remote keys that are compatible with this aggregate.
@@ -317,6 +356,6 @@ compatibleKeys c = do
     imap <- readTVar (contactSession c)
     return $ case IntMap.elems imap of
                     _ | isclosed -> Nothing -- none.
-                    con:_        -> Just ( ncMyPublicKey $ singleSession con
-                                         , ncTheirPublicKey (singleSession con))
+                    con:_        -> Just ( toPublic (sOurKey $ singleSession con)
+                                         , sTheirUserKey (singleSession con))
                     []           -> Nothing -- any.