dhtd & Network.Tox.Crypto.Handlers changes:

        * merge PerSession into NetCryptoSession
        * define defaultUnRecHook function
        * netcrypto command wip for testing
        * update to sessions command

1 files changed, 59 insertions, 2 deletions

diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index 0e349196..602b14cc 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -10,6 +10,7 @@ import Network.Tox.DHT.Transport (Cookie(..),CookieData(..), CookieRequest(..),
 import Network.Tox.DHT.Handlers (Client, cookieRequest, createCookie )
 import Crypto.Tox
 import Control.Concurrent.STM
+import Control.Concurrent.STM.TMChan
 import Network.Address
 import qualified Data.Map.Strict as Map
 import Crypto.Hash
@@ -37,6 +38,8 @@ import System.Random -- for ping fuzz
 import Control.Concurrent
 import GHC.Conc (labelThread)
 import PingMachine
+import qualified Data.IntMap.Strict as IntMap
+import Control.Concurrent.Supply
 
 -- util, todo: move to another module
 maybeToEither :: Maybe b -> Either String b
@@ -83,6 +86,20 @@ data SessionView = SessionView
 
 type SessionID = Word64
 
+-- | Application specific listener type (Word64)
+--
+-- This is some kind of information associated with a listening TChan.
+-- It may be used to indicate what kind of packets it is interested in.
+--
+-- 0 means listen to all messages and is done automatically in 'defaultUnRecHook'
+-- any other values are left open to application specific convention.
+--
+-- This module does not know what the different values here
+-- mean, but code that sets hooks may adhere to a convention
+-- defined elsewhere.
+--
+type ListenerType = Word64
+
 data NetCryptoSession = NCrypto
     { ncState              :: TVar NetCryptoSessionStatus
     , ncMyPublicKey        :: PublicKey
@@ -116,6 +133,18 @@ data NetCryptoSession = NCrypto
                                                 CryptoMessage
                                                 (CryptoPacket Encrypted)
                                                 CryptoData
+    , ncLastNMsgs :: PacketQueue (Bool{-Handled?-},CryptoMessage)
+    -- ^ cyclic buffer, holds the last N non-handshake crypto messages
+    --   even if there is no attached user interface.
+    , ncListeners :: TVar (IntMap.IntMap (ListenerType,TMChan CryptoMessage))
+    -- ^ user interfaces may "listen" by inserting themselves into this map
+    -- with a unique id and a new TChan, and then reading from the TChan
+    , ncMsgNumVar  :: TVar  Word32
+    -- ^ The number of non-handshake crypto messages recieved in this session
+    --   TODO: there is already a packet num etc, do we need two?
+    , ncDropCntVar :: TVar  Word32
+    -- ^ The number of crypto messages that were overwritten in the ncLastNMsgs
+    --   before anybody got to see them.
     }
 
 data NetCryptoSessions = NCSessions
@@ -131,6 +160,7 @@ data NetCryptoSessions = NCSessions
     , nextSessionId           :: TVar SessionID
     , announceNewSessionHooks :: TVar [IOHook (Maybe NoSpam) NetCryptoSession]
     , sessionTransport        :: Transport String SockAddr NetCrypto
+    , listenerIDSupply         :: TVar Supply
     }
 
 type NewSessionHook = IOHook (Maybe NoSpam) NetCryptoSession
@@ -173,6 +203,8 @@ newSessionsState crypto unrechook hooks = do
     svDownloadDir0 <- atomically $ newTVar (homedir </> "Downloads")
     nextSessionId0 <- atomically $ newTVar 0
     announceNewSessionHooks0 <- atomically $ newTVar []
+    lsupply <- newSupply
+    lsupplyVar <- atomically (newTVar lsupply)
     return NCSessions { netCryptoSessions       = x
                       , netCryptoSessionsByKey  = x2
                       , transportCrypto         = crypto
@@ -195,6 +227,7 @@ newSessionsState crypto unrechook hooks = do
                       , nextSessionId           = nextSessionId0
                       , announceNewSessionHooks = announceNewSessionHooks0
                       , sessionTransport = error "Need to set sessionTransport field of NetCryptoSessions!"
+                      , listenerIDSupply = lsupplyVar
                       }
 
 data HandshakeParams
@@ -338,6 +371,10 @@ freshCryptoSession  sessions
                 writeTVar ncMyPacketNonce0 n24plus1
                 return (return (f n24, n24, ncOutgoingIdMap0))
     pktoq <- atomically $ PQ.newOutGoing pktq ncToWire toWireIO 0 (outboundQueueCapacity sessions) 0
+    msgQ <- atomically (PQ.newOverwrite 10 0 :: STM (PacketQueue (Bool,CryptoMessage)))
+    listeners <- atomically $ newTVar IntMap.empty
+    msgNum <- atomically $ newTVar 0
+    dropNum <- atomically $ newTVar 0
     let netCryptoSession0 =
             NCrypto { ncState              = ncState0
                     , ncMyPublicKey        = toPublic key
@@ -362,6 +399,10 @@ freshCryptoSession  sessions
                     , ncDequeueThread      = Nothing -- error "you want the NetCrypto-Dequeue thread id, but is it started?"
                     , ncPingMachine        = Nothing -- error "you want the NetCrypto-PingMachine, but is it started?"
                     , ncOutgoingQueue      = pktoq
+                    , ncLastNMsgs          = msgQ
+                    , ncListeners          = listeners
+                    , ncMsgNumVar          = msgNum
+                    , ncDropCntVar         = dropNum
                     }
     -- launch dequeue thread
     threadid <- forkIO $ do
@@ -615,9 +656,25 @@ allMsgTypes fDefault = A.listArray (minBound,maxBound) (0:knownMsgs)
 defaultCryptoDataHooks :: Map.Map MessageType [NetCryptoHook]
 defaultCryptoDataHooks = Map.empty
 
--- | discards all unrecognized packets
+-- | updates ncLastNMsgs, and sends message to type-0 listeners
 defaultUnRecHook :: MessageType -> NetCryptoHook
-defaultUnRecHook _ _ _ = return Nothing
+defaultUnRecHook typ session cm | any ($ typ) [isKillPacket, isOFFLINE] = atomically $ do
+    tmchans <- map snd . IntMap.elems <$> readTVar (ncListeners session)
+    forM_ tmchans $ \chan -> closeTMChan chan
+    return Nothing
+
+defaultUnRecHook typ  session  cm = do
+    let msgQ = ncLastNMsgs session
+        msgNumVar = ncMsgNumVar session
+        dropCntVar = ncDropCntVar session
+    atomically $ do
+        num <- readTVar msgNumVar
+        (wraps,offset) <- PQ.enqueue msgQ num (False,cm)
+        capacity <- PQ.getCapacity msgQ
+        let dropped = wraps * capacity + offset
+        modifyTVar' msgNumVar (+1)
+        writeTVar dropCntVar dropped
+    return Nothing
 
 -- | use to add a single hook to a specific session.
 addCryptoDataHook1 :: Map.Map MessageType [NetCryptoHook] -> MessageType -> NetCryptoHook -> Map.Map MessageType [NetCryptoHook]