2 files changed, 54 insertions, 25 deletions

diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index d174b10c..f31a62e4 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -4,6 +4,7 @@
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE TupleSections   #-}
 {-# LANGUAGE TypeOperators   #-}
+{-# LANGUAGE ViewPatterns    #-}
 module Network.Tox.Crypto.Handlers where
 
 import Network.Tox.NodeId
@@ -341,12 +342,15 @@ data NetCryptoSession = NCrypto
     -- ^ a buffer in which incoming packets may be stored out of order
     -- but from which they may be extracted in sequence,
     -- helps ensure lossless packets are processed in order
-    , ncDequeueThread      :: Maybe ThreadId
+    , ncDequeueThread      :: TVar (Maybe ThreadId)
     -- ^ when the thread which dequeues from ncPacketQueue
     -- is started, its ThreadId is stored here
-    , ncPingMachine        :: Maybe PingMachine
-    , ncPingThread         :: Maybe ThreadId
-    -- ^ when the ping thread is started, store it here
+    , ncDequeueOutGoingThread :: TVar (Maybe ThreadId)
+    -- ^ the thread which actually sends lossless packets
+    , ncPingMachine        :: TVar (Maybe PingMachine)
+    -- ^ thread which triggers ping events
+    , ncPingThread         :: TVar (Maybe ThreadId)
+    -- ^ thread which actually queues outgoing pings
     , ncOutgoingQueue      :: TVar
                                 (UponHandshake
                                     (PQ.PacketOutQueue
@@ -599,6 +603,10 @@ freshCryptoSession  sessions
     dmsg $ "freshCryptoSession: Session ncTheirBaseNonce=" ++ show theirbasenonce
     dmsg $ "freshCryptoSession: My Session Public =" ++ show (key2id $ toPublic newsession)
     ncTheirSessionPublic0 <- newTVar (frmMaybe mbtheirSessionKey)
+    ncDequeueThread0 <- newTVar Nothing
+    ncDequeueOutGoingThread0 <- newTVar Nothing
+    ncPingMachine0 <- newTVar Nothing
+    ncPingThread0 <- newTVar Nothing
     let netCryptoSession0 =
             NCrypto { ncState              = ncState0
                     , ncMyPublicKey        = toPublic key
@@ -621,9 +629,10 @@ freshCryptoSession  sessions
                     , ncOutgoingIdMapEscapedLossless = losslessEscapeIdMap
                     , ncView               = ncView0
                     , ncPacketQueue        = pktq
-                    , 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?"
-                    , ncPingThread         = Nothing -- error "you want the NetCrypto-PingSender, but is it started?"
+                    , ncDequeueThread         = ncDequeueThread0
+                    , ncDequeueOutGoingThread = ncDequeueOutGoingThread0
+                    , ncPingMachine           = ncPingMachine0
+                    , ncPingThread            = ncPingThread0
                     , ncOutgoingQueue      = mbpktoqVar
                     , ncLastNMsgs          = lastNQ
                     , ncListeners          = listeners
@@ -713,33 +722,40 @@ runUponHandshake netCryptoSession0 addr pktoq = do
         crypto = transportCrypto sessions
         allsessions = netCryptoSessions sessions
         allsessionsByKey = netCryptoSessionsByKey sessions
+        sidStr = printf "(%x)" (ncSessionId netCryptoSession0)
     -- launch dequeue thread
     -- (In terms of data dependency, this thread could be launched prior to handshake)
     threadid <- forkIO $ do
         tid <- myThreadId
-        labelThread tid ("NetCryptoDequeue." ++ show (key2id remotePublicKey))
+        atomically $ writeTVar (ncDequeueThread netCryptoSession0) (Just tid)
+        labelThread tid ("NetCryptoDequeue." ++ show (key2id remotePublicKey) ++ sidStr)
         fix $ \loop -> do
             cd <- atomically $ PQ.dequeue pktq
-            _ <- runCryptoHook (netCryptoSession0 {ncDequeueThread=Just tid}) (bufferData cd)
+            dput XNetCrypto $ "Dequeued::" ++ show (bufferData cd) ++ " now running hook..."
+            _ <- runCryptoHook netCryptoSession0 (bufferData cd)
             loop
-    dput XNetCrypto $ "runUponHandshake: " ++ show threadid ++ " = NetCryptoDequeue." ++ show (key2id remotePublicKey)
+    dput XNetCrypto $ "runUponHandshake: " ++ show threadid ++ " = NetCryptoDequeue." ++ show (key2id remotePublicKey) ++ sidStr
     -- launch dequeueOutgoing thread
     threadidOutgoing <- forkIO $ do
         tid <- myThreadId
-        labelThread tid ("NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey))
+        atomically $ writeTVar (ncDequeueOutGoingThread netCryptoSession0) (Just tid)
+        labelThread tid ("NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey) ++ sidStr)
         fix $ \loop -> do
             (_,pkt) <- atomically $ PQ.dequeueOutgoing pktoq
             dput XNetCrypto "NetCryptoDequeueOutgoing thread... Sending encrypted Packet"
             sendSessionPacket sessions addr pkt
             loop
-    dput XNetCrypto $ "runUponHandshake: " ++ show threadidOutgoing ++ " = NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey)
+    dput XNetCrypto $ "runUponHandshake: " ++ show threadidOutgoing ++ " = NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey) ++ sidStr
     -- launch ping Machine thread
     fuzz <- return 0 -- randomRIO (0,2000) -- Fuzz to prevent simultaneous ping/pong exchanges.
                                            -- Disabled because tox has no pong event.
-    pingMachine <- forkPingMachine ("NetCrypto." ++ show (key2id remotePublicKey)) (8000 + fuzz) 4000
+    pingMachine <- forkPingMachine ("NetCrypto." ++ show (key2id remotePublicKey) ++ sidStr) (8000 + fuzz) 4000
+    atomically $ writeTVar (ncPingMachine netCryptoSession0) (Just pingMachine)
+    -- launch ping thread
     pingThreadId <- forkIO $ do
         tid <- myThreadId
-        labelThread tid ("NetCryptoPingSender." ++ show (key2id remotePublicKey) ++ "." ++ show (ncSessionId netCryptoSession0))
+        atomically $ writeTVar (ncPingThread netCryptoSession0) (Just tid)
+        labelThread tid ("NetCryptoPingSender." ++ show (key2id remotePublicKey) ++ sidStr)
         event <- atomically $ pingWait pingMachine
         case event of
             PingIdle -> do
@@ -756,9 +772,7 @@ runUponHandshake netCryptoSession0 addr pktoq = do
                     Right _ -> return ()
             PingTimeOut -> dput XNetCrypto $ "(pingThread session: " ++ show (ncSessionId netCryptoSession0) ++ ") PingTimeOut TODO"
     -- update session with thread ids
-    let netCryptoSession = netCryptoSession0 { ncDequeueThread=Just threadid
-                                             , ncPingMachine=Just pingMachine
-                                             , ncPingThread=Just pingThreadId}
+    let netCryptoSession = netCryptoSession0
     -- add this session to the lookup maps
     -- atomically $ addSessionToMapIfNotThere sessions addr netCryptoSession
     -- run announceNewSessionHooks
@@ -778,10 +792,10 @@ destroySession :: NetCryptoSession -> IO ()
 destroySession session = do
     let allsessions = ncAllSessions session
         sid = ncSessionId session
-        stopThread :: Maybe ThreadId -> IO ()
-        stopThread x = maybe (return ()) killThread x
-        stopMachine :: Maybe PingMachine -> IO ()
-        stopMachine x = maybe (return ()) pingCancel x
+        stopThread :: TVar (Maybe ThreadId) -> IO ()
+        stopThread x = atomically (readTVar x) >>= maybe (return ()) killThread
+        stopMachine :: TVar (Maybe PingMachine) -> IO ()
+        stopMachine x = atomically (readTVar x) >>= maybe (return ()) pingCancel
     atomically $ do
         modifyTVar (netCryptoSessionsByKey allsessions)
                    $ Map.map (filter ((/=sid) . ncSessionId))
@@ -790,6 +804,7 @@ destroySession session = do
     stopThread (ncPingThread session)
     stopMachine (ncPingMachine session)
     stopThread (ncDequeueThread session)
+    stopThread (ncDequeueOutGoingThread session)
 
 -- | Called when we get a handshake, but there's already a session entry.
 --
@@ -960,7 +975,8 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
                 Left s -> do
                     dput XNetCrypto $ "(NetCrypto)sessionPacketH: " ++ s
                     return Nothing -- decryption failed, ignore packet
-                Right cd@(CryptoData {bufferStart, bufferEnd, bufferData=cm}) -> do -- decryption succeeded,
+                Right cd'@(CryptoData {bufferStart, bufferEnd, bufferData=(unpadCryptoMsg -> cm)}) -> do -- decryption succeeded,
+                    let cd = cd' { bufferData= cm }
                     -- TODO: Why do I need bufferStart & bufferEnd?
                     --
                     --   buffer_start = highest packet number handled + 1
@@ -978,7 +994,8 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
                     -- then set session confirmed,
                     atomically $ writeTVar ncState {-Confirmed-}Established
                     -- bump ping machine
-                    case ncPingMachine of
+                    ncPingMachine0 <- atomically $ readTVar ncPingMachine
+                    case ncPingMachine0 of
                         Just pingMachine -> -- Normally, we would bump the PingMachine to mark the connection
                                             -- as non-idle so that we don't need to send a ping message.
                                             -- Because tox has no pong message, we need to send a ping every
@@ -995,10 +1012,10 @@ sessionPacketH sessions addrRaw (CryptoPacket nonce16 encrypted) = do
                         isLossy (Msg mid) | lossyness mid == Lossy = True
                         isLossy _ = False
                     if isLossy msgTypMapped
-                        then do dput XNetCrypto "enqueue ncPacketQueue Lossy"
+                        then do dput XNetCrypto $ "enqueue ncPacketQueue Lossy " ++ show cm
                                 atomically $ PQ.observeOutOfBand ncPacketQueue bufferEnd
                                 runCryptoHook session (bufferData cd)
-                        else do dput XNetCrypto "enqueue ncPacketQueue Lossless"
+                        else do dput XNetCrypto $ "enqueue ncPacketQueue Lossless " ++ show cm
                                 when (msgID cm == PING) $
                                     dput XNetCrypto $ "NetCrypto Recieved PING (session " ++ show ncSessionId ++")"
                                 atomically $ PQ.enqueue ncPacketQueue bufferEnd cd
diff --git a/src/Network/Tox/Crypto/Transport.hs b/src/Network/Tox/Crypto/Transport.hs
index ce3bafee..fab0f3e2 100644
--- a/src/Network/Tox/Crypto/Transport.hs
+++ b/src/Network/Tox/Crypto/Transport.hs
@@ -7,6 +7,7 @@
 module Network.Tox.Crypto.Transport
     ( parseCrypto
     , encodeCrypto
+    , unpadCryptoMsg
     , parseHandshakes
     , encodeHandshakes
     , CryptoData(..)
@@ -220,6 +221,17 @@ data UserStatus = Online | Away | Busy deriving (Show,Read,Eq,Ord,Enum)
 
 data TypingStatus = NotTyping | Typing deriving (Show,Read,Eq,Ord,Enum)
 
+unpadCryptoMsg :: CryptoMessage -> CryptoMessage
+unpadCryptoMsg x@(TwoByte Padding (toEnum8 -> mid))
+    | msgSizeParam mid == Just (True,0) = OneByte mid
+unpadCryptoMsg x@(UpToN mid0 (B.dropWhile (==0) -> B.uncons -> Just (toEnum8 -> mid,bytes)))
+    | mid0 == Padding
+        = case msgSizeParam mid of
+            Just (True,0)  -> OneByte mid
+            Just (True,1)  -> TwoByte mid (B.head bytes)
+            _ -> UpToN mid bytes
+unpadCryptoMsg x = x
+
 data CryptoMessage
         = OneByte  { msgID :: MessageID }
         | TwoByte  { msgID :: MessageID, msgByte  :: Word8 }