2 files changed, 31 insertions, 4 deletions

diff --git a/examples/dhtd.hs b/examples/dhtd.hs
index 6e2647d1..1aa36b77 100644
--- a/examples/dhtd.hs
+++ b/examples/dhtd.hs
@@ -94,6 +94,7 @@ import Data.Typeable
 import Network.Tox.ContactInfo as Tox
 import OnionRouter
 import PingMachine
+import Data.PacketQueue
 
 -- Presence imports.
 import ConsoleWriter
@@ -998,8 +999,20 @@ newXmmpSource session = do
                                  newXmmpSource session
 
 newXmmpSink :: Tox.NetCryptoSession -> C.Sink (Flush Tox.CryptoMessage) IO ()
-newXmmpSink sessions = C.awaitForever $ \flush_cyptomessage -> do
-    liftIO $ (_todo sessions {- send the fucking message -}) flush_cyptomessage
+newXmmpSink session@(Tox.NCrypto { ncOutgoingQueue, ncPacketQueue }) = C.awaitForever $ \flush_cyptomessage -> do
+    let sendit :: Tox.NetCryptoSession -> Flush Tox.CryptoMessage -> IO ()
+        sendit session (Chunk msg) = do
+            extra <- readyOutGoing ncOutgoingQueue
+            r <- atomically $ do
+                rTry <- tryAppendQueueOutgoing extra ncOutgoingQueue msg
+                case rTry of
+                    OGFull -> retry
+                    OGSuccess -> return OGSuccess
+                    OGEncodeFail -> return OGEncodeFail
+            when (r == OGEncodeFail) $
+                hPutStrLn stderr ("FAILURE to Encode Outgoing: " ++ show msg)
+        sendit session Flush = return ()
+    liftIO $ sendit session flush_cyptomessage
 
 -- | TODO
 --
diff --git a/src/Data/PacketQueue.hs b/src/Data/PacketQueue.hs
index cde26fb7..b7737656 100644
--- a/src/Data/PacketQueue.hs
+++ b/src/Data/PacketQueue.hs
@@ -12,8 +12,10 @@ module Data.PacketQueue
     , observeOutOfBand
     , PacketOutQueue
     , newOutGoing
+    , readyOutGoing
     , tryAppendQueueOutgoing
     , dequeueOutgoing
+    , getHighestHandledPacketPlus1
     , mapOutGoing
     , OutGoingResult(..)
     ) where
@@ -158,8 +160,18 @@ tryAppendQueueOutgoing getExtra q@(PacketOutQueue { pktoInPQ, pktoOutPQ, pktoPac
                         writeTVar pktoPacketNo $! pktno'
                         writeArray (pktq pktoOutPQ) i (Just (pktno,pkt))
                         return OGSuccess
-                    -- queue is full, block until its not
-                    _       -> return OGFull
+                    -- queue is full
+                    Just (n,_) -> do
+                        nn <- getHighestHandledPacketPlus1 q
+                        if (n < nn)
+                          -- but we can overwrite an old packet
+                          then do
+                            modifyTVar' (buffend pktoOutPQ) (+1)
+                            writeTVar pktoPacketNo $! pktno'
+                            writeArray (pktq pktoOutPQ) i (Just (pktno,pkt))
+                            return OGSuccess
+                          --  uh oh this packet is still needed...
+                          else return OGFull
         -- don't know how to send this message
         Nothing -> return OGEncodeFail
 
@@ -172,3 +184,5 @@ dequeueOutgoing (PacketOutQueue {pktoOutPQ=PacketQueue { pktq, seqno, qsize }})
     modifyTVar' seqno   succ
     return x
 
+getHighestHandledPacketPlus1 :: PacketOutQueue extra msg wire fromwire -> STM Word32
+getHighestHandledPacketPlus1 (PacketOutQueue { pktoInPQ }) = readTVar (buffend pktoInPQ)