dequeueOutgoing thread

2 files changed, 13 insertions, 1 deletions

diff --git a/src/Network/Tox.hs b/src/Network/Tox.hs
index dfa0ea9e..aa95df81 100644
--- a/src/Network/Tox.hs
+++ b/src/Network/Tox.hs
@@ -310,7 +310,7 @@ newTox :: TVar Onion.AnnouncedKeys   -- ^ Store of announced keys we are a rende
        -> IO Tox
 newTox keydb addr mbSessionsState suppliedDHTKey = do
     udp <- {- addVerbosity <$> -} udpTransport addr
-    (crypto0,sessionsState) <- case mbSessionsState of
+    (crypto0,sessionsState0) <- case mbSessionsState of
                                  Nothing -> do
                                     crypto <- newCrypto
                                     sessionsState <- newSessionsState crypto defaultUnRecHook defaultCryptoDataHooks
@@ -335,6 +335,7 @@ newTox keydb addr mbSessionsState suppliedDHTKey = do
     let ignoreErrors _ = return () -- Set this to (hPutStrLn stderr) to debug onion route building.
     orouter <- newOnionRouter ignoreErrors
     (dhtcrypt,onioncrypt,dtacrypt,cryptonet) <- toxTransport crypto orouter lookupClose udp
+    let sessionsState = sessionsState0 { sessionTransport = cryptonet }
     let dhtnet0 = layerTransportM (DHT.decrypt crypto) (DHT.encrypt crypto) dhtcrypt
         tbl4 = DHT.routing4 $ mkrouting (error "missing client")
         tbl6 = DHT.routing6 $ mkrouting (error "missing client")
diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index 50dd8c67..ef7afe98 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -3,6 +3,7 @@
 {-# LANGUAGE TypeOperators #-}
 module Network.Tox.Crypto.Handlers where
 
+import Network.QueryResponse
 import Network.Tox.NodeId
 import Network.Tox.Crypto.Transport
 import Network.Tox.DHT.Transport (Cookie(..),CookieData(..), CookieRequest(..), NoSpam(..))
@@ -127,6 +128,7 @@ data NetCryptoSessions = NCSessions
     , outboundQueueCapacity   :: Word32
     , nextSessionId           :: TVar SessionID
     , announceNewSessionHooks :: TVar [IOHook (Maybe NoSpam) NetCryptoSession]
+    , sessionTransport        :: Transport String SockAddr NetCrypto
     }
 
 type NewSessionHook = IOHook (Maybe NoSpam) NetCryptoSession
@@ -190,6 +192,7 @@ newSessionsState crypto unrechook hooks = do
                       , outboundQueueCapacity   = 400
                       , nextSessionId           = nextSessionId0
                       , announceNewSessionHooks = announceNewSessionHooks0
+                      , sessionTransport = error "Need to set sessionTransport field of NetCryptoSessions!"
                       }
 
 data HandshakeParams
@@ -376,6 +379,14 @@ freshCryptoSession  sessions
             cd <- atomically $ PQ.dequeue pktq
             _ <- runCryptoHook (netCryptoSession0 {ncDequeueThread=Just tid}) cd
             loop
+    -- launch dequeueOutgoing thread
+    threadidOutgoing <- forkIO $ do
+        tid <- myThreadId
+        labelThread tid ("NetCryptoDequeueOutgoing." ++ show (key2id remotePublicKey))
+        fix $ \loop -> do
+            (_,pkt) <- atomically $ PQ.dequeueOutgoing pktoq
+            sendMessage (sessionTransport sessions) addr (NetCrypto pkt)
+            loop
     -- launch ping thread
     fuzz <- randomRIO (0,2000)
     pingMachine <- forkPingMachine ("NetCrypto." ++ show (key2id remotePublicKey)) (15000 + fuzz) 2000