From acea477aab54b14da1b6586db2cf2422cbbf529f Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sun, 25 Jun 2023 08:27:50 +0100 Subject: [PATCH] batch subscriptions for ntf server (#778) * return updated ConnectionStats from switchConnectionAsync (#777) * batch ntf server subscriptions to SMP servers * refactor * fix filtering queues in reconnect * test, remove repeated grouping * enable all tests * reduce subscriptions count --------- Co-authored-by: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> --- src/Simplex/Messaging/Client.hs | 11 ++ src/Simplex/Messaging/Client/Agent.hs | 102 ++++++++++++++---- src/Simplex/Messaging/Notifications/Server.hs | 72 ++++++++----- .../Messaging/Notifications/Server/Env.hs | 5 +- .../Messaging/Notifications/Server/Store.hs | 5 +- src/Simplex/Messaging/Util.hs | 2 +- tests/AgentTests/NotificationTests.hs | 57 +++++++++- 7 files changed, 197 insertions(+), 57 deletions(-) diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 02fcbfc2d..6cd14fcda 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -41,6 +41,7 @@ module Simplex.Messaging.Client subscribeSMPQueues, getSMPMessage, subscribeSMPQueueNotifications, + subscribeSMPQueuesNtfs, secureSMPQueue, enableSMPQueueNotifications, disableSMPQueueNotifications, @@ -487,6 +488,16 @@ getSMPMessage c rpKey rId = subscribeSMPQueueNotifications :: SMPClient -> NtfPrivateSignKey -> NotifierId -> ExceptT SMPClientError IO () subscribeSMPQueueNotifications = okSMPCommand NSUB +-- | Subscribe to multiple SMP queues notifications batching commands if supported. +subscribeSMPQueuesNtfs :: SMPClient -> NonEmpty (NtfPrivateSignKey, NotifierId) -> IO (NonEmpty (Either SMPClientError ())) +subscribeSMPQueuesNtfs c qs = sendProtocolCommands c cs >>= mapM response + where + cs = L.map (\(npKey, nId) -> (Just npKey, nId, Cmd SNotifier NSUB)) qs + response r = pure $ case r of + Right OK -> Right () + Right r' -> Left . PCEUnexpectedResponse $ bshow r' + Left e -> Left e + -- | Secure the SMP queue by adding a sender public key. -- -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#secure-queue-command diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index b99e9d61d..dad9d709a 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -5,6 +5,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} {-# OPTIONS_GHC -fno-warn-orphans #-} @@ -16,27 +17,33 @@ import Control.Logger.Simple import Control.Monad.Except import Control.Monad.IO.Unlift import Control.Monad.Trans.Except +import Data.Bifunctor (first) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B +import Data.List (find, partition) +import Data.List.NonEmpty (NonEmpty) +import qualified Data.List.NonEmpty as L import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Data.Set (Set) import Data.Text.Encoding +import Data.Tuple (swap) import Numeric.Natural import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Client import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding.String -import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer) +import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer, NtfPrivateSignKey, NotifierId, RcvPrivateSignKey, RecipientId) import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport -import Simplex.Messaging.Util (catchAll_, tryE, unlessM, ($>>=)) +import Simplex.Messaging.Util (catchAll_, tryE, ($>>=)) import System.Timeout (timeout) -import UnliftIO (async, forConcurrently_) +import UnliftIO (async) import UnliftIO.Exception (Exception) import qualified UnliftIO.Exception as E import UnliftIO.STM +import Data.Either (isLeft) type SMPClientVar = TMVar (Either SMPClientError SMPClient) @@ -184,9 +191,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv = _ -> TM.insert srv sVar ps serverDown :: Map SMPSub C.APrivateSignKey -> IO () - serverDown ss = unless (M.null ss) . void . runExceptT $ do + serverDown ss = unless (M.null ss) $ do notify . CADisconnected srv $ M.keysSet ss - reconnectServer + void $ runExceptT reconnectServer reconnectServer :: ExceptT SMPClientError IO () reconnectServer = do @@ -201,27 +208,47 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv = reconnectClient :: ExceptT SMPClientError IO () reconnectClient = do withSMP ca srv $ \smp -> do - notify $ CAReconnected srv - cs <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca) - forConcurrently_ (maybe [] M.assocs cs) $ \sub@(s, _) -> - unlessM (atomically $ hasSub (srvSubs ca) srv s) $ - subscribe_ smp sub `catchE` handleError s + liftIO . notify $ CAReconnected srv + cs_ <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca) + forM_ cs_ $ \cs -> do + subs' <- filterM (fmap not . atomically . hasSub (srvSubs ca) srv . fst) $ M.assocs cs + let (nSubs, rSubs) = partition (isNotifier . fst . fst) subs' + nRs <- liftIO $ subscribe_ smp SPNotifier nSubs + rRs <- liftIO $ subscribe_ smp SPRecipient rSubs + case find isLeft $ nRs <> rRs of + Just (Left e) -> throwE e + _ -> pure () where - subscribe_ :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO () - subscribe_ smp sub@(s, _) = do - smpSubscribe smp sub - atomically $ addSubscription ca srv sub - notify $ CAResubscribed srv s + isNotifier = \case + SPNotifier -> True + SPRecipient -> False - handleError :: SMPSub -> SMPClientError -> ExceptT SMPClientError IO () - handleError s = \case - e@PCEResponseTimeout -> throwE e - e@PCENetworkError -> throwE e - e -> do - notify $ CASubError srv s e - atomically $ removePendingSubscription ca srv s + subscribe_ :: SMPClient -> SMPSubParty -> [(SMPSub, C.APrivateSignKey)] -> IO [Either SMPClientError ()] + subscribe_ smp party subs = + case L.nonEmpty subs of + Just subs' -> do + let subs'' = L.map (first snd) subs' + rs <- L.zip subs'' <$> smpSubscribeQueues party ca smp srv subs'' + rs' <- forM rs $ \(sub, r) -> do + let sub' = first (party,) sub + s = fst sub' + case snd r of + Right () -> do + atomically $ addSubscription ca srv sub' + notify $ CAResubscribed srv s + pure $ Right () + Left e -> do + case e of + PCEResponseTimeout -> pure $ Left e + PCENetworkError -> pure $ Left e + _ -> do + notify $ CASubError srv s e + atomically $ removePendingSubscription ca srv s + pure $ Right () + pure $ L.toList rs' + Nothing -> pure [] - notify :: SMPClientAgentEvent -> ExceptT SMPClientError IO () + notify :: SMPClientAgentEvent -> IO () notify evt = atomically $ writeTBQueue (agentQ ca) evt closeSMPClientAgent :: MonadUnliftIO m => SMPClientAgent -> m () @@ -263,6 +290,35 @@ subscribeQueue ca srv sub = do removePendingSubscription ca srv $ fst sub throwE e +subscribeQueuesSMP :: SMPClientAgent -> SMPServer -> NonEmpty (RecipientId, RcvPrivateSignKey) -> IO (NonEmpty (RecipientId, Either SMPClientError ())) +subscribeQueuesSMP = subscribeQueues_ SPRecipient + +subscribeQueuesNtfs :: SMPClientAgent -> SMPServer -> NonEmpty (NotifierId, NtfPrivateSignKey) -> IO (NonEmpty (NotifierId, Either SMPClientError ())) +subscribeQueuesNtfs = subscribeQueues_ SPNotifier + +subscribeQueues_ :: SMPSubParty -> SMPClientAgent -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ())) +subscribeQueues_ party ca srv subs = do + atomically $ forM_ subs $ addPendingSubscription ca srv . first (party,) + runExceptT (getSMPServerClient' ca srv) >>= \case + Left e -> pure $ L.map ((,Left e) . fst) subs + Right smp -> smpSubscribeQueues party ca smp srv subs + +smpSubscribeQueues :: SMPSubParty -> SMPClientAgent -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ())) +smpSubscribeQueues party ca smp srv subs = do + rs <- L.zip subs <$> subscribe smp (L.map swap subs) + atomically $ forM rs $ \(sub, r) -> (fst sub,) <$> case r of + Right () -> do + addSubscription ca srv $ first (party,) sub + pure $ Right () + Left e -> do + when (e /= PCENetworkError && e /= PCEResponseTimeout) $ + removePendingSubscription ca srv $ (party,) $ fst sub + pure $ Left e + where + subscribe = case party of + SPRecipient -> subscribeSMPQueues + SPNotifier -> subscribeSMPQueuesNtfs + showServer :: SMPServer -> ByteString showServer ProtocolServer {host, port} = strEncode host <> B.pack (if null port then "" else ':' : port) diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index ee5b3f508..209975d5b 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -18,10 +18,14 @@ import Control.Monad.Except import Control.Monad.Reader import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B +import Data.Function (on) import Data.Functor (($>)) import Data.Int (Int64) import Data.List (intercalate) +import Data.List.NonEmpty (NonEmpty(..)) +import qualified Data.List.NonEmpty as L import Data.Map.Strict (Map) +import qualified Data.Map.Strict as M import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1) import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime) @@ -55,6 +59,7 @@ import UnliftIO.Concurrent (forkIO, killThread, mkWeakThreadId, threadDelay) import UnliftIO.Directory (doesFileExist, renameFile) import UnliftIO.Exception import UnliftIO.STM +import Data.Bifunctor (second) runNtfServer :: NtfServerConfig -> IO () runNtfServer cfg = do @@ -141,11 +146,8 @@ ntfServer cfg@NtfServerConfig {transports, logTLSErrors} started = do resubscribe :: NtfSubscriber -> Map NtfSubscriptionId NtfSubData -> M () resubscribe NtfSubscriber {newSubQ} subs = do - d <- asks $ resubscribeDelay . config - forM_ subs $ \sub@NtfSubData {} -> - whenM (ntfShouldSubscribe <$> readTVarIO (subStatus sub)) $ do - atomically $ writeTBQueue newSubQ $ NtfSub sub - threadDelay d + subs' <- atomically $ filterM (fmap ntfShouldSubscribe . readTVar . subStatus) $ M.elems subs + mapM_ (atomically . writeTBQueue newSubQ . L.map NtfSub) $ L.nonEmpty subs' liftIO $ logInfo "SMP connections resubscribed" ntfSubscriber :: NtfSubscriber -> M () @@ -153,12 +155,18 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge raceAny_ [subscribe, receiveSMP, receiveAgent] where subscribe :: M () - subscribe = - forever $ - atomically (readTBQueue newSubQ) >>= \case - sub@(NtfSub NtfSubData {smpQueue = SMPQueueNtf {smpServer}}) -> do - SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber smpServer - atomically $ writeTQueue subscriberSubQ sub + subscribe = do + d <- asks $ resubscribeDelay . config + forever $ do + subs <- atomically (readTBQueue newSubQ) + let ss = L.groupBy ((==) `on` server) subs + forM_ ss $ \serverSubs -> do + SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber $ server $ L.head serverSubs + atomically $ writeTQueue subscriberSubQ serverSubs + when (length serverSubs > 10) $ threadDelay d + + server :: NtfEntityRec 'Subscription -> SMPServer + server (NtfSub sub) = ntfSubServer sub getSMPSubscriber :: SMPServer -> M SMPSubscriber getSMPSubscriber smpServer = @@ -173,21 +181,31 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge runSMPSubscriber :: SMPSubscriber -> M () runSMPSubscriber SMPSubscriber {newSubQ = subscriberSubQ} = - forever $ - atomically (peekTQueue subscriberSubQ) - >>= \(NtfSub NtfSubData {smpQueue, notifierKey}) -> do - updateSubStatus smpQueue NSPending - let SMPQueueNtf {smpServer, notifierId} = smpQueue - liftIO (runExceptT $ subscribeQueue ca smpServer ((SPNotifier, notifierId), notifierKey)) >>= \case - Right _ -> do - updateSubStatus smpQueue NSActive - void . atomically $ readTQueue subscriberSubQ - Left err -> do - handleSubError smpQueue err - case err of - PCEResponseTimeout -> pure () - PCENetworkError -> pure () - _ -> void . atomically $ readTQueue subscriberSubQ + forever $ do + subs <- atomically (peekTQueue subscriberSubQ) + let subs' = L.map (\(NtfSub sub) -> sub) subs + mapM_ (\NtfSubData {smpQueue} -> updateSubStatus smpQueue NSPending) subs' + rs <- liftIO $ subscribeQueues (server $ L.head subs) subs' + subs_ <- L.nonEmpty <$> foldM process [] rs + atomically $ do + void $ readTQueue subscriberSubQ + mapM_ (writeTQueue subscriberSubQ . L.map NtfSub) subs_ + where + process subs (sub@NtfSubData {smpQueue}, r) = case r of + Right _ -> updateSubStatus smpQueue NSActive $> subs + Left err -> do + handleSubError smpQueue err + pure $ case err of + PCEResponseTimeout -> sub : subs + PCENetworkError -> sub : subs + _ -> subs + + -- | Subscribe to queues. The list of results can have a different order. + subscribeQueues :: SMPServer -> NonEmpty NtfSubData -> IO (NonEmpty (NtfSubData, Either SMPClientError ())) + subscribeQueues srv subs = + L.map (second snd) . L.zip subs <$> subscribeQueuesNtfs ca srv (L.map sub subs) + where + sub NtfSubData {smpQueue = SMPQueueNtf {notifierId}, notifierKey} = (notifierId, notifierKey) receiveSMP :: M () receiveSMP = forever $ do @@ -504,7 +522,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu sub <- atomically $ mkNtfSubData subId newSub resp <- atomically (addNtfSubscription st subId sub) >>= \case - Just _ -> atomically (writeTBQueue newSubQ $ NtfSub sub) $> NRSubId subId + Just _ -> atomically (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId _ -> pure $ NRErr AUTH withNtfLog (`logCreateSubscription` sub) incNtfStat subCreated diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 3736dffb1..5bff6df21 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -12,6 +12,7 @@ import Control.Monad.IO.Unlift import Crypto.Random import Data.ByteString.Char8 (ByteString) import Data.Int (Int64) +import Data.List.NonEmpty (NonEmpty) import Data.Time.Clock (getCurrentTime) import Data.Time.Clock.System (SystemTime) import Data.Word (Word16) @@ -93,7 +94,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo data NtfSubscriber = NtfSubscriber { smpSubscribers :: TMap SMPServer SMPSubscriber, - newSubQ :: TBQueue (NtfEntityRec 'Subscription), + newSubQ :: TBQueue (NonEmpty (NtfEntityRec 'Subscription)), smpAgent :: SMPClientAgent } @@ -105,7 +106,7 @@ newNtfSubscriber qSize smpAgentCfg = do pure NtfSubscriber {smpSubscribers, newSubQ, smpAgent} data SMPSubscriber = SMPSubscriber - { newSubQ :: TQueue (NtfEntityRec 'Subscription), + { newSubQ :: TQueue (NonEmpty (NtfEntityRec 'Subscription)), subThreadId :: TVar (Maybe (Weak ThreadId)) } diff --git a/src/Simplex/Messaging/Notifications/Server/Store.hs b/src/Simplex/Messaging/Notifications/Server/Store.hs index ee413cab1..7be6b3d54 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store.hs +++ b/src/Simplex/Messaging/Notifications/Server/Store.hs @@ -20,7 +20,7 @@ import qualified Data.Set as S import Data.Word (Word16) import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Notifications.Protocol -import Simplex.Messaging.Protocol (NtfPrivateSignKey) +import Simplex.Messaging.Protocol (NtfPrivateSignKey, SMPServer) import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Util (whenM, ($>>=)) @@ -68,6 +68,9 @@ data NtfSubData = NtfSubData subStatus :: TVar NtfSubStatus } +ntfSubServer :: NtfSubData -> SMPServer +ntfSubServer NtfSubData {smpQueue = SMPQueueNtf {smpServer}} = smpServer + data NtfEntityRec (e :: NtfEntity) where NtfTkn :: NtfTknData -> NtfEntityRec 'Token NtfSub :: NtfSubData -> NtfEntityRec 'Subscription diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index 2804ff869..9db3901aa 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -16,7 +16,7 @@ import Data.Int (Int64) import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (decodeUtf8With) -import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds) +import Data.Time (NominalDiffTime) import UnliftIO.Async raceAny_ :: MonadUnliftIO m => [m a] -> m () diff --git a/tests/AgentTests/NotificationTests.hs b/tests/AgentTests/NotificationTests.hs index 83d973015..36eaf8f7c 100644 --- a/tests/AgentTests/NotificationTests.hs +++ b/tests/AgentTests/NotificationTests.hs @@ -19,8 +19,8 @@ import qualified Data.ByteString.Base64.URL as U import Data.ByteString.Char8 (ByteString) import Data.Text.Encoding (encodeUtf8) import NtfClient -import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2) -import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn, xit') +import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2) +import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, xit') import Simplex.Messaging.Agent import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers) import Simplex.Messaging.Agent.Protocol @@ -31,6 +31,7 @@ import Simplex.Messaging.Notifications.Server.Push.APNS import Simplex.Messaging.Notifications.Types (NtfToken (..)) import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsgMeta (..)) import qualified Simplex.Messaging.Protocol as SMP +import Simplex.Messaging.Server.Env.STM (ServerConfig (..)) import Simplex.Messaging.Transport (ATransport) import Simplex.Messaging.Util (tryE) import System.Directory (doesFileExist, removeFile) @@ -85,6 +86,10 @@ notificationTests t = it "should resume subscriptions after SMP server is restarted" $ \_ -> withAPNSMockServer $ \apns -> withNtfServer t $ testNotificationsSMPRestart t apns + describe "Notifications after SMP server restart" $ + it "should resume batched subscriptions after SMP server is restarted" $ \_ -> + withAPNSMockServer $ \apns -> + withNtfServer t $ testNotificationsSMPRestartBatch 100 t apns describe "should switch notifications to the new queue" $ testServerMatrix2 t $ \servers -> withAPNSMockServer $ \apns -> @@ -480,6 +485,52 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False liftIO $ killThread threadId +testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO () +testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do + a <- getSMPAgentClient' agentCfg initAgentServers2 testDB + b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2 + conns <- runServers $ do + conns <- forM [1 .. n :: Int] . const $ makeConnection a b + _ <- registerTestToken a "abcd" NMInstant apnsQ + liftIO $ threadDelay 1500000 + forM_ conns $ \(aliceId, bobId) -> do + msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello" + get b ##> ("", aliceId, SENT msgId) + void $ messageNotification apnsQ + get a =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False + ackMessage a bobId msgId + pure conns + + runRight_ @AgentErrorType $ do + ("", "", DOWN _ bcs1) <- nGet a + ("", "", DOWN _ bcs2) <- nGet a + liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns + ("", "", DOWN _ acs1) <- nGet b + ("", "", DOWN _ acs2) <- nGet b + liftIO $ length (acs1 <> acs2) `shouldBe` length conns + + runServers $ do + ("", "", UP _ bcs1) <- nGet a + ("", "", UP _ bcs2) <- nGet a + liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns + ("", "", UP _ acs1) <- nGet b + ("", "", UP _ acs2) <- nGet b + liftIO $ length (acs1 <> acs2) `shouldBe` length conns + liftIO $ threadDelay 1500000 + forM_ conns $ \(aliceId, bobId) -> do + msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello again" + get b ##> ("", aliceId, SENT msgId) + _ <- messageNotificationData a apnsQ + get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False + where + runServers :: ExceptT AgentErrorType IO a -> IO a + runServers a = do + withSmpServerStoreLogOn t testPort $ \t1 -> do + res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 -> + runRight a `finally` killThread t2 + killThread t1 + pure res + testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO () testSwitchNotifications servers APNSMockServer {apnsQ} = do a <- getSMPAgentClient' agentCfg servers testDB @@ -496,7 +547,7 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do get a =##> \case ("", c, Msg msg') -> c == bId && msg == msg'; _ -> False ackMessage a bId msgId testMessage "hello" - switchConnectionAsync a "" bId + _ <- switchConnectionAsync a "" bId switchComplete a bId b aId liftIO $ threadDelay 500000 testMessage "hello again"