dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-28 23:08:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

send servers in agent events (#478)

Browse Source 
* send servers in agent events

* remove some changes

* command/function to get connection servers

* getConnectionServers return type
This commit is contained in:
Evgeny Poberezkin
2022-07-20 13:56:43 +01:00
committed by GitHub
parent 0ab90cb204
commit d810db4eed
6 changed files with 80 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent.hs
+22 -4
View File
@@ -54,6 +54,7 @@ module Simplex.Messaging.Agent
ackMessage,
suspendConnection,
deleteConnection,
getConnectionServers,
setSMPServers,
setNtfServers,
registerNtfToken,
@@ -194,6 +195,10 @@ suspendConnection c = withAgentEnv c . suspendConnection' c
deleteConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
deleteConnection c = withAgentEnv c . deleteConnection' c
-- | get servers used for connection
getConnectionServers :: AgentErrorMonad m => AgentClient -> ConnId -> m ConnectionStats
getConnectionServers c = withAgentEnv c . getConnectionServers' c
-- | Change servers to be used for creating new queues
setSMPServers :: AgentErrorMonad m => AgentClient -> NonEmpty SMPServer -> m ()
setSMPServers c = withAgentEnv c . setSMPServers' c
@@ -273,6 +278,7 @@ processCommand c (connId, cmd) = case cmd of
ACK msgId -> ackMessage' c connId msgId $> (connId, OK)
OFF -> suspendConnection' c connId $> (connId, OK)
DEL -> deleteConnection' c connId $> (connId, OK)
CHK -> (connId,) . STAT <$> getConnectionServers' c connId
newConn :: AgentMonad m => AgentClient -> ConnId -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
newConn c connId cMode = do
@@ -708,6 +714,16 @@ deleteConnection' c connId =
ns <- asks ntfSupervisor
atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCDelete)
getConnectionServers' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
getConnectionServers' c connId = connServers <$> withStore c (`getConn` connId)
where
connServers :: SomeConn -> ConnectionStats
connServers = \case
SomeConn _ (RcvConnection _ RcvQueue {server}) -> ConnectionStats {rcvServers = [server], sndServers = []}
SomeConn _ (SndConnection _ SndQueue {server}) -> ConnectionStats {rcvServers = [], sndServers = [server]}
SomeConn _ (DuplexConnection _ RcvQueue {server = s1} SndQueue {server = s2}) -> ConnectionStats {rcvServers = [s1], sndServers = [s2]}
SomeConn _ (ContactConnection _ RcvQueue {server}) -> ConnectionStats {rcvServers = [server], sndServers = []}
-- | Change servers to be used for creating new queues, in Reader monad
setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
setSMPServers' c = atomically . writeTVar (smpServers c)
@@ -1064,7 +1080,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
parseMessage agentMsgBody >>= \case
AgentConnInfo connInfo ->
processConf connInfo SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = []} False
AgentConnInfoReply smpQueues connInfo -> do
AgentConnInfoReply smpQueues connInfo ->
processConf connInfo SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues = L.toList smpQueues} True
_ -> prohibited
where
@@ -1074,7 +1090,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
confId <- withStore c $ \db -> do
setHandshakeVersion db connId agentVersion duplexHS
createConfirmation db g newConfirmation
notify $ CONF confId connInfo
let srvs = map (\SMPQueueInfo {smpServer = s} -> s) $ smpReplyQueues senderConf
notify $ CONF confId srvs connInfo
_ -> prohibited
-- party accepting connection
(DuplexConnection _ _ sq, Nothing) -> do
@@ -1120,14 +1137,15 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
_ -> prohibited
smpInvitation :: ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
smpInvitation connReq cInfo = do
smpInvitation connReq@(CRInvitationUri crData _) cInfo = do
logServer "<--" c srv rId "MSG <KEY>"
case conn of
ContactConnection {} -> do
g <- asks idsDrg
let newInv = NewInvitation {contactConnId = connId, connReq, recipientConnInfo = cInfo}
invId <- withStore c $ \db -> createInvitation db g newInv
notify $ REQ invId cInfo
let srvs = L.map (\SMPQueueUri {smpServer = s} -> s) $ crSmpQueues crData
notify $ REQ invId srvs cInfo
_ -> prohibited
checkMsgIntegrity :: PrevExternalSndId -> ExternalSndId -> PrevRcvMsgHash -> ByteString -> MsgIntegrity
src/Simplex/Messaging/Agent/Client.hs
+1 -1
View File
@@ -342,7 +342,7 @@ newProtocolClient c srv clients connectClient reconnectClient clientVar = tryCon
atomically $ putTMVar clientVar r
successAction client
Left e -> do
if e == BROKER NETWORK || e == BROKER TIMEOUT
if temporaryAgentError e
then retryAction
else atomically $ do
putTMVar clientVar (Left e)
src/Simplex/Messaging/Agent/Protocol.hs
+33 -9
View File
@@ -44,6 +44,7 @@ module Simplex.Messaging.Agent.Protocol
SAParty (..),
MsgHash,
MsgMeta (..),
ConnectionStats (..),
SMPConfirmation (..),
AgentMsgEnvelope (..),
AgentMessage (..),
@@ -208,9 +209,9 @@ data ACommand (p :: AParty) where
NEW :: AConnectionMode -> ACommand Client -- response INV
INV :: AConnectionRequestUri -> ACommand Agent
JOIN :: AConnectionRequestUri -> ConnInfo -> ACommand Client -- response OK
CONF :: ConfirmationId -> ConnInfo -> ACommand Agent -- ConnInfo is from sender
CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand Agent -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
LET :: ConfirmationId -> ConnInfo -> ACommand Client -- ConnInfo is from client
REQ :: InvitationId -> ConnInfo -> ACommand Agent -- ConnInfo is from sender
REQ :: InvitationId -> L.NonEmpty SMPServer -> ConnInfo -> ACommand Agent -- ConnInfo is from sender
ACPT :: InvitationId -> ConnInfo -> ACommand Client -- ConnInfo is from client
RJCT :: InvitationId -> ACommand Client
INFO :: ConnInfo -> ACommand Agent
@@ -227,6 +228,8 @@ data ACommand (p :: AParty) where
ACK :: AgentMsgId -> ACommand Client
OFF :: ACommand Client
DEL :: ACommand Client
CHK :: ACommand Client
STAT :: ConnectionStats -> ACommand Agent
OK :: ACommand Agent
ERR :: AgentErrorType -> ACommand Agent
SUSPENDED :: ACommand Agent
@@ -235,6 +238,22 @@ deriving instance Eq (ACommand p)
deriving instance Show (ACommand p)
data ConnectionStats = ConnectionStats
{ rcvServers :: [SMPServer],
sndServers :: [SMPServer]
}
deriving (Eq, Show, Generic)
instance StrEncoding ConnectionStats where
strEncode ConnectionStats {rcvServers, sndServers} =
"rcv=" <> strEncodeList rcvServers <> " snd=" <> strEncodeList sndServers
strP = do
rcvServers <- "rcv=" *> strListP
sndServers <- " snd=" *> strListP
pure ConnectionStats {rcvServers, sndServers}
instance ToJSON ConnectionStats where toEncoding = J.genericToEncoding J.defaultOptions
data NotificationsMode = NMPeriodic | NMInstant
deriving (Eq, Show)
@@ -747,7 +766,7 @@ data AgentErrorType
AGENT {agentErr :: SMPAgentError}
| -- | agent implementation or dependency errors
INTERNAL {internalErr :: String}
deriving (Eq, Generic, Read, Show, Exception)
deriving (Eq, Generic, Show, Exception)
instance ToJSON AgentErrorType where
toJSON = J.genericToJSON $ sumTypeJSON id
@@ -882,6 +901,8 @@ commandP =
<|> "ACK " *> ackCmd
<|> "OFF" $> ACmd SClient OFF
<|> "DEL" $> ACmd SClient DEL
<|> "CHK" $> ACmd SClient CHK
<|> "STAT " *> statResp
<|> "ERR " *> agentError
<|> "CON" $> ACmd SAgent CON
<|> "OK" $> ACmd SAgent OK
@@ -889,9 +910,9 @@ commandP =
newCmd = ACmd SClient . NEW <$> strP
invResp = ACmd SAgent . INV <$> strP
joinCmd = ACmd SClient .: JOIN <$> strP_ <*> A.takeByteString
confMsg = ACmd SAgent .: CONF <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> A.takeByteString
letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
reqMsg = ACmd SAgent .: REQ <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> A.takeByteString
acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> A.takeByteString
rjctCmd = ACmd SClient . RJCT <$> A.takeByteString
infoCmd = ACmd SAgent . INFO <$> A.takeByteString
@@ -903,6 +924,7 @@ commandP =
msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> A.takeByteString
ackCmd = ACmd SClient . ACK <$> A.decimal
statResp = ACmd SAgent . STAT <$> strP
connections = strP `A.sepBy'` A.char ','
msgMetaP = do
integrity <- strP
@@ -922,9 +944,9 @@ serializeCommand = \case
NEW cMode -> "NEW " <> strEncode cMode
INV cReq -> "INV " <> strEncode cReq
JOIN cReq cInfo -> B.unwords ["JOIN", strEncode cReq, serializeBinary cInfo]
CONF confId cInfo -> B.unwords ["CONF", confId, serializeBinary cInfo]
CONF confId srvs cInfo -> B.unwords ["CONF", confId, strEncodeList srvs, serializeBinary cInfo]
LET confId cInfo -> B.unwords ["LET", confId, serializeBinary cInfo]
REQ invId cInfo -> B.unwords ["REQ", invId, serializeBinary cInfo]
REQ invId srvs cInfo -> B.unwords ["REQ", invId, strEncode srvs, serializeBinary cInfo]
ACPT invId cInfo -> B.unwords ["ACPT", invId, serializeBinary cInfo]
RJCT invId -> "RJCT " <> invId
INFO cInfo -> "INFO " <> serializeBinary cInfo
@@ -940,6 +962,8 @@ serializeCommand = \case
ACK mId -> "ACK " <> bshow mId
OFF -> "OFF"
DEL -> "DEL"
CHK -> "CHK"
STAT srvs -> "STAT " <> strEncode srvs
CON -> "CON"
ERR e -> "ERR " <> strEncode e
OK -> "OK"
@@ -1012,9 +1036,9 @@ tGet party h = liftIO (tGetRaw h) >>= tParseLoadBody
SEND msgFlags body -> SEND msgFlags <$$> getBody body
MSG msgMeta msgFlags body -> MSG msgMeta msgFlags <$$> getBody body
JOIN qUri cInfo -> JOIN qUri <$$> getBody cInfo
CONF confId cInfo -> CONF confId <$$> getBody cInfo
CONF confId srvs cInfo -> CONF confId srvs <$$> getBody cInfo
LET confId cInfo -> LET confId <$$> getBody cInfo
REQ invId cInfo -> REQ invId <$$> getBody cInfo
REQ invId srvs cInfo -> REQ invId srvs <$$> getBody cInfo
ACPT invId cInfo -> ACPT invId <$$> getBody cInfo
INFO cInfo -> INFO <$$> getBody cInfo
cmd -> pure $ Right cmd