mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-28 07:14:59 +00:00
agent: remove protocol encodings and agent TCP server (#1189)
* rfc: remove agent protocol encodings * agent: remove protocol encodings and agent TCP server * update * remove unused code * remove * move tests * add delay to tests * stabilize test * test * more delays * reduce delays * enable all tests * delays * style
This commit is contained in:
Evgeny Poberezkin
@@ -324,8 +324,8 @@ xftpDeleteRcvFiles' c rcvFileEntityIds = do
|
||||
batchFiles :: (DB.Connection -> DBRcvFileId -> IO a) -> [RcvFile] -> AM' [Either AgentErrorType a]
|
||||
batchFiles f rcvFiles = withStoreBatch' c $ \db -> map (\RcvFile {rcvFileId} -> f db rcvFileId) rcvFiles
|
||||
|
||||
notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> EntityId -> ACommand 'Agent e -> m ()
|
||||
notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, APC (sAEntity @e) cmd)
|
||||
notify :: forall m e. (MonadIO m, AEntityI e) => AgentClient -> EntityId -> AEvent e -> m ()
|
||||
notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
xftpSendFile' :: AgentClient -> UserId -> CryptoFile -> Int -> AM SndFileId
|
||||
xftpSendFile' c userId file numRecipients = do
|
||||
|
||||
@@ -29,10 +29,7 @@
|
||||
--
|
||||
-- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md
|
||||
module Simplex.Messaging.Agent
|
||||
( -- * queue-based SMP agent
|
||||
runAgentClient,
|
||||
|
||||
-- * SMP agent functional API
|
||||
( -- * SMP agent functional API
|
||||
AgentClient (..),
|
||||
AE,
|
||||
SubscriptionsInfo (..),
|
||||
@@ -185,7 +182,6 @@ import Simplex.RemoteControl.Client
|
||||
import Simplex.RemoteControl.Invitation
|
||||
import Simplex.RemoteControl.Types
|
||||
import System.Mem.Weak (deRefWeak)
|
||||
import UnliftIO.Async (race_)
|
||||
import UnliftIO.Concurrent (forkFinally, forkIO, killThread, mkWeakThreadId, threadDelay)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
@@ -219,7 +215,7 @@ getSMPAgentClient_ clientId cfg initServers store backgroundMode =
|
||||
run AgentClient {subQ, acThread} name a =
|
||||
a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
|
||||
logError $ "Agent thread " <> name <> " crashed: " <> tshow e
|
||||
atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR $ CRITICAL True $ show e)
|
||||
atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ CRITICAL True $ show e)
|
||||
|
||||
disconnectAgentClient :: AgentClient -> IO ()
|
||||
disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do
|
||||
@@ -573,40 +569,6 @@ logConnection c connected =
|
||||
let event = if connected then "connected to" else "disconnected from"
|
||||
in logInfo $ T.unwords ["client", tshow (clientId c), event, "Agent"]
|
||||
|
||||
-- | Runs an SMP agent instance that receives commands and sends responses via 'TBQueue's.
|
||||
runAgentClient :: AgentClient -> AM' ()
|
||||
runAgentClient c = race_ (subscriber c) (client c)
|
||||
{-# INLINE runAgentClient #-}
|
||||
|
||||
client :: AgentClient -> AM' ()
|
||||
client c@AgentClient {rcvQ, subQ} = forever $ do
|
||||
(corrId, entId, cmd) <- atomically $ readTBQueue rcvQ
|
||||
runExceptT (processCommand c (entId, cmd))
|
||||
>>= atomically . writeTBQueue subQ . \case
|
||||
Left e -> (corrId, entId, APC SAEConn $ ERR e)
|
||||
Right (entId', resp) -> (corrId, entId', resp)
|
||||
|
||||
-- | execute any SMP agent command
|
||||
processCommand :: AgentClient -> (EntityId, APartyCmd 'Client) -> AM (EntityId, APartyCmd 'Agent)
|
||||
processCommand c (connId, APC e cmd) =
|
||||
second (APC e) <$> case cmd of
|
||||
NEW enableNtfs (ACM cMode) pqIK subMode -> second (INV . ACR cMode) <$> newConn c userId connId enableNtfs cMode Nothing pqIK subMode
|
||||
JOIN enableNtfs (ACR _ cReq) pqEnc subMode connInfo -> (,OK) <$> joinConn c userId connId False enableNtfs cReq connInfo pqEnc subMode
|
||||
LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK)
|
||||
ACPT invId pqEnc ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo pqEnc SMSubscribe
|
||||
RJCT invId -> rejectContact' c connId invId $> (connId, OK)
|
||||
SUB -> subscribeConnection' c connId $> (connId, OK)
|
||||
SEND pqEnc msgFlags msgBody -> (connId,) . uncurry MID <$> sendMessage' c connId pqEnc msgFlags msgBody
|
||||
ACK msgId rcptInfo_ -> ackMessage' c connId msgId rcptInfo_ $> (connId, OK)
|
||||
SWCH -> switchConnection' c connId $> (connId, OK)
|
||||
OFF -> suspendConnection' c connId $> (connId, OK)
|
||||
DEL -> deleteConnection' c connId $> (connId, OK)
|
||||
CHK -> (connId,) . STAT <$> getConnectionServers' c connId
|
||||
where
|
||||
-- command interface does not support different users
|
||||
userId :: UserId
|
||||
userId = 1
|
||||
|
||||
createUser' :: AgentClient -> NonEmpty SMPServerWithAuth -> NonEmpty XFTPServerWithAuth -> AM UserId
|
||||
createUser' c smp xftp = do
|
||||
userId <- withStore' c createUserRecord
|
||||
@@ -623,12 +585,12 @@ deleteUser' c userId delSMPQueues = do
|
||||
where
|
||||
delUser =
|
||||
whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", APC SAENone $ DEL_USER userId)
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAENone $ DEL_USER userId)
|
||||
|
||||
newConnAsync :: ConnectionModeI c => AgentClient -> UserId -> ACorrId -> Bool -> SConnectionMode c -> CR.InitialKeys -> SubscriptionMode -> AM ConnId
|
||||
newConnAsync c userId corrId enableNtfs cMode pqInitKeys subMode = do
|
||||
connId <- newConnNoQueues c userId "" enableNtfs cMode (CR.connPQEncryption pqInitKeys)
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn $ NEW enableNtfs (ACM cMode) pqInitKeys subMode
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ NEW enableNtfs (ACM cMode) pqInitKeys subMode
|
||||
pure connId
|
||||
|
||||
newConnNoQueues :: AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> PQSupport -> AM ConnId
|
||||
@@ -647,7 +609,7 @@ joinConnAsync c userId corrId enableNtfs cReqUri@CRInvitationUri {} cInfo pqSup
|
||||
let pqSupport = pqSup `CR.pqSupportAnd` versionPQSupport_ connAgentVersion (Just v)
|
||||
cData = ConnData {userId, connId = "", connAgentVersion, enableNtfs, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk, pqSupport}
|
||||
connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn $ JOIN enableNtfs (ACR sConnectionMode cReqUri) pqSupport subMode cInfo
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ JOIN enableNtfs (ACR sConnectionMode cReqUri) pqSupport subMode cInfo
|
||||
pure connId
|
||||
Nothing -> throwE $ AGENT A_VERSION
|
||||
joinConnAsync _c _userId _corrId _enableNtfs (CRContactUri _) _subMode _cInfo _pqEncryption =
|
||||
@@ -657,7 +619,7 @@ allowConnectionAsync' :: AgentClient -> ACorrId -> ConnId -> ConfirmationId -> C
|
||||
allowConnectionAsync' c corrId connId confId ownConnInfo =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (RcvConnection _ RcvQueue {server}) ->
|
||||
enqueueCommand c corrId connId (Just server) $ AClientCommand $ APC SAEConn $ LET confId ownConnInfo
|
||||
enqueueCommand c corrId connId (Just server) $ AClientCommand $ LET confId ownConnInfo
|
||||
_ -> throwE $ CMD PROHIBITED "allowConnectionAsync"
|
||||
|
||||
acceptContactAsync' :: AgentClient -> ACorrId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
|
||||
@@ -687,7 +649,7 @@ ackMessageAsync' c corrId connId msgId rcptInfo_ = do
|
||||
RcvMsg {msgType} <- withStore c $ \db -> getRcvMsg db connId mId
|
||||
when (isJust rcptInfo_ && msgType /= AM_A_MSG_) $ throwE $ CMD PROHIBITED "ackMessageAsync: receipt not allowed"
|
||||
(RcvQueue {server}, _) <- withStore c $ \db -> setMsgUserAck db connId mId
|
||||
enqueueCommand c corrId connId (Just server) . AClientCommand $ APC SAEConn $ ACK msgId rcptInfo_
|
||||
enqueueCommand c corrId connId (Just server) . AClientCommand $ ACK msgId rcptInfo_
|
||||
|
||||
deleteConnectionAsync' :: AgentClient -> Bool -> ConnId -> AM ()
|
||||
deleteConnectionAsync' c waitDelivery connId = deleteConnectionsAsync' c waitDelivery [connId]
|
||||
@@ -717,7 +679,7 @@ switchConnectionAsync' c corrId connId =
|
||||
| otherwise -> do
|
||||
when (ratchetSyncSendProhibited cData) $ throwE $ CMD PROHIBITED "switchConnectionAsync: send prohibited"
|
||||
rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSwitchStarted
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn SWCH
|
||||
enqueueCommand c corrId connId Nothing $ AClientCommand SWCH
|
||||
let rqs' = updatedQs rq1 rqs
|
||||
pure . connectionStats $ DuplexConnection cData rqs' sqs
|
||||
_ -> throwE $ CMD PROHIBITED "switchConnectionAsync: not duplex"
|
||||
@@ -984,7 +946,7 @@ subscribeConnections' c connIds = do
|
||||
let actual = M.size rs
|
||||
expected = length connIds
|
||||
when (actual /= expected) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
|
||||
resubscribeConnection' :: AgentClient -> ConnId -> AM ()
|
||||
resubscribeConnection' c connId = toConnResult connId =<< resubscribeConnections' c [connId]
|
||||
@@ -1114,7 +1076,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
|
||||
where
|
||||
processCmd :: RetryInterval -> PendingCommand -> AM ()
|
||||
processCmd ri PendingCommand {cmdId, corrId, userId, connId, command} = case command of
|
||||
AClientCommand (APC _ cmd) -> case cmd of
|
||||
AClientCommand cmd -> case cmd of
|
||||
NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do
|
||||
usedSrvs <- newTVarIO ([] :: [SMPServer])
|
||||
tryCommand . withNextSrv c userId usedSrvs [] $ \srv -> do
|
||||
@@ -1223,8 +1185,8 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
|
||||
tryWithLock name = tryCommand . withConnLock c connId name
|
||||
internalErr s = cmdError $ INTERNAL $ s <> ": " <> show (agentCommandTag command)
|
||||
cmdError e = notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
|
||||
notify :: forall e. AEntityI e => ACommand 'Agent e -> AM ()
|
||||
notify cmd = atomically $ writeTBQueue subQ (corrId, connId, APC (sAEntity @e) cmd)
|
||||
notify :: forall e. AEntityI e => AEvent e -> AM ()
|
||||
notify cmd = atomically $ writeTBQueue subQ (corrId, connId, AEvt (sAEntity @e) cmd)
|
||||
-- ^ ^ ^ async command processing /
|
||||
|
||||
enqueueMessages :: AgentClient -> ConnData -> NonEmpty SndQueue -> MsgFlags -> AMessage -> AM (AgentMsgId, PQEncryption)
|
||||
@@ -1460,9 +1422,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
|
||||
delMsg = delMsgKeep False
|
||||
delMsgKeep :: Bool -> InternalId -> AM ()
|
||||
delMsgKeep keepForReceipt msgId = withStore' c $ \db -> deleteSndMsgDelivery db connId sq msgId keepForReceipt
|
||||
notify :: forall e. AEntityI e => ACommand 'Agent e -> AM ()
|
||||
notify cmd = atomically $ writeTBQueue subQ ("", connId, APC (sAEntity @e) cmd)
|
||||
notifyDel :: AEntityI e => InternalId -> ACommand 'Agent e -> AM ()
|
||||
notify :: forall e. AEntityI e => AEvent e -> AM ()
|
||||
notify cmd = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) cmd)
|
||||
notifyDel :: AEntityI e => InternalId -> AEvent e -> AM ()
|
||||
notifyDel msgId cmd = notify cmd >> delMsg msgId
|
||||
connError msgId = notifyDel msgId . ERR . CONN
|
||||
qError msgId = notifyDel msgId . ERR . AGENT . A_QUEUE
|
||||
@@ -1663,7 +1625,7 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
|
||||
-- ! between completed deletions of connections, and deletions delayed due to wait for delivery (see deleteConn)
|
||||
deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing
|
||||
rs' <- lift $ catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) (M.keys delRs))
|
||||
forM_ rs' $ \cId -> notify ("", cId, APC SAEConn DEL_CONN)
|
||||
forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN)
|
||||
pure (errs' <> delRs, rqs, connIds')
|
||||
where
|
||||
rcvQueues :: SomeConn -> Either (Either AgentErrorType ()) [RcvQueue]
|
||||
@@ -1678,7 +1640,7 @@ deleteConnQueues c waitDelivery ntf rqs = do
|
||||
let connIds = M.keys $ M.filter isRight rs
|
||||
deliveryTimeout <- if waitDelivery then asks (Just . connDeleteDeliveryTimeout . config) else pure Nothing
|
||||
rs' <- catMaybes . rights <$> withStoreBatch' c (\db -> map (deleteConn db deliveryTimeout) connIds)
|
||||
forM_ rs' $ \cId -> notify ("", cId, APC SAEConn DEL_CONN)
|
||||
forM_ rs' $ \cId -> notify ("", cId, AEvt SAEConn DEL_CONN)
|
||||
pure rs
|
||||
where
|
||||
deleteQueueRecs :: [(RcvQueue, Either AgentErrorType ())] -> AM' [(RcvQueue, Either AgentErrorType ())]
|
||||
@@ -1698,7 +1660,7 @@ deleteConnQueues c waitDelivery ntf rqs = do
|
||||
Left e
|
||||
| temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> incRcvDeleteErrors db rq $> ((rq, r), Nothing)
|
||||
| otherwise -> deleteConnRcvQueue db rq $> ((rq, Right ()), Just (notifyRQ rq (Just e)))
|
||||
notifyRQ rq e_ = notify ("", qConnId rq, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
|
||||
notifyRQ rq e_ = notify ("", qConnId rq, AEvt SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
|
||||
notify = when ntf . atomically . writeTBQueue (subQ c)
|
||||
connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
|
||||
connResults = M.map snd . foldl' addResult M.empty
|
||||
@@ -1735,7 +1697,7 @@ deleteConnections_ getConnections ntf waitDelivery c connIds = do
|
||||
let actual = M.size rs
|
||||
expected = length connIds
|
||||
when (actual /= expected) . atomically $
|
||||
writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)
|
||||
|
||||
getConnectionServers' :: AgentClient -> ConnId -> AM ConnectionStats
|
||||
getConnectionServers' c connId = do
|
||||
@@ -1968,7 +1930,7 @@ sendNtfConnCommands c cmd = do
|
||||
Just (ConnData {enableNtfs}, _) ->
|
||||
when enableNtfs . atomically $ writeTBQueue (ntfSubQ ns) (connId, cmd)
|
||||
_ ->
|
||||
atomically $ writeTBQueue (subQ c) ("", connId, APC SAEConn $ ERR $ INTERNAL "no connection data")
|
||||
atomically $ writeTBQueue (subQ c) ("", connId, AEvt SAEConn $ ERR $ INTERNAL "no connection data")
|
||||
|
||||
setNtfServers :: AgentClient -> [NtfServer] -> IO ()
|
||||
setNtfServers c = atomically . writeTVar (ntfServers c)
|
||||
@@ -2050,7 +2012,7 @@ cleanupManager c@AgentClient {subQ} = do
|
||||
run SFERR deleteExpiredReplicasForDeletion
|
||||
liftIO $ threadDelay' int
|
||||
where
|
||||
run :: forall e. AEntityI e => (AgentErrorType -> ACommand 'Agent e) -> AM () -> AM' ()
|
||||
run :: forall e. AEntityI e => (AgentErrorType -> AEvent e) -> AM () -> AM' ()
|
||||
run err a = do
|
||||
waitActive . runExceptT $ a `catchAgentError` (notify "" . err)
|
||||
step <- asks $ cleanupStepInterval . config
|
||||
@@ -2097,8 +2059,8 @@ cleanupManager c@AgentClient {subQ} = do
|
||||
deleteExpiredReplicasForDeletion = do
|
||||
rcvFilesTTL <- asks $ rcvFilesTTL . config
|
||||
withStore' c (`deleteDeletedSndChunkReplicasExpired` rcvFilesTTL)
|
||||
notify :: forall e. AEntityI e => EntityId -> ACommand 'Agent e -> AM ()
|
||||
notify entId cmd = atomically $ writeTBQueue subQ ("", entId, APC (sAEntity @e) cmd)
|
||||
notify :: forall e. AEntityI e => EntityId -> AEvent e -> AM ()
|
||||
notify entId cmd = atomically $ writeTBQueue subQ ("", entId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
data ACKd = ACKd | ACKPending
|
||||
|
||||
@@ -2151,8 +2113,8 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
atomically . whenM (isPendingSub connId) $ failSubscription c rq e
|
||||
lift $ notifyErr connId e
|
||||
isPendingSub connId = (&&) <$> hasPendingSubscription c connId <*> activeClientSession c tSess sessId
|
||||
notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> ACommand 'Agent e -> m ()
|
||||
notify' connId msg = atomically $ writeTBQueue subQ ("", connId, APC (sAEntity @e) msg)
|
||||
notify' :: forall e m. (AEntityI e, MonadIO m) => ConnId -> AEvent e -> m ()
|
||||
notify' connId msg = atomically $ writeTBQueue subQ ("", connId, AEvt (sAEntity @e) msg)
|
||||
notifyErr :: ConnId -> SMPClientError -> AM' ()
|
||||
notifyErr connId = notify' connId . ERR . protocolClientError SMP (B.unpack $ strEncode srv)
|
||||
processSMP :: forall c. RcvQueue -> Connection c -> ConnData -> BrokerMsg -> AM ()
|
||||
@@ -2343,7 +2305,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(_, srv, _), _v, sessId, ts)
|
||||
SMP.ERR e -> notify $ ERR $ SMP (B.unpack $ strEncode srv) e
|
||||
r -> unexpected r
|
||||
where
|
||||
notify :: forall e m. (AEntityI e, MonadIO m) => ACommand 'Agent e -> m ()
|
||||
notify :: forall e m. (AEntityI e, MonadIO m) => AEvent e -> m ()
|
||||
notify = notify' connId
|
||||
|
||||
prohibited :: String -> AM ()
|
||||
|
||||
@@ -273,8 +273,7 @@ type XFTPTransportSession = TransportSession FileResponse
|
||||
data AgentClient = AgentClient
|
||||
{ acThread :: TVar (Maybe (Weak ThreadId)),
|
||||
active :: TVar Bool,
|
||||
rcvQ :: TBQueue (ATransmission 'Client),
|
||||
subQ :: TBQueue (ATransmission 'Agent),
|
||||
subQ :: TBQueue ATransmission,
|
||||
msgQ :: TBQueue (ServerTransmissionBatch SMPVersion ErrorType BrokerMsg),
|
||||
smpServers :: TMap UserId (NonEmpty SMPServerWithAuth),
|
||||
smpClients :: TMap SMPTransportSession SMPClientVar,
|
||||
@@ -373,7 +372,7 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
|
||||
notifyErr err = do
|
||||
let e = either ((", error: " <>) . show) (\_ -> ", no error") e_
|
||||
msg = "Worker " <> name <> " for " <> show key <> " terminated " <> show (restartCount rc) <> " times" <> e
|
||||
writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err msg)
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err msg)
|
||||
|
||||
newWorker :: AgentClient -> STM Worker
|
||||
newWorker c = do
|
||||
@@ -449,7 +448,6 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
|
||||
qSize = tbqSize cfg
|
||||
acThread <- newTVar Nothing
|
||||
active <- newTVar True
|
||||
rcvQ <- newTBQueue qSize
|
||||
subQ <- newTBQueue qSize
|
||||
msgQ <- newTBQueue qSize
|
||||
smpServers <- newTVar smp
|
||||
@@ -487,7 +485,6 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv =
|
||||
AgentClient
|
||||
{ acThread,
|
||||
active,
|
||||
rcvQ,
|
||||
subQ,
|
||||
msgQ,
|
||||
smpServers,
|
||||
@@ -586,7 +583,7 @@ instance ProtocolServerClient XFTPVersion XFTPErrorType FileResponse where
|
||||
|
||||
getSMPServerClient :: AgentClient -> SMPTransportSession -> AM SMPConnectedClient
|
||||
getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess smpClients ts)
|
||||
>>= either newClient (waitForProtocolClient c tSess smpClients)
|
||||
@@ -597,7 +594,7 @@ getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
|
||||
|
||||
getSMPProxyClient :: AgentClient -> SMPTransportSession -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay)
|
||||
getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq} destSess@(userId, destSrv, qId) = do
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
proxySrv <- getNextServer c userId [destSrv]
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getClientVar proxySrv ts) >>= \(tSess, auth, v) ->
|
||||
@@ -633,7 +630,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq
|
||||
liftIO $ incClientStat c userId clnt "PROXY" "OK"
|
||||
pure $ Right sess
|
||||
Left e -> do
|
||||
liftIO $ incClientStat c userId clnt "PROXY" $ strEncode e
|
||||
liftIO $ incClientStat c userId clnt "PROXY" $ bshow e
|
||||
atomically $ do
|
||||
unless (serverHostError e) $ do
|
||||
removeSessVar rv destSrv prs
|
||||
@@ -692,8 +689,8 @@ smpClientDisconnected c@AgentClient {active, smpClients, smpProxiedRelays} tSess
|
||||
atomically $ mapM_ (releaseGetLock c) qs
|
||||
runReaderT (resubscribeSMPSession c tSess) env
|
||||
|
||||
notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
|
||||
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
|
||||
notifySub :: forall e. AEntityI e => ConnId -> AEvent e -> IO ()
|
||||
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
resubscribeSMPSession :: AgentClient -> SMPTransportSession -> AM' ()
|
||||
resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do
|
||||
@@ -743,12 +740,12 @@ reconnectSMPClient c tSess@(_, srv, _) qs = handleNotify $ do
|
||||
where
|
||||
handleNotify :: AM' () -> AM' ()
|
||||
handleNotify = E.handleAny $ notifySub "" . ERR . INTERNAL . show
|
||||
notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> AM' ()
|
||||
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, APC (sAEntity @e) cmd)
|
||||
notifySub :: forall e. AEntityI e => ConnId -> AEvent e -> AM' ()
|
||||
notifySub connId cmd = atomically $ writeTBQueue (subQ c) ("", connId, AEvt (sAEntity @e) cmd)
|
||||
|
||||
getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
|
||||
getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(userId, srv, _) = do
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess ntfClients ts)
|
||||
>>= either
|
||||
@@ -767,12 +764,12 @@ getNtfServerClient c@AgentClient {active, ntfClients, workerSeq} tSess@(userId,
|
||||
clientDisconnected v client = do
|
||||
atomically $ removeSessVar v tSess ntfClients
|
||||
incClientStat c userId client "DISCONNECT" ""
|
||||
atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
|
||||
getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(userId, srv, _) = do
|
||||
unlessM (readTVarIO active) . throwE $ INACTIVE
|
||||
unlessM (readTVarIO active) $ throwE INACTIVE
|
||||
ts <- liftIO getCurrentTime
|
||||
atomically (getSessVar workerSeq tSess xftpClients ts)
|
||||
>>= either
|
||||
@@ -791,7 +788,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(userId
|
||||
clientDisconnected v client = do
|
||||
atomically $ removeSessVar v tSess xftpClients
|
||||
incClientStat c userId client "DISCONNECT" ""
|
||||
atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent DISCONNECT client)
|
||||
logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
|
||||
|
||||
waitForProtocolClient ::
|
||||
@@ -831,10 +828,10 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
|
||||
logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
|
||||
atomically $ putTMVar (sessionVar v) (Right client)
|
||||
liftIO $ incClientStat c userId client "CLIENT" "OK"
|
||||
atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
|
||||
atomically $ writeTBQueue (subQ c) ("", "", AEvt SAENone $ hostEvent CONNECT client)
|
||||
pure client
|
||||
Left e -> do
|
||||
liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
|
||||
liftIO $ incServerStat c userId srv "CLIENT" $ bshow e
|
||||
ei <- asks $ persistErrorInterval . config
|
||||
if ei == 0
|
||||
then atomically $ do
|
||||
@@ -845,11 +842,11 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
|
||||
atomically $ putTMVar (sessionVar v) (Left (e, Just ts))
|
||||
throwE e -- signal error to caller
|
||||
|
||||
hostEvent :: forall v err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient v err msg) => (AProtocolType -> TransportHost -> ACommand 'Agent 'AENone) -> Client msg -> ACommand 'Agent 'AENone
|
||||
hostEvent :: forall v err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient v err msg) => (AProtocolType -> TransportHost -> AEvent 'AENone) -> Client msg -> AEvent 'AENone
|
||||
hostEvent event = hostEvent' event . protocolClient
|
||||
{-# INLINE hostEvent #-}
|
||||
|
||||
hostEvent' :: forall v err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient v err msg) => (AProtocolType -> TransportHost -> ACommand 'Agent 'AENone) -> ProtoClient msg -> ACommand 'Agent 'AENone
|
||||
hostEvent' :: forall v err msg. (ProtocolTypeI (ProtoType msg), ProtocolServerClient v err msg) => (AProtocolType -> TransportHost -> AEvent 'AENone) -> ProtoClient msg -> AEvent 'AENone
|
||||
hostEvent' event = event (AProtocolType $ protocolTypeI @(ProtoType msg)) . clientTransportHost
|
||||
|
||||
getClientConfig :: AgentClient -> (AgentConfig -> ProtocolClientConfig v) -> AM' (ProtocolClientConfig v)
|
||||
@@ -986,8 +983,8 @@ withClient_ c tSess@(userId, srv, _) statCmd action = do
|
||||
stat cl = liftIO . incClientStat c userId cl statCmd
|
||||
logServerError :: Client msg -> AgentErrorType -> AM a
|
||||
logServerError cl e = do
|
||||
logServer "<--" c srv "" $ strEncode e
|
||||
stat cl $ strEncode e
|
||||
logServer "<--" c srv "" $ bshow e
|
||||
stat cl $ bshow e
|
||||
throwE e
|
||||
|
||||
withProxySession :: AgentClient -> SMPTransportSession -> SMP.SenderId -> ByteString -> ((SMPConnectedClient, ProxiedRelay) -> AM a) -> AM a
|
||||
@@ -1005,8 +1002,8 @@ withProxySession c destSess@(userId, destSrv, _) entId cmdStr action = do
|
||||
proxySrv = showServer . protocolClientServer' . protocolClient
|
||||
logServerError :: SMPConnectedClient -> AgentErrorType -> AM a
|
||||
logServerError cl e = do
|
||||
logServer ("<-- " <> proxySrv cl <> " <") c destSrv "" $ strEncode e
|
||||
stat cl $ strEncode e
|
||||
logServer ("<-- " <> proxySrv cl <> " <") c destSrv "" $ bshow e
|
||||
stat cl $ bshow e
|
||||
throwE e
|
||||
|
||||
withLogClient_ :: ProtocolServerClient v err msg => AgentClient -> TransportSession msg -> EntityId -> ByteString -> (Client msg -> AM a) -> AM a
|
||||
@@ -1719,7 +1716,7 @@ withWork c doWork getWork action =
|
||||
Left e -> notifyErr INTERNAL e
|
||||
where
|
||||
noWork = liftIO $ noWorkToDo doWork
|
||||
notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err $ show e)
|
||||
notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", AEvt SAEConn $ ERR $ err $ show e)
|
||||
|
||||
noWorkToDo :: TMVar () -> IO ()
|
||||
noWorkToDo = void . atomically . tryTakeTMVar
|
||||
@@ -1762,7 +1759,7 @@ suspendOperation c op endedAction = do
|
||||
notifySuspended :: AgentClient -> STM ()
|
||||
notifySuspended c = do
|
||||
-- unsafeIOToSTM $ putStrLn "notifySuspended"
|
||||
writeTBQueue (subQ c) ("", "", APC SAENone SUSPENDED)
|
||||
writeTBQueue (subQ c) ("", "", AEvt SAENone SUSPENDED)
|
||||
writeTVar (agentState c) ASSuspended
|
||||
|
||||
endOperation :: AgentClient -> AgentOperation -> STM () -> STM ()
|
||||
|
||||
@@ -29,7 +29,7 @@ import Data.Time (UTCTime, addUTCTime, getCurrentTime)
|
||||
import Data.Time.Clock (diffUTCTime)
|
||||
import Simplex.Messaging.Agent.Client
|
||||
import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.Protocol (ACommand (..), APartyCmd (..), AgentErrorType (..), BrokerErrorType (..), ConnId, NotificationsMode (..), SAEntity (..))
|
||||
import Simplex.Messaging.Agent.Protocol (AEvent (..), AEvt (..), AgentErrorType (..), BrokerErrorType (..), ConnId, NotificationsMode (..), SAEntity (..))
|
||||
import Simplex.Messaging.Agent.RetryInterval
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Agent.Store.SQLite
|
||||
@@ -306,7 +306,7 @@ workerInternalError c connId internalErrStr = do
|
||||
|
||||
-- TODO change error
|
||||
notifyInternalError :: MonadIO m => AgentClient -> ConnId -> String -> m ()
|
||||
notifyInternalError AgentClient {subQ} connId internalErrStr = atomically $ writeTBQueue subQ ("", connId, APC SAEConn $ ERR $ INTERNAL internalErrStr)
|
||||
notifyInternalError AgentClient {subQ} connId internalErrStr = atomically $ writeTBQueue subQ ("", connId, AEvt SAEConn $ ERR $ INTERNAL internalErrStr)
|
||||
{-# INLINE notifyInternalError #-}
|
||||
|
||||
getNtfToken :: AM' (Maybe NtfToken)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,85 +0,0 @@
|
||||
{-# LANGUAGE FlexibleContexts #-}
|
||||
{-# LANGUAGE NamedFieldPuns #-}
|
||||
{-# LANGUAGE OverloadedStrings #-}
|
||||
{-# LANGUAGE ScopedTypeVariables #-}
|
||||
|
||||
module Simplex.Messaging.Agent.Server
|
||||
( -- * SMP agent over TCP
|
||||
runSMPAgent,
|
||||
runSMPAgentBlocking,
|
||||
)
|
||||
where
|
||||
|
||||
import Control.Logger.Simple (logInfo)
|
||||
import Control.Monad
|
||||
import Control.Monad.Reader
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Text.Encoding (decodeUtf8)
|
||||
import Network.Socket (ServiceName)
|
||||
import Simplex.Messaging.Agent
|
||||
import Simplex.Messaging.Agent.Client (newAgentClient)
|
||||
import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
|
||||
import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
|
||||
import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, loadTLSServerParams, runTransportServer)
|
||||
import Simplex.Messaging.Util (bshow)
|
||||
import UnliftIO.Async (race_)
|
||||
import qualified UnliftIO.Exception as E
|
||||
import UnliftIO.STM
|
||||
|
||||
-- | Runs an SMP agent as a TCP service using passed configuration.
|
||||
--
|
||||
-- See a full agent executable here: https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-agent/Main.hs
|
||||
runSMPAgent :: ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> IO ()
|
||||
runSMPAgent t cfg initServers store =
|
||||
runSMPAgentBlocking t cfg initServers store 0 =<< newEmptyTMVarIO
|
||||
|
||||
-- | Runs an SMP agent as a TCP service using passed configuration with signalling.
|
||||
--
|
||||
-- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
|
||||
-- and when it is disconnected from the TCP socket once the server thread is killed (False).
|
||||
runSMPAgentBlocking :: ATransport -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Int -> TMVar Bool -> IO ()
|
||||
runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile, certificateFile, privateKeyFile} initServers store initClientId started =
|
||||
case tcpPort of
|
||||
Just port -> newSMPAgentEnv cfg store >>= smpAgent t port
|
||||
Nothing -> E.throwIO $ userError "no agent port"
|
||||
where
|
||||
smpAgent :: forall c. Transport c => TProxy c -> ServiceName -> Env -> IO ()
|
||||
smpAgent _ port env = do
|
||||
-- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
|
||||
tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile Nothing
|
||||
clientId <- newTVarIO initClientId
|
||||
runTransportServer started port tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
|
||||
putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
|
||||
cId <- atomically $ stateTVar clientId $ \i -> (i + 1, i + 1)
|
||||
c <- atomically $ newAgentClient cId initServers env
|
||||
logConnection c True
|
||||
race_ (connectClient h c) (runAgentClient c `runReaderT` env)
|
||||
`E.finally` (disconnectAgentClient c)
|
||||
|
||||
connectClient :: Transport c => c -> AgentClient -> IO ()
|
||||
connectClient h c = race_ (send h c) (receive h c)
|
||||
|
||||
receive :: forall c. Transport c => c -> AgentClient -> IO ()
|
||||
receive h c@AgentClient {rcvQ, subQ} = forever $ do
|
||||
(corrId, entId, cmdOrErr) <- tGet SClient h
|
||||
case cmdOrErr of
|
||||
Right cmd -> write rcvQ (corrId, entId, cmd)
|
||||
Left e -> write subQ (corrId, entId, APC SAEConn $ ERR e)
|
||||
where
|
||||
write :: TBQueue (ATransmission p) -> ATransmission p -> IO ()
|
||||
write q t = do
|
||||
logClient c "-->" t
|
||||
atomically $ writeTBQueue q t
|
||||
|
||||
send :: Transport c => c -> AgentClient -> IO ()
|
||||
send h c@AgentClient {subQ} = forever $ do
|
||||
t <- atomically $ readTBQueue subQ
|
||||
tPut h t
|
||||
logClient c "<--" t
|
||||
|
||||
logClient :: AgentClient -> ByteString -> ATransmission a -> IO ()
|
||||
logClient AgentClient {clientId} dir (corrId, connId, APC _ cmd) = do
|
||||
logInfo . decodeUtf8 $ B.unwords [bshow clientId, dir, "A :", corrId, connId, B.takeWhile (/= ' ') $ serializeCommand cmd]
|
||||
@@ -47,7 +47,6 @@ import Simplex.Messaging.Protocol
|
||||
VersionSMPC,
|
||||
)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Util ((<$?>))
|
||||
|
||||
-- * Queue types
|
||||
|
||||
@@ -344,20 +343,20 @@ instance StrEncoding AgentCmdType where
|
||||
_ -> fail "bad AgentCmdType"
|
||||
|
||||
data AgentCommand
|
||||
= AClientCommand (APartyCmd 'Client)
|
||||
= AClientCommand ACommand
|
||||
| AInternalCommand InternalCommand
|
||||
|
||||
instance StrEncoding AgentCommand where
|
||||
strEncode = \case
|
||||
AClientCommand (APC _ cmd) -> strEncode (ACClient, Str $ serializeCommand cmd)
|
||||
AClientCommand cmd -> strEncode (ACClient, Str $ serializeCommand cmd)
|
||||
AInternalCommand cmd -> strEncode (ACInternal, cmd)
|
||||
strP =
|
||||
strP_ >>= \case
|
||||
ACClient -> AClientCommand <$> ((\(ACmd _ e cmd) -> checkParty $ APC e cmd) <$?> dbCommandP)
|
||||
ACClient -> AClientCommand <$> dbCommandP
|
||||
ACInternal -> AInternalCommand <$> strP
|
||||
|
||||
data AgentCommandTag
|
||||
= AClientCommandTag (APartyCmdTag 'Client)
|
||||
= AClientCommandTag ACommandTag
|
||||
| AInternalCommandTag InternalCommandTag
|
||||
deriving (Show)
|
||||
|
||||
@@ -436,7 +435,7 @@ instance StrEncoding InternalCommandTag where
|
||||
|
||||
agentCommandTag :: AgentCommand -> AgentCommandTag
|
||||
agentCommandTag = \case
|
||||
AClientCommand cmd -> AClientCommandTag $ aPartyCmdTag cmd
|
||||
AClientCommand cmd -> AClientCommandTag $ aCommandTag cmd
|
||||
AInternalCommand cmd -> AInternalCommandTag $ internalCmdTag cmd
|
||||
|
||||
internalCmdTag :: InternalCommand -> InternalCommandTag
|
||||
|
||||
Reference in New Issue
Block a user