mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-31 09:38:24 +00:00
Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a406159ed4 | ||
|
|
14780a47d6 | ||
|
|
4a66f68c55 | ||
|
|
d1e6147adf | ||
|
|
7d1fdadef0 | ||
|
|
09e2e75c42 | ||
|
|
dff5cad1be | ||
|
|
c1f5f9d846 | ||
|
|
9c4778b129 | ||
|
|
137afb68fe | ||
|
|
229e2607d7 | ||
|
|
c380c79560 | ||
|
|
c9994c3a2c |
@@ -1,3 +1,15 @@
|
||||
# 1.0.3
|
||||
|
||||
SMP server:
|
||||
- Reduce server message queue quota to 128 messages.
|
||||
|
||||
SMP agent:
|
||||
- Add "yes to migrations" option.
|
||||
- Make new SMP client attempt to reconnect on network error.
|
||||
- Reduce connection handshake expiration to 2 days.
|
||||
|
||||
JSON encoding of types used in simplex-chat, some other minor adjustments.
|
||||
|
||||
# 1.0.2
|
||||
|
||||
General:
|
||||
|
||||
@@ -277,8 +277,8 @@ runServer IniOptions {enableStoreLog, port, enableWebsockets} = do
|
||||
ServerConfig
|
||||
{ transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets],
|
||||
tbqSize = 16,
|
||||
serverTbqSize = 128,
|
||||
msgQueueQuota = 256,
|
||||
serverTbqSize = 64,
|
||||
msgQueueQuota = 128,
|
||||
queueIdBytes = 24,
|
||||
msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
|
||||
caCertificateFile = caCrtFile,
|
||||
|
||||
+1
-1
@@ -1,5 +1,5 @@
|
||||
name: simplexmq
|
||||
version: 1.0.2
|
||||
version: 1.0.3
|
||||
synopsis: SimpleXMQ message broker
|
||||
description: |
|
||||
This package includes <./docs/Simplex-Messaging-Server.html server>,
|
||||
|
||||
+1
-1
@@ -5,7 +5,7 @@ cabal-version: 1.12
|
||||
-- see: https://github.com/sol/hpack
|
||||
|
||||
name: simplexmq
|
||||
version: 1.0.2
|
||||
version: 1.0.3
|
||||
synopsis: SimpleXMQ message broker
|
||||
description: This package includes <./docs/Simplex-Messaging-Server.html server>,
|
||||
<./docs/Simplex-Messaging-Client.html client> and
|
||||
|
||||
@@ -305,16 +305,6 @@ subscribeConnection' c connId =
|
||||
SomeConn _ (DuplexConnection _ rq sq) -> do
|
||||
resumeMsgDelivery c connId sq
|
||||
subscribeQueue c rq connId
|
||||
case status (sq :: SndQueue) of
|
||||
Confirmed -> do
|
||||
-- TODO if there is no confirmation saved, just update the status without securing the queue
|
||||
AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
|
||||
withStore (`getAcceptedConfirmation` connId)
|
||||
secureQueue c rq senderKey
|
||||
withStore $ \st -> setRcvQueueStatus st rq Secured
|
||||
Secured -> pure ()
|
||||
Active -> pure ()
|
||||
_ -> throwError $ INTERNAL "unexpected queue status"
|
||||
SomeConn _ (SndConnection _ sq) -> do
|
||||
resumeMsgDelivery c connId sq
|
||||
case status (sq :: SndQueue) of
|
||||
|
||||
@@ -37,7 +37,7 @@ module Simplex.Messaging.Agent.Client
|
||||
where
|
||||
|
||||
import Control.Concurrent (forkIO)
|
||||
import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
|
||||
import Control.Concurrent.Async (Async, uninterruptibleCancel)
|
||||
import Control.Concurrent.STM (stateTVar)
|
||||
import Control.Logger.Simple
|
||||
import Control.Monad.Except
|
||||
@@ -62,8 +62,10 @@ import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
|
||||
import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError)
|
||||
import Simplex.Messaging.Version
|
||||
import System.Timeout (timeout)
|
||||
import UnliftIO (async)
|
||||
import UnliftIO.Exception (Exception, IOException)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
@@ -76,11 +78,13 @@ data AgentClient = AgentClient
|
||||
msgQ :: TBQueue SMPServerTransmission,
|
||||
smpClients :: TVar (Map SMPServer SMPClientVar),
|
||||
subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
|
||||
pendingSubscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
|
||||
subscrConns :: TVar (Map ConnId SMPServer),
|
||||
connMsgsQueued :: TVar (Map ConnId Bool),
|
||||
smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
|
||||
smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
|
||||
reconnections :: TVar [Async ()],
|
||||
asyncClients :: TVar [Async ()],
|
||||
clientId :: Int,
|
||||
agentEnv :: Env,
|
||||
smpSubscriber :: Async (),
|
||||
@@ -95,14 +99,16 @@ newAgentClient agentEnv = do
|
||||
msgQ <- newTBQueue qSize
|
||||
smpClients <- newTVar M.empty
|
||||
subscrSrvrs <- newTVar M.empty
|
||||
pendingSubscrSrvrs <- newTVar M.empty
|
||||
subscrConns <- newTVar M.empty
|
||||
connMsgsQueued <- newTVar M.empty
|
||||
smpQueueMsgQueues <- newTVar M.empty
|
||||
smpQueueMsgDeliveries <- newTVar M.empty
|
||||
reconnections <- newTVar []
|
||||
asyncClients <- newTVar []
|
||||
clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
|
||||
lock <- newTMVar ()
|
||||
return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
|
||||
return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}
|
||||
|
||||
-- | Agent monad with MonadReader Env and MonadError AgentErrorType
|
||||
type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
|
||||
@@ -133,20 +139,39 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
|
||||
pure smpVar
|
||||
|
||||
waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
|
||||
waitForSMPClient = liftIOEither . atomically . readTMVar
|
||||
waitForSMPClient smpVar = do
|
||||
SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config
|
||||
smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)
|
||||
liftEither $ case smpClient_ of
|
||||
Just (Right smpClient) -> Right smpClient
|
||||
Just (Left e) -> Left e
|
||||
Nothing -> Left $ BROKER TIMEOUT
|
||||
|
||||
newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
|
||||
newSMPClient smpVar =
|
||||
tryError connectClient >>= \r -> case r of
|
||||
Right smp -> do
|
||||
logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
|
||||
atomically $ putTMVar smpVar r
|
||||
pure smp
|
||||
Left e -> do
|
||||
atomically $ do
|
||||
putTMVar smpVar r
|
||||
modifyTVar smpClients $ M.delete srv
|
||||
throwError e
|
||||
newSMPClient smpVar = tryConnectClient pure tryConnectAsync
|
||||
where
|
||||
tryConnectClient :: (SMPClient -> m a) -> m () -> m a
|
||||
tryConnectClient successAction retryAction =
|
||||
tryError connectClient >>= \r -> case r of
|
||||
Right smp -> do
|
||||
logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
|
||||
atomically $ putTMVar smpVar r
|
||||
successAction smp
|
||||
Left e -> do
|
||||
if e == BROKER NETWORK || e == BROKER TIMEOUT
|
||||
then retryAction
|
||||
else atomically $ do
|
||||
putTMVar smpVar (Left e)
|
||||
modifyTVar smpClients $ M.delete srv
|
||||
throwError e
|
||||
tryConnectAsync :: m ()
|
||||
tryConnectAsync = do
|
||||
a <- async connectAsync
|
||||
atomically $ modifyTVar (asyncClients c) (a :)
|
||||
connectAsync :: m ()
|
||||
connectAsync = do
|
||||
ri <- asks $ reconnectInterval . config
|
||||
withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop
|
||||
|
||||
connectClient :: m SMPClient
|
||||
connectClient = do
|
||||
@@ -169,33 +194,43 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
|
||||
cs <- M.lookup srv <$> readTVar (subscrSrvrs c)
|
||||
modifyTVar (subscrSrvrs c) $ M.delete srv
|
||||
modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs
|
||||
mapM_ (modifyTVar (pendingSubscrSrvrs c) . addPendingSubs) cs
|
||||
return cs
|
||||
where
|
||||
addPendingSubs :: Map ConnId RcvQueue -> Map SMPServer (Map ConnId RcvQueue) -> Map SMPServer (Map ConnId RcvQueue)
|
||||
addPendingSubs cs = M.alter (Just . addSubs cs) srv
|
||||
addSubs cs = maybe cs (M.union cs)
|
||||
deleteKeys :: Ord k => Set k -> Map k a -> Map k a
|
||||
deleteKeys ks m = S.foldr' M.delete m ks
|
||||
|
||||
serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
|
||||
serverDown u cs = unless (M.null cs) $ do
|
||||
mapM_ (notifySub DOWN) $ M.keysSet cs
|
||||
a <- async . unliftIO u $ tryReconnectClient cs
|
||||
unliftIO u reconnectServer
|
||||
|
||||
reconnectServer :: m ()
|
||||
reconnectServer = do
|
||||
a <- async tryReconnectClient
|
||||
atomically $ modifyTVar (reconnections c) (a :)
|
||||
|
||||
tryReconnectClient :: Map ConnId RcvQueue -> m ()
|
||||
tryReconnectClient cs = do
|
||||
tryReconnectClient :: m ()
|
||||
tryReconnectClient = do
|
||||
ri <- asks $ reconnectInterval . config
|
||||
withRetryInterval ri $ \loop ->
|
||||
reconnectClient cs `catchError` const loop
|
||||
reconnectClient `catchError` const loop
|
||||
|
||||
reconnectClient :: Map ConnId RcvQueue -> m ()
|
||||
reconnectClient cs = do
|
||||
reconnectClient :: m ()
|
||||
reconnectClient = do
|
||||
withAgentLock c . withSMP c srv $ \smp -> do
|
||||
subs <- readTVarIO $ subscrConns c
|
||||
forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
|
||||
cs <- M.lookup srv <$> readTVarIO (pendingSubscrSrvrs c)
|
||||
forM_ (maybe [] M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
|
||||
when (isNothing $ M.lookup connId subs) $ do
|
||||
subscribeSMPQueue smp rcvPrivateKey rcvId
|
||||
`catchError` \case
|
||||
SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId
|
||||
e -> throwError e
|
||||
e@SMPResponseTimeout -> throwError e
|
||||
e@SMPNetworkError -> throwError e
|
||||
e -> liftIO $ notifySub (ERR $ smpClientError e) connId
|
||||
addSubscription c rq connId
|
||||
liftIO $ notifySub UP connId
|
||||
|
||||
@@ -206,6 +241,7 @@ closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
|
||||
closeAgentClient c = liftIO $ do
|
||||
closeSMPServerClients c
|
||||
cancelActions $ reconnections c
|
||||
cancelActions $ asyncClients c
|
||||
cancelActions $ smpQueueMsgDeliveries c
|
||||
|
||||
closeSMPServerClients :: AgentClient -> IO ()
|
||||
@@ -294,26 +330,40 @@ newRcvQueue_ a c srv = do
|
||||
|
||||
subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
|
||||
withLogSMP c server rcvId "SUB" $ \smp ->
|
||||
subscribeSMPQueue smp rcvPrivateKey rcvId
|
||||
addSubscription c rq connId
|
||||
addPendingSubscription c rq connId
|
||||
withLogSMP c server rcvId "SUB" $ \smp -> do
|
||||
liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case
|
||||
Left e -> do
|
||||
atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $
|
||||
removePendingSubscription c server connId
|
||||
throwError e
|
||||
Right _ -> addSubscription c rq connId
|
||||
|
||||
addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
addSubscription c rq@RcvQueue {server} connId = atomically $ do
|
||||
modifyTVar (subscrConns c) $ M.insert connId server
|
||||
modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
|
||||
where
|
||||
addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue
|
||||
addSub (Just cs) = M.insert connId rq cs
|
||||
addSub _ = M.singleton connId rq
|
||||
addSubs_ (subscrSrvrs c) rq connId
|
||||
removePendingSubscription c server connId
|
||||
|
||||
removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
|
||||
cs <- readTVar subscrConns
|
||||
writeTVar subscrConns $ M.delete connId cs
|
||||
mapM_
|
||||
(modifyTVar subscrSrvrs . M.alter (>>= delSub))
|
||||
(M.lookup connId cs)
|
||||
addPendingSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
|
||||
addPendingSubscription c rq connId =
|
||||
atomically $ addSubs_ (pendingSubscrSrvrs c) rq connId
|
||||
|
||||
addSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> RcvQueue -> ConnId -> STM ()
|
||||
addSubs_ ss rq@RcvQueue {server} connId = modifyTVar ss $ M.alter (Just . addSub) server
|
||||
where
|
||||
addSub = maybe (M.singleton connId rq) (M.insert connId rq)
|
||||
|
||||
removeSubscription :: MonadUnliftIO m => AgentClient -> ConnId -> m ()
|
||||
removeSubscription c@AgentClient {subscrConns} connId = atomically $ do
|
||||
server_ <- stateTVar subscrConns $ \cs -> (M.lookup connId cs, M.delete connId cs)
|
||||
mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
|
||||
|
||||
removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
|
||||
removePendingSubscription c = removeSubs_ (pendingSubscrSrvrs c)
|
||||
|
||||
removeSubs_ :: TVar (Map SMPServer (Map ConnId RcvQueue)) -> SMPServer -> ConnId -> STM ()
|
||||
removeSubs_ ss server connId = modifyTVar ss $ M.update delSub server
|
||||
where
|
||||
delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)
|
||||
delSub cs =
|
||||
@@ -370,7 +420,6 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
|
||||
withLogSMP c server rcvId "DEL" $ \smp ->
|
||||
deleteSMPQueue smp rcvPrivateKey rcvId
|
||||
|
||||
-- TODO this is just wrong
|
||||
sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
|
||||
sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg =
|
||||
withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do
|
||||
|
||||
@@ -35,6 +35,7 @@ data AgentConfig = AgentConfig
|
||||
tbqSize :: Natural,
|
||||
dbFile :: FilePath,
|
||||
dbPoolSize :: Int,
|
||||
yesToMigrations :: Bool,
|
||||
smpCfg :: SMPClientConfig,
|
||||
reconnectInterval :: RetryInterval,
|
||||
helloTimeout :: NominalDiffTime,
|
||||
@@ -50,9 +51,10 @@ defaultAgentConfig =
|
||||
smpServers = undefined, -- TODO move it elsewhere?
|
||||
cmdSignAlg = C.SignAlg C.SEd448,
|
||||
connIdBytes = 12,
|
||||
tbqSize = 16,
|
||||
tbqSize = 64,
|
||||
dbFile = "smp-agent.db",
|
||||
dbPoolSize = 4,
|
||||
yesToMigrations = False,
|
||||
smpCfg = smpDefaultConfig,
|
||||
reconnectInterval =
|
||||
RetryInterval
|
||||
@@ -60,7 +62,7 @@ defaultAgentConfig =
|
||||
increaseAfter = 10 * second,
|
||||
maxInterval = 10 * second
|
||||
},
|
||||
helloTimeout = 7 * nominalDay,
|
||||
helloTimeout = 2 * nominalDay,
|
||||
-- CA certificate private key is not needed for initialization
|
||||
-- ! we do not generate these
|
||||
caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
|
||||
@@ -79,9 +81,9 @@ data Env = Env
|
||||
}
|
||||
|
||||
newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
|
||||
newSMPAgentEnv cfg = do
|
||||
newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do
|
||||
idsDrg <- newTVarIO =<< drgNew
|
||||
store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
|
||||
store <- liftIO $ createSQLiteStore dbFile dbPoolSize Migrations.app yesToMigrations
|
||||
clientCounter <- newTVarIO 0
|
||||
randomServer <- newTVarIO =<< liftIO newStdGen
|
||||
return Env {config = cfg, store, idsDrg, clientCounter, randomServer}
|
||||
return Env {config, store, idsDrg, clientCounter, randomServer}
|
||||
|
||||
@@ -76,13 +76,13 @@ data SQLiteStore = SQLiteStore
|
||||
dbNew :: Bool
|
||||
}
|
||||
|
||||
createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
|
||||
createSQLiteStore dbFilePath poolSize migrations = do
|
||||
createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore
|
||||
createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do
|
||||
let dbDir = takeDirectory dbFilePath
|
||||
createDirectoryIfMissing False dbDir
|
||||
st <- connectSQLiteStore dbFilePath poolSize
|
||||
checkThreadsafe st
|
||||
migrateSchema st migrations
|
||||
migrateSchema st migrations yesToMigrations
|
||||
pure st
|
||||
|
||||
checkThreadsafe :: SQLiteStore -> IO ()
|
||||
@@ -94,15 +94,16 @@ checkThreadsafe st = withConnection st $ \db -> do
|
||||
Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
|
||||
_ -> return ()
|
||||
|
||||
migrateSchema :: SQLiteStore -> [Migration] -> IO ()
|
||||
migrateSchema st migrations = withConnection st $ \db -> do
|
||||
migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
|
||||
migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
|
||||
Migrations.initialize db
|
||||
Migrations.get db migrations >>= \case
|
||||
Left e -> confirmOrExit $ "Database error: " <> e
|
||||
Right [] -> pure ()
|
||||
Right ms -> do
|
||||
unless (dbNew st) $ do
|
||||
confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
|
||||
unless yesToMigrations $
|
||||
confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
|
||||
let f = dbFilePath st
|
||||
copyFile f (f <> ".bak")
|
||||
Migrations.run db ms
|
||||
|
||||
@@ -107,7 +107,7 @@ data SMPClientConfig = SMPClientConfig
|
||||
smpDefaultConfig :: SMPClientConfig
|
||||
smpDefaultConfig =
|
||||
SMPClientConfig
|
||||
{ qSize = 16,
|
||||
{ qSize = 64,
|
||||
defaultTransport = ("5223", transport @TLS),
|
||||
tcpTimeout = 4_000_000,
|
||||
smpPing = 30_000_000
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
{-# LANGUAGE CPP #-}
|
||||
{-# LANGUAGE LambdaCase #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
|
||||
@@ -97,13 +98,18 @@ enumJSON tagModifier =
|
||||
}
|
||||
|
||||
sumTypeJSON :: (String -> String) -> J.Options
|
||||
#if defined(darwin_HOST_OS)
|
||||
sumTypeJSON = singleFieldJSON
|
||||
#else
|
||||
sumTypeJSON = taggedObjectJSON
|
||||
#endif
|
||||
|
||||
taggedObjectJSON :: (String -> String) -> J.Options
|
||||
taggedObjectJSON tagModifier =
|
||||
J.defaultOptions
|
||||
{ J.sumEncoding = J.TaggedObject "type" "data",
|
||||
J.constructorTagModifier = tagModifier,
|
||||
J.allNullaryToStringTag = False,
|
||||
J.nullaryToObject = True,
|
||||
J.omitNothingFields = True
|
||||
}
|
||||
@@ -113,6 +119,7 @@ singleFieldJSON tagModifier =
|
||||
J.defaultOptions
|
||||
{ J.sumEncoding = J.ObjectWithSingleField,
|
||||
J.constructorTagModifier = tagModifier,
|
||||
J.allNullaryToStringTag = False,
|
||||
J.nullaryToObject = True,
|
||||
J.omitNothingFields = True
|
||||
}
|
||||
|
||||
@@ -96,7 +96,7 @@ supportedSMPVersions :: VersionRange
|
||||
supportedSMPVersions = mkVersionRange 1 1
|
||||
|
||||
simplexMQVersion :: String
|
||||
simplexMQVersion = "1.0.2"
|
||||
simplexMQVersion = "1.0.3"
|
||||
|
||||
-- * Transport connection class
|
||||
|
||||
|
||||
+40
-1
@@ -62,9 +62,11 @@ agentTests (ATransport t) = do
|
||||
smpAgentTest3_1_1 $ testSubscription t
|
||||
it "should send notifications to client when server disconnects" $
|
||||
smpAgentServerTest $ testSubscrNotification t
|
||||
describe "Message delivery" $ do
|
||||
describe "Message delivery and server reconnection" $ do
|
||||
it "should deliver messages after losing server connection and re-connecting" $
|
||||
smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
|
||||
it "should connect to the server when server goes up if it initially was down" $
|
||||
smpAgentTestN [] $ testServerConnectionAfterError t
|
||||
it "should deliver pending messages after agent restarting" $
|
||||
smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
|
||||
it "should concurrently deliver messages to connections without blocking" $
|
||||
@@ -300,6 +302,43 @@ testMsgDeliveryServerRestart t alice bob = do
|
||||
where
|
||||
withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
|
||||
|
||||
testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()
|
||||
testServerConnectionAfterError t _ = do
|
||||
withAgent1 $ \bob -> do
|
||||
withAgent2 $ \alice -> do
|
||||
withServer $ do
|
||||
connect (bob, "bob") (alice, "alice")
|
||||
|
||||
bob <# ("", "alice", DOWN)
|
||||
alice <# ("", "bob", DOWN)
|
||||
alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)
|
||||
alice #:# "nothing else delivered before the server is restarted"
|
||||
bob #:# "nothing else delivered before the server is restarted"
|
||||
|
||||
withAgent1 $ \bob -> do
|
||||
withAgent2 $ \alice -> do
|
||||
bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
|
||||
alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
|
||||
withServer $ do
|
||||
alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
|
||||
alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 4; _ -> False
|
||||
bob <# ("", "alice", UP)
|
||||
bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
|
||||
bob #: ("2", "alice", "ACK 4") #> ("2", "alice", OK)
|
||||
alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 5)
|
||||
alice <# ("", "bob", SENT 5)
|
||||
bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
|
||||
|
||||
removeFile testStoreLogFile
|
||||
removeFile testDB
|
||||
removeFile testDB2
|
||||
where
|
||||
withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
|
||||
withAgent1 = withAgent agentTestPort testDB
|
||||
withAgent2 = withAgent agentTestPort2 testDB2
|
||||
withAgent :: String -> String -> (c -> IO a) -> IO a
|
||||
withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
|
||||
|
||||
testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
|
||||
testMsgDeliveryAgentRestart t bob = do
|
||||
withAgent $ \alice -> do
|
||||
|
||||
@@ -51,7 +51,7 @@ createStore = do
|
||||
-- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
|
||||
-- IO operations on multiple similarly named files; error seems to be environment specific
|
||||
r <- randomIO :: IO Word32
|
||||
createSQLiteStore (testDB <> show r) 4 Migrations.app
|
||||
createSQLiteStore (testDB <> show r) 4 Migrations.app True
|
||||
|
||||
removeStore :: SQLiteStore -> IO ()
|
||||
removeStore store = do
|
||||
|
||||
Reference in New Issue
Block a user