Fix: shutdown on win32

src/ATrade/Broker/Server.hs

@@ -66,6 +66,7 @@ startBrokerServer brokers c ep = do
   }
   mapM_ (\bro -> setNotificationCallback bro (Just $ notificationCallback state)) brokers
 
+  debugM "Broker.Server" "Forking broker server thread"
   BrokerServerHandle <$> forkIO (brokerServerThread state) <*> pure compMv
 
 notificationCallback :: IORef BrokerServerState -> Notification -> IO ()
@@ -85,29 +86,31 @@ brokerServerThread state = finally brokerServerThread' cleanup
   where
     brokerServerThread' = forever $ do
       sock <- bsSocket <$> readIORef state
-      msg <- receiveMulti sock
+      evs <- poll 200 [Sock sock [In] Nothing] 
-      case msg of
+      when ((L.length . L.head) evs > 0) $ do
-        [peerId, _, payload] ->
+        msg <- receiveMulti sock
-          case decode . BL.fromStrict $ payload of
+        case msg of
-            Just request -> do
+          [peerId, _, payload] ->
-              let sqnum = requestSqnum request
+            case decode . BL.fromStrict $ payload of
-              -- Here, we should check if previous packet sequence number is the same
+              Just request -> do
-              -- If it is, we should resend previous response
+                let sqnum = requestSqnum request
-              lastPackMap <- lastPacket <$> readIORef state
+                -- Here, we should check if previous packet sequence number is the same
-              case shouldResend sqnum peerId lastPackMap of
+                -- If it is, we should resend previous response
-                Just response -> sendMessage sock peerId response -- Resend
+                lastPackMap <- lastPacket <$> readIORef state
-                Nothing -> do
+                case shouldResend sqnum peerId lastPackMap of
-                  -- Handle incoming request, send response
+                  Just response -> sendMessage sock peerId response -- Resend
-                  response <- handleMessage peerId request
+                  Nothing -> do
-                  sendMessage sock peerId response
+                    -- Handle incoming request, send response
-                  -- and store response in case we'll need to resend it
+                    response <- handleMessage peerId request
-                  atomicMapIORef state (\s -> s { lastPacket = M.insert peerId (sqnum, response) (lastPacket s)})
+                    sendMessage sock peerId response
-            Nothing -> do
+                    -- and store response in case we'll need to resend it
-              -- If we weren't able to parse request, we should send error
+                    atomicMapIORef state (\s -> s { lastPacket = M.insert peerId (sqnum, response) (lastPacket s)})
-              -- but shouldn't update lastPacket
+              Nothing -> do
-              let response = ResponseError "Invalid request"
+                -- If we weren't able to parse request, we should send error
-              sendMessage sock peerId response
+                -- but shouldn't update lastPacket
-        _ -> warningM "Broker.Server" ("Invalid packet received: " ++ show msg)
+                let response = ResponseError "Invalid request"
+                sendMessage sock peerId response
+          _ -> warningM "Broker.Server" ("Invalid packet received: " ++ show msg)
 
     shouldResend sqnum peerId lastPackMap = case M.lookup peerId lastPackMap of
       Just (lastSqnum, response) -> if sqnum == lastSqnum

src/ATrade/QuoteSource/Client.hs

@@ -9,10 +9,12 @@ import ATrade.Types
 import Control.Concurrent hiding (readChan, writeChan)
 import Control.Concurrent.BoundedChan
 import Control.Concurrent.MVar
+import Control.Monad
 import Control.Exception
 import Data.List.NonEmpty
 import qualified Data.Text as T
 import qualified Data.ByteString.Lazy as BL
+import qualified Data.List as L
 import Data.Text.Encoding
 import System.ZMQ4
 import System.Log.Logger
@@ -33,10 +35,12 @@ startQuoteSourceClient chan tickers ctx endpoint = do
   return QuoteSourceClientHandle { tid = tid, completionMvar = compMv }
   where
     clientThread sock = do
-      rawTick <- fmap BL.fromStrict <$> receiveMulti sock
+      evs <- poll 200 [Sock sock [In] Nothing] 
-      case deserializeTick rawTick of
+      when ((L.length . L.head) evs > 0) $ do
-        Just tick -> writeChan chan tick
+        rawTick <- fmap BL.fromStrict <$> receiveMulti sock
-        Nothing -> warningM "QuoteSource.Client" "Error: can't deserialize tick"
+        case deserializeTick rawTick of
+          Just tick -> writeChan chan tick
+          Nothing -> warningM "QuoteSource.Client" "Error: can't deserialize tick"
     cleanup compMv sock = close sock >> putMVar compMv ()
 
 stopQuoteSourceClient :: QuoteSourceClientHandle -> IO ()