BrokerServer: start NotificationSqnum from arbitrary value

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

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

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"
-      )))
--}