From 1d4eebc853ccb4cebc2f530cd37c4622613d6a0e Mon Sep 17 00:00:00 2001
From: Denis Tereshkin <denis@kasan.ws>
Date: Tue, 29 Nov 2016 16:13:39 +0700
Subject: [PATCH] Trade sink: handle timeout

---
 src/ATrade/Broker/Server.hs | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/src/ATrade/Broker/Server.hs b/src/ATrade/Broker/Server.hs
index 2d4bdf3..bcce506 100644
--- a/src/ATrade/Broker/Server.hs
+++ b/src/ATrade/Broker/Server.hs
@@ -102,15 +102,18 @@ tradeSinkHandler c state tradeSinkEp = when (tradeSinkEp /= "") $
       withSocket c Req (\sock -> do
         chan <- tradeSink <$> readIORef state
         connect sock $ T.unpack tradeSinkEp
-        setReceiveTimeout (restrict 5000) sock
-        whileM_ (fmap killMvar (readIORef state) >>= fmap isNothing . tryReadMVar) $ do
+        timeoutMv <- newEmptyMVar
+        whileM_ (andM [fmap killMvar (readIORef state) >>= fmap isNothing . tryReadMVar, isNothing <$> tryReadMVar timeoutMv]) $ do
           threadDelay 500000
           maybeTrade <- tryReadChan chan
           case maybeTrade of
             Just trade -> send sock [] $ encodeTrade trade
             Nothing -> do
               send sock [] $ BL.toStrict $ encode TradeSinkHeartBeat
-              void $ receive sock -- anything will do
+              events <- poll 5000 [Sock sock [In] Nothing]
+              if not . L.null . L.head $ events
+                then void . receive $ sock -- anything will do
+                else putMVar timeoutMv ()
         )
 
     where