From 48f6c6ad3c2ba006499ac89b8f1938fa6f16dc8f Mon Sep 17 00:00:00 2001 From: Denis Tereshkin Date: Thu, 2 Jun 2022 22:40:51 +0700 Subject: [PATCH] BrokerServer: start NotificationSqnum from arbitrary value --- src/ATrade/Broker/Server.hs | 12 +++++--- test/TestBrokerClient.hs | 6 ++-- test/TestBrokerServer.hs | 61 ++++++++----------------------------- 3 files changed, 23 insertions(+), 56 deletions(-) diff --git a/src/ATrade/Broker/Server.hs b/src/ATrade/Broker/Server.hs index b9815fc..9c0bc13 100644 --- a/src/ATrade/Broker/Server.hs +++ b/src/ATrade/Broker/Server.hs @@ -84,7 +84,9 @@ data BrokerServerState = BrokerServerState { completionMvar :: MVar (), killMvar :: MVar (), orderIdCounter :: OrderId, - tradeSink :: BoundedChan Trade + tradeSink :: BoundedChan Trade, + initialSqnum :: NotificationSqnum + } data BrokerServerHandle = BrokerServerHandle ThreadId ThreadId (MVar ()) (MVar ()) @@ -95,11 +97,12 @@ startBrokerServer :: [BrokerBackend] -> Context -> T.Text -> T.Text -> + NotificationSqnum -> [TradeSink] -> ServerSecurityParams -> LogAction IO Message -> IO BrokerServerHandle -startBrokerServer brokers c ep notificationsEp tradeSinks params logger = do +startBrokerServer brokers c ep notificationsEp initialSqnum tradeSinks params logger = do sock <- socket c Router notificationsSock <- socket c Pub setLinger (restrict 0) sock @@ -139,7 +142,8 @@ startBrokerServer brokers c ep notificationsEp tradeSinks params logger = do completionMvar = compMv, killMvar = killMv, orderIdCounter = 1, - tradeSink = tsChan + tradeSink = tsChan, + initialSqnum = initialSqnum } mapM_ (\bro -> setNotificationCallback bro (Just $ notificationCallback state logger)) brokers @@ -161,7 +165,7 @@ notificationCallback state logger n = do orders <- orderMap <$> readIORef state case BM.lookupR (backendNotificationOrderId n) orders of Just (FullOrderId clientIdentity localOrderId) -> do - sqnum <- atomicModifyIORef' state (\s -> let sqnum = M.findWithDefault (NotificationSqnum 1) clientIdentity (notificationSqnum s) in + sqnum <- atomicModifyIORef' state (\s -> let sqnum = M.findWithDefault (initialSqnum s) clientIdentity (notificationSqnum s) in (s { notificationSqnum = M.insert clientIdentity (nextSqnum sqnum) (notificationSqnum s) }, sqnum)) case n of diff --git a/test/TestBrokerClient.hs b/test/TestBrokerClient.hs index 07ce6ec..b31d6ed 100644 --- a/test/TestBrokerClient.hs +++ b/test/TestBrokerClient.hs @@ -65,7 +65,7 @@ testBrokerClientStartStop = testCase "Broker client: submit order" $ withContext (ep, notifEp) <- makeEndpoints (ref, callback) <- makeNotificationCallback (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> bracket (startBrokerClient "foo" ctx ep notifEp [callback] defaultClientSecurityParams emptyLogger) stopBrokerClient (\broC -> do result <- submitOrder broC defaultOrder case result of @@ -84,7 +84,7 @@ testBrokerClientCancelOrder = testCase "Broker client: submit and cancel order" (ep, notifEp) <- makeEndpoints (ref, callback) <- makeNotificationCallback (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> bracket (startBrokerClient "foo" ctx ep notifEp [callback] defaultClientSecurityParams emptyLogger) stopBrokerClient (\broC -> do maybeOid <- submitOrder broC defaultOrder case maybeOid of @@ -100,7 +100,7 @@ testBrokerClientGetNotifications = testCase "Broker client: get notifications" $ (ep, notifEp) <- makeEndpoints (ref, callback) <- makeNotificationCallback (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\broS -> bracket (startBrokerClient "foo" ctx ep notifEp [callback] defaultClientSecurityParams emptyLogger) stopBrokerClient (\broC -> do maybeOid <- submitOrder broC defaultOrder case maybeOid of diff --git a/test/TestBrokerServer.hs b/test/TestBrokerServer.hs index e2e07d5..fb04002 100644 --- a/test/TestBrokerServer.hs +++ b/test/TestBrokerServer.hs @@ -96,7 +96,7 @@ makeTestTradeSink = do testBrokerServerStartStop :: TestTree testBrokerServerStartStop = testCase "Broker Server starts and stops" $ withContext (\ctx -> do (ep, notifEp) <- makeEndpoints - broS <- startBrokerServer [] ctx ep notifEp [] defaultServerSecurityParams emptyLogger + broS <- startBrokerServer [] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger stopBrokerServer broS) testBrokerServerSubmitOrder :: TestTree @@ -104,7 +104,7 @@ testBrokerServerSubmitOrder = testCaseSteps "Broker Server submits order" $ \ste step "Setup" (mockBroker, broState) <- mkMockBroker ["demo"] (ep, notifEp) <- makeEndpoints - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do withSocket ctx Req $ \sock -> do connectAndSendOrder step sock defaultOrder ep @@ -126,7 +126,7 @@ testBrokerServerSubmitOrderDifferentIdentities = testCaseSteps "Broker Server su (ep, notifEp) <- makeEndpoints let orderId1 = 42 let orderId2 = 76 - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do withSocket ctx Req $ \sock1 -> do withSocket ctx Req $ \sock2 -> do connectAndSendOrderWithIdentity step sock1 defaultOrder {orderId = orderId1} "identity1" ep @@ -156,7 +156,7 @@ testBrokerServerSubmitOrderToUnknownAccount = testCaseSteps "Broker Server retur step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, _) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> withSocket ctx Req (\sock -> do connectAndSendOrder step sock (defaultOrder { orderAccountId = "foobar" }) ep @@ -175,7 +175,7 @@ testBrokerServerCancelOrder = testCaseSteps "Broker Server: submitted order canc step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> withSocket ctx Req $ \sock -> do connectAndSendOrder step sock defaultOrder ep Just ResponseOk <- decode . BL.fromStrict <$> receive sock @@ -201,7 +201,7 @@ testBrokerServerCancelUnknownOrder = testCaseSteps "Broker Server: order cancell step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, _) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> withSocket ctx Req (\sock -> do connectAndSendOrder step sock defaultOrder ep receive sock @@ -224,7 +224,7 @@ testBrokerServerCorruptedPacket = testCaseSteps "Broker Server: corrupted packet step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, _) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer (\_ -> withSocket ctx Req (\sock -> do step "Connecting" connect sock (T.unpack ep) @@ -249,7 +249,7 @@ testBrokerServerGetNotifications = testCaseSteps "Broker Server: notifications r step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> withSocket ctx Req $ \sock -> do -- We have to actually submit order, or else server won't know that we should -- be notified about this order @@ -307,7 +307,7 @@ testBrokerServerGetNotificationsFromSameSqnum = testCaseSteps "Broker Server: no step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> withSocket ctx Req $ \sock -> do connectAndSendOrder step sock defaultOrder ep Just ResponseOk <- decode . BL.fromStrict <$> receive sock @@ -365,7 +365,7 @@ testBrokerServerGetNotificationsRemovesEarlierNotifications = testCaseSteps "Bro step "Setup" (ep, notifEp) <- makeEndpoints (mockBroker, broState) <- mkMockBroker ["demo"] - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> withSocket ctx Req $ \sock -> do connectAndSendOrder step sock defaultOrder ep Just ResponseOk <- decode . BL.fromStrict <$> receive sock @@ -418,7 +418,7 @@ testBrokerServerDuplicateRequest = testCaseSteps "Broker Server: duplicate reque step "Setup" (mockBroker, broState) <- mkMockBroker ["demo"] (ep, notifEp) <- makeEndpoints - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do withSocket ctx Req $ \sock -> do connectAndSendOrder step sock defaultOrder ep @@ -445,7 +445,7 @@ testBrokerServerNotificationSocket = testCaseSteps "Broker Server: sends notific (mockBroker, broState) <- mkMockBroker ["demo"] (ep, notifEp) <- makeEndpoints - bracket (startBrokerServer [mockBroker] ctx ep notifEp [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do + bracket (startBrokerServer [mockBroker] ctx ep notifEp (NotificationSqnum 1) [] defaultServerSecurityParams emptyLogger) stopBrokerServer $ \_ -> do withSocket ctx Req $ \sock -> do nSocket <- socket ctx Sub connect nSocket (T.unpack notifEp) @@ -486,40 +486,3 @@ testBrokerServerNotificationSocket = testCaseSteps "Broker Server: sends notific let (Just (TradeNotification notifSqnum2 incomingTrade)) = decode . BL.fromStrict $ payload incomingTrade @?= trade { tradeOrderId = orderId defaultOrder } -{- -testBrokerServerTradeSink :: TestTree -testBrokerServerTradeSink = testCaseSteps "Broker Server: sends trades to trade sink" $ \step -> withContext (\ctx -> do - step "Setup" - (mockBroker, broState) <- mkMockBroker ["demo"] - ep <- makeEndpoint - (tradeRef, sink) <- makeTestTradeSink - bracket (startBrokerServer [mockBroker] ctx ep [sink] defaultServerSecurityParams) stopBrokerServer (\_ -> do - withSocket ctx Req (\sock -> do - step "Connecting" - connectAndSendOrder step sock defaultOrder ep - (Just (ResponseOrderSubmitted orderId)) <- decode . BL.fromStrict <$> receive sock - - (Just cb) <- notificationCallback <$> readIORef broState - let trade = Trade { - tradeOrderId = orderId, - tradePrice = 19.82, - tradeQuantity = 1, - tradeVolume = 1982, - tradeVolumeCurrency = "TEST_CURRENCY", - tradeOperation = Buy, - tradeAccount = "demo", - tradeSecurity = "FOO", - tradeTimestamp = UTCTime (fromGregorian 2016 9 28) 16000, - tradeCommission = 0, - tradeSignalId = SignalId "Foo" "bar" "baz" } - cb (BackendTradeNotification trade) - - threadDelay 100000 - step "Testing" - maybeTrade <- readIORef tradeRef - case maybeTrade of - Just trade' -> do - trade' @?= trade - _ -> assertFailure "Invalid trade in sink" - ))) --}