diff --git a/package.yaml b/package.yaml index a6e2ef8e0..9b432d30e 100644 --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: simplexmq -version: 5.2.1 +version: 5.3.0.0 synopsis: SimpleXMQ message broker description: | This package includes <./docs/Simplex-Messaging-Server.html server>, diff --git a/simplexmq.cabal b/simplexmq.cabal index a3d6af40b..e61491639 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -5,7 +5,7 @@ cabal-version: 1.12 -- see: https://github.com/sol/hpack name: simplexmq -version: 5.2.1 +version: 5.3.0.0 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and @@ -109,6 +109,7 @@ library Simplex.Messaging.Protocol Simplex.Messaging.Server Simplex.Messaging.Server.CLI + Simplex.Messaging.Server.Control Simplex.Messaging.Server.Env.STM Simplex.Messaging.Server.Expiration Simplex.Messaging.Server.Main diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 9b2499c44..bf34f4df4 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1095,7 +1095,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl SMP SMP.QUOTA -> case msgType of AM_CONN_INFO -> connError msgId NOT_AVAILABLE AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE - _ -> retrySndMsg RISlow + _ -> retryIfNotExpired RISlow err SMP SMP.AUTH -> case msgType of AM_CONN_INFO -> connError msgId NOT_AVAILABLE AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE @@ -1124,11 +1124,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl _ -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server), -- the message sending would be retried - | temporaryOrHostError e -> do - let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout - ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast) + | temporaryOrHostError e -> retryIfNotExpired RIFast err | otherwise -> notifyDel msgId err where + retryIfNotExpired riMode err = do + let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout + ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg riMode) msgExpired timeoutSel = do msgTimeout <- asks $ timeoutSel . config currentTime <- liftIO getCurrentTime diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index dcd02ca5e..5826bcf7b 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -58,11 +58,13 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime) import Data.Time.Format.ISO8601 (iso8601Show) import Data.Type.Equality import GHC.TypeLits (KnownNat) -import Network.Socket (ServiceName) +import Network.Socket (ServiceName, Socket, socketToHandle) +import Simplex.Messaging.Agent.Lock import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Encoding (Encoding (smpEncode)) import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol +import Simplex.Messaging.Server.Control import Simplex.Messaging.Server.Env.STM import Simplex.Messaging.Server.Expiration import Simplex.Messaging.Server.MsgStore @@ -74,10 +76,11 @@ import Simplex.Messaging.Server.StoreLog import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport +import Simplex.Messaging.Transport.Buffer (trimCR) import Simplex.Messaging.Transport.Server import Simplex.Messaging.Util import System.Exit (exitFailure) -import System.IO (hPutStrLn) +import System.IO (hPutStrLn, hSetNewlineMode, universalNewlineMode) import System.Mem.Weak (deRefWeak) import UnliftIO.Concurrent import UnliftIO.Directory (doesFileExist, renameFile) @@ -110,19 +113,22 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do raceAny_ ( serverThread s subscribedQ subscribers subscriptions cancelSub : serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) : - map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg + map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg ) - `finally` (withLog closeStoreLog >> saveServerMessages >> saveServerStats) + `finally` withLock (savingLock s) "final" (saveServer False) where runServer :: (ServiceName, ATransport) -> M () runServer (tcpPort, ATransport t) = do serverParams <- asks tlsServerParams runTransportServer started tcpPort serverParams tCfg (runClient t) + saveServer :: Bool -> M () + saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerStats + serverThread :: forall s. Server -> - (Server -> TBQueue (QueueId, Client)) -> + (Server -> TQueue (QueueId, Client)) -> (Server -> TMap QueueId Client) -> (Client -> TMap QueueId s) -> (s -> IO ()) -> @@ -134,7 +140,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do where updateSubscribers :: STM (Maybe (QueueId, Client)) updateSubscribers = do - (qId, clnt) <- readTBQueue $ subQ s + (qId, clnt) <- readTQueue $ subQ s let clientToBeNotified = \c' -> if sameClientSession clnt c' then pure Nothing @@ -223,6 +229,57 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do Right th -> runClientTransport th Left _ -> pure () + controlPortThread_ :: ServerConfig -> [M ()] + controlPortThread_ ServerConfig {controlPort = Just port} = [runCPServer port] + controlPortThread_ _ = [] + + runCPServer :: ServiceName -> M () + runCPServer port = do + srv <- asks server + cpStarted <- newEmptyTMVarIO + u <- askUnliftIO + liftIO $ runTCPServer cpStarted port $ runCPClient u srv + where + runCPClient :: UnliftIO (ReaderT Env IO) -> Server -> Socket -> IO () + runCPClient u srv sock = do + h <- socketToHandle sock ReadWriteMode + hSetBuffering h LineBuffering + hSetNewlineMode h universalNewlineMode + hPutStrLn h "SMP server control port\n'help' for supported commands" + cpLoop h + where + cpLoop h = do + s <- B.hGetLine h + case strDecode $ trimCR s of + Right CPQuit -> hClose h + Right cmd -> processCP h cmd >> cpLoop h + Left err -> hPutStrLn h ("error: " <> err) >> cpLoop h + processCP h = \case + CPSuspend -> hPutStrLn h "suspend not implemented" + CPResume -> hPutStrLn h "resume not implemented" + CPClients -> hPutStrLn h "clients not implemented" + CPStats -> do + ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgSentNtf, msgRecvNtf, qCount, msgCount} <- unliftIO u $ asks serverStats + putStat "fromTime" fromTime + putStat "qCreated" qCreated + putStat "qSecured" qSecured + putStat "qDeleted" qDeleted + putStat "msgSent" msgSent + putStat "msgRecv" msgRecv + putStat "msgSentNtf" msgSentNtf + putStat "msgRecvNtf" msgRecvNtf + putStat "qCount" qCount + putStat "msgCount" msgCount + where + putStat :: Show a => String -> TVar a -> IO () + putStat label var = readTVarIO var >>= \v -> hPutStrLn h $ label <> ": " <> show v + CPSave -> withLock (savingLock srv) "control" $ do + hPutStrLn h "saving server state..." + unliftIO u $ saveServer True + hPutStrLn h "server state saved!" + CPHelp -> hPutStrLn h "commands: stats, save, help, quit" + CPQuit -> pure () + runClientTransport :: Transport c => THandle c -> M () runClientTransport th@THandle {thVersion, sessionId} = do q <- asks $ tbqSize . config @@ -477,7 +534,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv where newSub :: m (TVar Sub) newSub = time "SUB newSub" . atomically $ do - writeTBQueue subscribedQ (rId, clnt) + writeTQueue subscribedQ (rId, clnt) sub <- newTVar =<< newSubscription NoSub TM.insert rId sub subscriptions pure sub @@ -522,7 +579,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv subscribeNotifications :: m (Transmission BrokerMsg) subscribeNotifications = time "NSUB" . atomically $ do unlessM (TM.member queueId ntfSubscriptions) $ do - writeTBQueue ntfSubscribedQ (queueId, clnt) + writeTQueue ntfSubscribedQ (queueId, clnt) TM.insert queueId () ntfSubscriptions pure ok @@ -720,8 +777,8 @@ randomId n = do gVar <- asks idsDrg atomically (C.pseudoRandomBytes n gVar) -saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => m () -saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages +saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => Bool -> m () +saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages where saveMessages f = do logInfo $ "saving messages to file " <> T.pack f @@ -730,8 +787,9 @@ saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages readTVarIO ms >>= mapM_ (saveQueueMsgs ms h) . M.keys logInfo "messages saved" where + getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue saveQueueMsgs ms h rId = - atomically (flushMsgQueue ms rId) + atomically (getMessages ms rId) >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId) restoreServerMessages :: forall m. (MonadUnliftIO m, MonadReader Env m) => m () diff --git a/src/Simplex/Messaging/Server/Control.hs b/src/Simplex/Messaging/Server/Control.hs new file mode 100644 index 000000000..184e94775 --- /dev/null +++ b/src/Simplex/Messaging/Server/Control.hs @@ -0,0 +1,36 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedStrings #-} + +module Simplex.Messaging.Server.Control where + +import qualified Data.Attoparsec.ByteString.Char8 as A +import Simplex.Messaging.Encoding.String + +data ControlProtocol + = CPSuspend + | CPResume + | CPClients + | CPStats + | CPSave + | CPHelp + | CPQuit + +instance StrEncoding ControlProtocol where + strEncode = \case + CPSuspend -> "suspend" + CPResume -> "resume" + CPClients -> "clients" + CPStats -> "stats" + CPSave -> "save" + CPHelp -> "help" + CPQuit -> "quit" + strP = + A.takeTill (== ' ') >>= \case + "suspend" -> pure CPSuspend + "resume" -> pure CPResume + "clients" -> pure CPClients + "stats" -> pure CPStats + "save" -> pure CPSave + "help" -> pure CPHelp + "quit" -> pure CPQuit + _ -> fail "bad ControlProtocol command" diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 609438848..386355e80 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -19,6 +19,7 @@ import Data.X509.Validation (Fingerprint (..)) import Network.Socket (ServiceName) import qualified Network.TLS as T import Numeric.Natural +import Simplex.Messaging.Agent.Lock import Simplex.Messaging.Crypto (KeyHash (..)) import Simplex.Messaging.Protocol import Simplex.Messaging.Server.Expiration @@ -39,7 +40,7 @@ import UnliftIO.STM data ServerConfig = ServerConfig { transports :: [(ServiceName, ATransport)], tbqSize :: Natural, - serverTbqSize :: Natural, + -- serverTbqSize :: Natural, msgQueueQuota :: Int, queueIdBytes :: Int, msgIdBytes :: Int, @@ -70,7 +71,9 @@ data ServerConfig = ServerConfig -- | SMP client-server protocol version range smpServerVRange :: VersionRange, -- | TCP transport config - transportConfig :: TransportServerConfig + transportConfig :: TransportServerConfig, + -- | run listener on control port + controlPort :: Maybe ServiceName } defMsgExpirationDays :: Int64 @@ -103,10 +106,11 @@ data Env = Env } data Server = Server - { subscribedQ :: TBQueue (RecipientId, Client), + { subscribedQ :: TQueue (RecipientId, Client), subscribers :: TMap RecipientId Client, - ntfSubscribedQ :: TBQueue (NotifierId, Client), - notifiers :: TMap NotifierId Client + ntfSubscribedQ :: TQueue (NotifierId, Client), + notifiers :: TMap NotifierId Client, + savingLock :: Lock } data Client = Client @@ -127,13 +131,14 @@ data Sub = Sub delivered :: TMVar MsgId } -newServer :: Natural -> STM Server -newServer qSize = do - subscribedQ <- newTBQueue qSize +newServer :: STM Server +newServer = do + subscribedQ <- newTQueue subscribers <- TM.empty - ntfSubscribedQ <- newTBQueue qSize + ntfSubscribedQ <- newTQueue notifiers <- TM.empty - return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers} + savingLock <- createLock + return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock} newClient :: Natural -> Version -> ByteString -> SystemTime -> STM Client newClient qSize thVersion sessionId ts = do @@ -152,26 +157,25 @@ newSubscription subThread = do newEnv :: forall m. (MonadUnliftIO m, MonadRandom m) => ServerConfig -> m Env newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile} = do - server <- atomically $ newServer (serverTbqSize config) + server <- atomically newServer queueStore <- atomically newQueueStore msgStore <- atomically newMsgStore idsDrg <- drgNew >>= newTVarIO - storeLog <- liftIO $ openReadStoreLog `mapM` storeLogFile - s' <- restoreQueues queueStore `mapM` storeLog + storeLog <- restoreQueues queueStore `mapM` storeLogFile tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile let serverIdentity = KeyHash fp serverStats <- atomically . newServerStats =<< liftIO getCurrentTime - return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog = s', tlsServerParams, serverStats} + return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog, tlsServerParams, serverStats} where - restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode) - restoreQueues QueueStore {queues, senders, notifiers} s = do - (qs, s') <- liftIO $ readWriteStoreLog s + restoreQueues :: QueueStore -> FilePath -> m (StoreLog 'WriteMode) + restoreQueues QueueStore {queues, senders, notifiers} f = do + (qs, s) <- liftIO $ readWriteStoreLog f atomically $ do writeTVar queues =<< mapM newTVar qs writeTVar senders $! M.foldr' addSender M.empty qs writeTVar notifiers $! M.foldr' addNotifier M.empty qs - pure s' + pure s addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId addSender q = M.insert (senderId q) (recipientId q) addNotifier :: QueueRec -> Map NotifierId RecipientId -> Map NotifierId RecipientId diff --git a/src/Simplex/Messaging/Server/Main.hs b/src/Simplex/Messaging/Server/Main.hs index df15cebda..749a8d6ba 100644 --- a/src/Simplex/Messaging/Server/Main.hs +++ b/src/Simplex/Messaging/Server/Main.hs @@ -130,7 +130,8 @@ smpServerCLI cfgPath logPath = <> ("host: " <> host <> "\n") <> ("port: " <> defaultServerPort <> "\n") <> "log_tls_errors: off\n\ - \websockets: off\n\n\ + \websockets: off\n\ + \# control_port: 5224\n\n\ \[INACTIVE_CLIENTS]\n\ \# TTL and interval to check inactive clients\n\ \disconnect: off\n" @@ -166,7 +167,7 @@ smpServerCLI cfgPath logPath = ServerConfig { transports = iniTransports ini, tbqSize = 64, - serverTbqSize = 1024, + -- serverTbqSize = 1024, msgQueueQuota = 128, queueIdBytes = 24, msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20 @@ -202,7 +203,8 @@ smpServerCLI cfgPath logPath = transportConfig = defaultTransportServerConfig { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini - } + }, + controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini } data CliCommand diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index e9dd95eec..95e425d8e 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -13,6 +13,7 @@ module Simplex.Messaging.Server.MsgStore.STM getMsgQueue, delMsgQueue, flushMsgQueue, + snapshotMsgQueue, writeMsg, tryPeekMsg, peekMsg, @@ -62,6 +63,14 @@ delMsgQueue st rId = TM.delete rId st flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message] flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue) +snapshotMsgQueue :: STMMsgStore -> RecipientId -> STM [Message] +snapshotMsgQueue st rId = TM.lookup rId st >>= maybe (pure []) (snapshotTQueue . msgQueue) + where + snapshotTQueue q = do + msgs <- flushTQueue q + mapM_ (writeTQueue q) msgs + pure msgs + writeMsg :: MsgQueue -> Message -> STM (Maybe Message) writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do canWrt <- readTVar canWrite diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index b462eaffa..493bd5ac1 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/src/Simplex/Messaging/Server/Stats.hs @@ -94,7 +94,7 @@ setServerStats s d = do writeTVar (msgRecvNtf s) $! _msgRecvNtf d setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d) writeTVar (qCount s) $! _qCount d - writeTVar (msgCount s) $! _qCount d + writeTVar (msgCount s) $! _msgCount d instance StrEncoding ServerStatsData where strEncode ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _msgSentNtf, _msgRecvNtf, _activeQueues, _activeQueuesNtf} = diff --git a/src/Simplex/Messaging/Server/StoreLog.hs b/src/Simplex/Messaging/Server/StoreLog.hs index c6b276283..fceae16f4 100644 --- a/src/Simplex/Messaging/Server/StoreLog.hs +++ b/src/Simplex/Messaging/Server/StoreLog.hs @@ -25,20 +25,17 @@ module Simplex.Messaging.Server.StoreLog where import Control.Applicative (optional, (<|>)) -import Control.Monad (unless) -import Data.Bifunctor (first, second) +import Control.Monad (foldM, unless, when) import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B -import qualified Data.ByteString.Lazy.Char8 as LB -import Data.Either (partitionEithers) import Data.Functor (($>)) -import Data.List (foldl') import Data.Map.Strict (Map) import qualified Data.Map.Strict as M import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..)) import Simplex.Messaging.Transport.Buffer (trimCR) +import Simplex.Messaging.Util (ifM) import System.Directory (doesFileExist) import System.IO @@ -139,37 +136,33 @@ logDeleteQueue s = writeStoreLogRecord s . DeleteQueue logDeleteNotifier :: StoreLog 'WriteMode -> QueueId -> IO () logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier -readWriteStoreLog :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode) -readWriteStoreLog s@(ReadStoreLog f _) = do - qs <- readQueues s - closeStoreLog s - s' <- openWriteStoreLog f - writeQueues s' qs - pure (qs, s') +readWriteStoreLog :: FilePath -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode) +readWriteStoreLog f = do + qs <- ifM (doesFileExist f) (readQueues f) (pure M.empty) + s <- openWriteStoreLog f + writeQueues s qs + pure (qs, s) writeQueues :: StoreLog 'WriteMode -> Map RecipientId QueueRec -> IO () -writeQueues s = mapM_ (writeStoreLogRecord s . CreateQueue) . M.filter active +writeQueues s = mapM_ $ \q -> when (active q) $ logCreateQueue s q where active QueueRec {status} = status == QueueActive -type LogParsingError = (String, ByteString) - -readQueues :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec) -readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLog +readQueues :: FilePath -> IO (Map RecipientId QueueRec) +readQueues f = foldM processLine M.empty . B.lines =<< B.readFile f where - procStoreLog :: LB.ByteString -> ([LogParsingError], Map RecipientId QueueRec) - procStoreLog = second (foldl' procLogRecord M.empty) . partitionEithers . map parseLogRecord . LB.lines - returnResult :: ([LogParsingError], Map RecipientId QueueRec) -> IO (Map RecipientId QueueRec) - returnResult (errs, res) = mapM_ printError errs $> res - parseLogRecord :: LB.ByteString -> Either LogParsingError StoreLogRecord - parseLogRecord = (\s -> first (,s) $ strDecode s) . trimCR . LB.toStrict - procLogRecord :: Map RecipientId QueueRec -> StoreLogRecord -> Map RecipientId QueueRec - procLogRecord m = \case - CreateQueue q -> M.insert (recipientId q) q m - SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m - AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m - SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m - DeleteQueue qId -> M.delete qId m - DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m - printError :: LogParsingError -> IO () - printError (e, s) = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s + processLine :: Map RecipientId QueueRec -> ByteString -> IO (Map RecipientId QueueRec) + processLine m s = case strDecode $ trimCR s of + Right r -> pure $ procLogRecord r + Left e -> printError e $> m + where + procLogRecord :: StoreLogRecord -> Map RecipientId QueueRec + procLogRecord = \case + CreateQueue q -> M.insert (recipientId q) q m + SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m + AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m + SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m + DeleteQueue qId -> M.delete qId m + DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m + printError :: String -> IO () + printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s diff --git a/tests/SMPClient.hs b/tests/SMPClient.hs index c49b65ee6..639f254d0 100644 --- a/tests/SMPClient.hs +++ b/tests/SMPClient.hs @@ -82,7 +82,7 @@ cfg = ServerConfig { transports = undefined, tbqSize = 1, - serverTbqSize = 1, + -- serverTbqSize = 1, msgQueueQuota = 4, queueIdBytes = 24, msgIdBytes = 24, @@ -100,7 +100,8 @@ cfg = privateKeyFile = "tests/fixtures/server.key", certificateFile = "tests/fixtures/server.crt", smpServerVRange = supportedSMPServerVRange, - transportConfig = defaultTransportServerConfig + transportConfig = defaultTransportServerConfig, + controlPort = Nothing } withSmpServerStoreMsgLogOnV2 :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a