dequeue thread & handle session close

1 files changed, 12 insertions, 12 deletions

diff --git a/src/Network/Tox/Crypto/Handlers.hs b/src/Network/Tox/Crypto/Handlers.hs
index 4f53888b..50dd8c67 100644
--- a/src/Network/Tox/Crypto/Handlers.hs
+++ b/src/Network/Tox/Crypto/Handlers.hs
@@ -28,7 +28,6 @@ import qualified Data.Set         as Set
 import qualified Data.Array.Unboxed as A
 import SensibleDir
 import System.FilePath
-import System.IO
 import System.IO.Temp
 import System.Environment
 import System.Directory
@@ -108,6 +107,7 @@ data NetCryptoSession = NCrypto
     , ncView               :: TVar SessionView
     , ncPacketQueue        :: PacketQueue CryptoData
     , ncBufferStart        :: TVar Word32
+    , ncDequeueThread      :: Maybe ThreadId
     , ncPingMachine        :: Maybe PingMachine
     , ncOutgoingQueue      :: PQ.PacketOutQueue (State,Nonce24,TVar MsgOutMap)
                                                 CryptoMessage
@@ -364,24 +364,23 @@ freshCryptoSession  sessions
                     , ncView               = ncView0
                     , ncPacketQueue        = pktq
                     , ncBufferStart        = bufstart
+                    , 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
                     }
-
-    hooks <- atomically $ readTVar (announceNewSessionHooks sessions)
-
-    -- Dequeue thread:
-    --
-    --   Hopefully, somebody will launch a thread to repeatedly call
-    --   'receiveCrypto' in order to dequeue messages from ncPacketQueue.
-    when (null hooks) $ do
-        hPutStrLn stderr "Warning: Missing new-session handler.  Lost session!"
-
+    -- launch dequeue thread
+    threadid <- forkIO $ do
+        tid <- myThreadId
+        labelThread tid ("NetCryptoDequeue." ++ show (key2id remotePublicKey))
+        fix $ \loop -> do
+            cd <- atomically $ PQ.dequeue pktq
+            _ <- runCryptoHook (netCryptoSession0 {ncDequeueThread=Just tid}) cd
+            loop
     -- launch ping thread
     fuzz <- randomRIO (0,2000)
     pingMachine <- forkPingMachine ("NetCrypto." ++ show (key2id remotePublicKey)) (15000 + fuzz) 2000
     -- update session with thread ids
-    let netCryptoSession = netCryptoSession0 {ncPingMachine=Just pingMachine}
+    let netCryptoSession = netCryptoSession0 {ncDequeueThread=Just threadid, ncPingMachine=Just pingMachine}
     -- add this session to the lookup maps
     atomically $ do
         modifyTVar allsessions (Map.insert addr netCryptoSession)
@@ -390,6 +389,7 @@ freshCryptoSession  sessions
             Nothing -> modifyTVar allsessionsByKey (Map.insert remotePublicKey [netCryptoSession])
             Just xs -> modifyTVar allsessionsByKey (Map.insert remotePublicKey (netCryptoSession:xs))
     -- run announceNewSessionHooks
+    hooks <- atomically $ readTVar (announceNewSessionHooks sessions)
     flip fix (hooks,netCryptoSession) $ \loop (hooks,session) ->
         case hooks of
             [] -> return ()