asakul
/
libatrade
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

BrokerServer: hslogger => co-log

master
Denis Tereshkin 4 years ago
parent
3ae040d89b
commit
9c95b7291a
1 changed files with 31 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 47
      src/ATrade/Broker/Server.hs

47
src/ATrade/Broker/Server.hs
Unescape View File

@ -18,11 +18,16 @@ import ATrade.Broker.Protocol (BrokerServerRequest (..),
RequestSqnum, RequestSqnum,
getNotificationSqnum, getNotificationSqnum,
nextSqnum, requestSqnum) nextSqnum, requestSqnum)
import ATrade.Logging (Message (Message),
Severity (Debug, Warning),
logWith)
import ATrade.Logging (Severity (Info))
import ATrade.Types (Order (orderAccountId, orderId), import ATrade.Types (Order (orderAccountId, orderId),
OrderId, OrderId,
ServerSecurityParams (sspCertificate, sspDomain), ServerSecurityParams (sspCertificate, sspDomain),
Trade (tradeOrderId)) Trade (tradeOrderId))
import ATrade.Util (atomicMapIORef) import ATrade.Util (atomicMapIORef)
import Colog (LogAction)
import Control.Concurrent (MVar, ThreadId, forkIO, import Control.Concurrent (MVar, ThreadId, forkIO,
killThread, myThreadId, killThread, myThreadId,
newEmptyMVar, putMVar, newEmptyMVar, putMVar,
@ -93,8 +98,9 @@ startBrokerServer :: [BrokerBackend] ->
T.Text -> T.Text ->
[TradeSink] -> [TradeSink] ->
ServerSecurityParams -> ServerSecurityParams ->
LogAction IO Message ->
IO BrokerServerHandle IO BrokerServerHandle
startBrokerServer brokers c ep notificationsEp tradeSinks params = do startBrokerServer brokers c ep notificationsEp tradeSinks params logger = do
sock <- socket c Router sock <- socket c Router
notificationsSock <- socket c Pub notificationsSock <- socket c Pub
setLinger (restrict 0) sock setLinger (restrict 0) sock
@ -136,14 +142,19 @@ startBrokerServer brokers c ep notificationsEp tradeSinks params = do
orderIdCounter = 1, orderIdCounter = 1,
tradeSink = tsChan tradeSink = tsChan
} }
mapM_ (\bro -> setNotificationCallback bro (Just $ notificationCallback state)) brokers mapM_ (\bro -> setNotificationCallback bro (Just $ notificationCallback state logger)) brokers
debugM "Broker.Server" "Forking broker server thread" log Info "Broker.Server" "Forking broker server thread"
BrokerServerHandle <$> forkIO (brokerServerThread state) <*> forkIO (tradeSinkHandler c state tradeSinks) <*> pure compMv <*> pure killMv BrokerServerHandle <$> forkIO (brokerServerThread state logger) <*> forkIO (tradeSinkHandler c state tradeSinks) <*> pure compMv <*> pure killMv
where
log = logWith logger
notificationCallback :: IORef BrokerServerState -> BrokerBackendNotification -> IO () notificationCallback :: IORef BrokerServerState ->
notificationCallback state n = do LogAction IO Message ->
debugM "Broker.Server" $ "Notification: " ++ show n BrokerBackendNotification ->
IO ()
notificationCallback state logger n = do
log Debug "Broker.Server" $ "Notification: " <> (T.pack . show) n
chan <- tradeSink <$> readIORef state chan <- tradeSink <$> readIORef state
case n of case n of
BackendTradeNotification trade -> tryWriteChan chan trade BackendTradeNotification trade -> tryWriteChan chan trade
@ -157,11 +168,12 @@ notificationCallback state n = do
case n of case n of
BackendTradeNotification trade -> addNotification clientIdentity (TradeNotification sqnum trade { tradeOrderId = localOrderId }) BackendTradeNotification trade -> addNotification clientIdentity (TradeNotification sqnum trade { tradeOrderId = localOrderId })
BackendOrderNotification globalOrderId newstate -> addNotification clientIdentity (OrderNotification sqnum localOrderId newstate) BackendOrderNotification globalOrderId newstate -> addNotification clientIdentity (OrderNotification sqnum localOrderId newstate)
Nothing -> warningM "Broker.Server" "Notification: unknown order" Nothing -> log Warning "Broker.Server" "Notification: unknown order"
where where
log = logWith logger
addNotification clientIdentity n = do addNotification clientIdentity n = do
debugM "Broker.Server" $ "Sending notification to client [" <> T.unpack clientIdentity <> "]" log Debug "Broker.Server" $ "Sending notification to client [" <> clientIdentity <> "]"
atomicMapIORef state (\s -> atomicMapIORef state (\s ->
case M.lookup clientIdentity . pendingNotifications $ s of case M.lookup clientIdentity . pendingNotifications $ s of
Just ns -> s { pendingNotifications = M.insert clientIdentity (n : ns) (pendingNotifications s)} Just ns -> s { pendingNotifications = M.insert clientIdentity (n : ns) (pendingNotifications s)}
@ -178,12 +190,15 @@ tradeSinkHandler c state tradeSinks = unless (null tradeSinks) $
Just trade -> mapM_ (\x -> x trade) tradeSinks Just trade -> mapM_ (\x -> x trade) tradeSinks
Nothing -> threadDelay 100000 Nothing -> threadDelay 100000
where where
wasKilled = isJust <$> (killMvar <$> readIORef state >>= tryReadMVar) wasKilled = isJust <$> (readIORef state >>= tryReadMVar . killMvar)
brokerServerThread :: IORef BrokerServerState -> IO () brokerServerThread :: IORef BrokerServerState ->
brokerServerThread state = finally brokerServerThread' cleanup LogAction IO Message ->
IO ()
brokerServerThread state logger = finally brokerServerThread' cleanup
where where
log = logWith logger
brokerServerThread' = whileM_ (fmap killMvar (readIORef state) >>= fmap isNothing . tryReadMVar) $ do brokerServerThread' = whileM_ (fmap killMvar (readIORef state) >>= fmap isNothing . tryReadMVar) $ do
sock <- bsSocket <$> readIORef state sock <- bsSocket <$> readIORef state
events <- poll 100 [Sock sock [In] Nothing] events <- poll 100 [Sock sock [In] Nothing]
@ -199,7 +214,7 @@ brokerServerThread state = finally brokerServerThread' cleanup
lastPackMap <- lastPacket <$> readIORef state lastPackMap <- lastPacket <$> readIORef state
case shouldResend sqnum peerId lastPackMap of case shouldResend sqnum peerId lastPackMap of
Just response -> do Just response -> do
debugM "Broker.Server" $ "Resending packet for peerId: " ++ show peerId log Debug "Broker.Server" $ "Resending packet for peerId: " <> (T.pack . show) peerId
sendMessage sock peerId response -- Resend sendMessage sock peerId response -- Resend
atomicMapIORef state (\s -> s { lastPacket = M.delete peerId (lastPacket s)}) atomicMapIORef state (\s -> s { lastPacket = M.delete peerId (lastPacket s)})
Nothing -> do Nothing -> do
@ -213,7 +228,7 @@ brokerServerThread state = finally brokerServerThread' cleanup
-- but shouldn't update lastPacket -- but shouldn't update lastPacket
let response = ResponseError $ "Invalid request: " <> T.pack errmsg let response = ResponseError $ "Invalid request: " <> T.pack errmsg
sendMessage sock peerId response sendMessage sock peerId response
_ -> warningM "Broker.Server" ("Invalid packet received: " ++ show msg) _ -> log Warning "Broker.Server" ("Invalid packet received: " <> (T.pack . show) msg)
shouldResend sqnum peerId lastPackMap = case M.lookup peerId lastPackMap of shouldResend sqnum peerId lastPackMap = case M.lookup peerId lastPackMap of
Just (lastSqnum, response) -> if sqnum == lastSqnum Just (lastSqnum, response) -> if sqnum == lastSqnum
@ -233,7 +248,7 @@ brokerServerThread state = finally brokerServerThread' cleanup
bros <- brokers <$> readIORef state bros <- brokers <$> readIORef state
case request of case request of
RequestSubmitOrder sqnum clientIdentity order -> do RequestSubmitOrder sqnum clientIdentity order -> do
debugM "Broker.Server" $ "Request: submit order:" ++ show request log Debug "Broker.Server" $ "Request: submit order:" <> (T.pack . show) request
case findBrokerForAccount (orderAccountId order) bros of case findBrokerForAccount (orderAccountId order) bros of
Just bro -> do Just bro -> do
globalOrderId <- nextOrderId globalOrderId <- nextOrderId
@ -245,7 +260,7 @@ brokerServerThread state = finally brokerServerThread' cleanup
return ResponseOk return ResponseOk
Nothing -> do Nothing -> do
warningM "Broker.Server" $ "Unknown account: " ++ T.unpack (orderAccountId order) log Warning "Broker.Server" $ "Unknown account: " <> (orderAccountId order)
return $ ResponseError "Unknown account" return $ ResponseError "Unknown account"
RequestCancelOrder sqnum clientIdentity localOrderId -> do RequestCancelOrder sqnum clientIdentity localOrderId -> do
m <- orderToBroker <$> readIORef state m <- orderToBroker <$> readIORef state

Write Preview
Loading…
Cancel
Save