mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-28 00:44:38 +00:00
correlation IDs and command tags for async commands (#519)
* agent protocol command tags * store/send async command correlation IDs * fix, refactor * delete failed command
This commit is contained in:
Evgeny Poberezkin
@@ -83,7 +83,7 @@ import Control.Monad.Reader
|
||||
import Crypto.Random (MonadRandom)
|
||||
import Data.Bifunctor (bimap, first, second)
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import Data.Composition ((.:), (.:.))
|
||||
import Data.Composition ((.:), (.:.), (.::))
|
||||
import Data.Functor (($>))
|
||||
import Data.List (deleteFirstsBy)
|
||||
import Data.List.NonEmpty (NonEmpty (..))
|
||||
@@ -146,20 +146,20 @@ resumeAgentClient c = atomically $ writeTVar (active c) True
|
||||
type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m)
|
||||
|
||||
-- | Create SMP agent connection (NEW command) asynchronously, synchronous response is new connection id
|
||||
createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId
|
||||
createConnectionAsync c enableNtfs cMode = withAgentEnv c $ newConnAsync c enableNtfs cMode
|
||||
createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId
|
||||
createConnectionAsync c corrId enableNtfs cMode = withAgentEnv c $ newConnAsync c corrId enableNtfs cMode
|
||||
|
||||
-- | Join SMP agent connection (JOIN command) asynchronously, synchronous response is new connection id
|
||||
joinConnectionAsync :: AgentErrorMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnectionAsync c enableNtfs = withAgentEnv c .: joinConnAsync c enableNtfs
|
||||
joinConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnectionAsync c corrId enableNtfs = withAgentEnv c .: joinConnAsync c corrId enableNtfs
|
||||
|
||||
-- | Allow connection to continue after CONF notification (LET command), no synchronous response
|
||||
allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync c = withAgentEnv c .:. allowConnectionAsync' c
|
||||
allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync c = withAgentEnv c .:: allowConnectionAsync' c
|
||||
|
||||
-- | Acknowledge message (ACK command) asynchronously, no synchronous response
|
||||
ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync c = withAgentEnv c .: ackMessageAsync' c
|
||||
ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync c = withAgentEnv c .:. ackMessageAsync' c
|
||||
|
||||
-- | Create SMP agent connection (NEW command)
|
||||
createConnection :: AgentErrorMonad m => AgentClient -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
@@ -315,17 +315,17 @@ processCommand c (connId, cmd) = case cmd of
|
||||
DEL -> deleteConnection' c connId $> (connId, OK)
|
||||
CHK -> (connId,) . STAT <$> getConnectionServers' c connId
|
||||
|
||||
newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId
|
||||
newConnAsync c enableNtfs cMode = do
|
||||
newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId
|
||||
newConnAsync c corrId enableNtfs cMode = do
|
||||
g <- asks idsDrg
|
||||
connAgentVersion <- asks $ maxVersion . smpAgentVRange . config
|
||||
let cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
connId <- withStore c $ \db -> createNewConn db g cData cMode
|
||||
enqueueCommand c connId Nothing $ NEW enableNtfs (ACM cMode)
|
||||
enqueueCommand c corrId connId Nothing $ NEW enableNtfs (ACM cMode)
|
||||
pure connId
|
||||
|
||||
joinConnAsync :: AgentMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnAsync c enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do
|
||||
joinConnAsync :: AgentMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
|
||||
joinConnAsync c corrId enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
case agentVRange `compatibleVersion` aVRange of
|
||||
Just (Compatible connAgentVersion) -> do
|
||||
@@ -333,21 +333,21 @@ joinConnAsync c enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRang
|
||||
let duplexHS = connAgentVersion /= 1
|
||||
cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS}
|
||||
connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation
|
||||
enqueueCommand c connId Nothing $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo
|
||||
enqueueCommand c corrId connId Nothing $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo
|
||||
pure connId
|
||||
_ -> throwError $ AGENT A_VERSION
|
||||
joinConnAsync _c _enableNtfs (CRContactUri _) _cInfo =
|
||||
joinConnAsync _c _corrId _enableNtfs (CRContactUri _) _cInfo =
|
||||
throwError $ CMD PROHIBITED
|
||||
|
||||
allowConnectionAsync' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync' c connId confId ownConnInfo =
|
||||
allowConnectionAsync' :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m ()
|
||||
allowConnectionAsync' c corrId connId confId ownConnInfo =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (RcvConnection _ RcvQueue {server}) ->
|
||||
enqueueCommand c connId (Just server) $ LET confId ownConnInfo
|
||||
enqueueCommand c corrId connId (Just server) $ LET confId ownConnInfo
|
||||
_ -> throwError $ CMD PROHIBITED
|
||||
|
||||
ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync' c connId msgId =
|
||||
ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m ()
|
||||
ackMessageAsync' c corrId connId msgId =
|
||||
withStore c (`getConn` connId) >>= \case
|
||||
SomeConn _ (DuplexConnection _ rq _) -> enqueueAck rq
|
||||
SomeConn _ (RcvConnection _ rq) -> enqueueAck rq
|
||||
@@ -357,7 +357,7 @@ ackMessageAsync' c connId msgId =
|
||||
where
|
||||
enqueueAck :: RcvQueue -> m ()
|
||||
enqueueAck RcvQueue {server} = do
|
||||
enqueueCommand c connId (Just server) $ ACK msgId
|
||||
enqueueCommand c corrId connId (Just server) $ ACK msgId
|
||||
|
||||
newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
newConn c connId asyncMode enableNtfs cMode =
|
||||
@@ -638,10 +638,10 @@ sendMessage' c connId msgFlags msg =
|
||||
|
||||
-- / async command processing v v v
|
||||
|
||||
enqueueCommand :: forall m. AgentMonad m => AgentClient -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m ()
|
||||
enqueueCommand c connId server aCommand = do
|
||||
enqueueCommand :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m ()
|
||||
enqueueCommand c corrId connId server aCommand = do
|
||||
resumeSrvCmds c server
|
||||
commandId <- withStore c $ \db -> runExceptT . liftIO $ createCommand db connId server aCommand
|
||||
commandId <- withStore' c $ \db -> createCommand db corrId connId server aCommand
|
||||
queuePendingCommands c server [commandId]
|
||||
|
||||
resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m ()
|
||||
@@ -688,36 +688,38 @@ runCommandProcessing c@AgentClient {subQ} server = do
|
||||
cmdId <- atomically $ readTQueue cq
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
|
||||
Left (e :: E.SomeException) -> notify "" $ ERR (INTERNAL $ show e)
|
||||
Right (connId, ACmd _ cmd) -> processCmd ri connId cmdId cmd
|
||||
Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e)
|
||||
Right (corrId, connId, ACmd _ cmd) -> processCmd ri corrId connId cmdId cmd
|
||||
where
|
||||
processCmd :: RetryInterval -> ConnId -> AsyncCmdId -> ACommand p -> m ()
|
||||
processCmd ri connId cmdId = \case
|
||||
processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand p -> m ()
|
||||
processCmd ri corrId connId cmdId = \case
|
||||
NEW enableNtfs (ACM cMode) -> do
|
||||
usedSrvs <- newTVarIO ([] :: [SMPServer])
|
||||
tryCommand . withNextSrv usedSrvs [] $ \srv -> do
|
||||
(_, cReq) <- newConnSrv c connId True enableNtfs cMode srv
|
||||
notify connId $ INV (ACR cMode cReq)
|
||||
notify $ INV (ACR cMode cReq)
|
||||
JOIN enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = SMPQueueUri {queueAddress} :| _} _)) connInfo -> do
|
||||
let initUsed = [smpServer (queueAddress :: SMPQueueAddress)]
|
||||
usedSrvs <- newTVarIO initUsed
|
||||
tryCommand . withNextSrv usedSrvs initUsed $ \srv ->
|
||||
tryCommand . withNextSrv usedSrvs initUsed $ \srv -> do
|
||||
void $ joinConnSrv c connId True enableNtfs cReq connInfo srv
|
||||
LET confId ownCInfo -> tryCommand $ allowConnection' c connId confId ownCInfo
|
||||
ACK msgId -> tryCommand $ ackMessage' c connId msgId
|
||||
cmd -> notify connId $ ERR $ INTERNAL $ "unsupported async command " <> show cmd
|
||||
notify OK
|
||||
LET confId ownCInfo -> tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
|
||||
ACK msgId -> tryCommand $ ackMessage' c connId msgId >> notify OK
|
||||
cmd -> notify $ ERR $ INTERNAL $ "unsupported async command " <> show (aCommandTag cmd)
|
||||
where
|
||||
tryCommand action = withRetryInterval ri $ \loop ->
|
||||
tryError action >>= \case
|
||||
Left e
|
||||
| temporaryAgentError e || e == BROKER HOST -> retryCommand loop
|
||||
| otherwise -> notify connId $ ERR e
|
||||
| otherwise -> notify (ERR e) >> withStore' c (`deleteCommand` cmdId)
|
||||
Right () -> withStore' c (`deleteCommand` cmdId)
|
||||
retryCommand loop = do
|
||||
-- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
atomically $ beginAgentOperation c AOSndNetwork
|
||||
loop
|
||||
notify cmd = atomically $ writeTBQueue subQ (corrId, connId, cmd)
|
||||
withNextSrv :: TVar [SMPServer] -> [SMPServer] -> (SMPServer -> m ()) -> m ()
|
||||
withNextSrv usedSrvs initUsed action = do
|
||||
used <- readTVarIO usedSrvs
|
||||
@@ -727,8 +729,6 @@ runCommandProcessing c@AgentClient {subQ} server = do
|
||||
let used' = if length used + 1 >= L.length srvs then initUsed else srv : used
|
||||
writeTVar usedSrvs used'
|
||||
action srv
|
||||
notify :: ConnId -> ACommand 'Agent -> m ()
|
||||
notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
|
||||
-- ^ ^ ^ async command processing /
|
||||
|
||||
enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId
|
||||
|
||||
@@ -40,9 +40,13 @@ module Simplex.Messaging.Agent.Protocol
|
||||
-- * SMP agent protocol types
|
||||
ConnInfo,
|
||||
ACommand (..),
|
||||
ACommandTag (..),
|
||||
aCommandTag,
|
||||
ACmd (..),
|
||||
ACmdTag (..),
|
||||
AParty (..),
|
||||
SAParty (..),
|
||||
APartyI (..),
|
||||
MsgHash,
|
||||
MsgMeta (..),
|
||||
ConnectionStats (..),
|
||||
@@ -119,7 +123,6 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.ByteString.Base64
|
||||
import Data.ByteString.Char8 (ByteString)
|
||||
import qualified Data.ByteString.Char8 as B
|
||||
import Data.Composition ((.:), (.:.))
|
||||
import Data.Functor (($>))
|
||||
import Data.Int (Int64)
|
||||
import Data.Kind (Type)
|
||||
@@ -213,6 +216,12 @@ instance TestEquality SAParty where
|
||||
testEquality SClient SClient = Just Refl
|
||||
testEquality _ _ = Nothing
|
||||
|
||||
class APartyI (p :: AParty) where sAParty :: SAParty p
|
||||
|
||||
instance APartyI Agent where sAParty = SAgent
|
||||
|
||||
instance APartyI Client where sAParty = SClient
|
||||
|
||||
data ACmd = forall p. ACmd (SAParty p) (ACommand p)
|
||||
|
||||
deriving instance Show ACmd
|
||||
@@ -255,6 +264,73 @@ deriving instance Eq (ACommand p)
|
||||
|
||||
deriving instance Show (ACommand p)
|
||||
|
||||
data ACmdTag = forall p. APartyI p => ACmdTag (SAParty p) (ACommandTag p)
|
||||
|
||||
data ACommandTag (p :: AParty) where
|
||||
NEW_ :: ACommandTag Client
|
||||
INV_ :: ACommandTag Agent
|
||||
JOIN_ :: ACommandTag Client
|
||||
CONF_ :: ACommandTag Agent
|
||||
LET_ :: ACommandTag Client
|
||||
REQ_ :: ACommandTag Agent
|
||||
ACPT_ :: ACommandTag Client
|
||||
RJCT_ :: ACommandTag Client
|
||||
INFO_ :: ACommandTag Agent
|
||||
CON_ :: ACommandTag Agent
|
||||
SUB_ :: ACommandTag Client
|
||||
END_ :: ACommandTag Agent
|
||||
CONNECT_ :: ACommandTag Agent
|
||||
DISCONNECT_ :: ACommandTag Agent
|
||||
DOWN_ :: ACommandTag Agent
|
||||
UP_ :: ACommandTag Agent
|
||||
SEND_ :: ACommandTag Client
|
||||
MID_ :: ACommandTag Agent
|
||||
SENT_ :: ACommandTag Agent
|
||||
MERR_ :: ACommandTag Agent
|
||||
MSG_ :: ACommandTag Agent
|
||||
ACK_ :: ACommandTag Client
|
||||
OFF_ :: ACommandTag Client
|
||||
DEL_ :: ACommandTag Client
|
||||
CHK_ :: ACommandTag Client
|
||||
STAT_ :: ACommandTag Agent
|
||||
OK_ :: ACommandTag Agent
|
||||
ERR_ :: ACommandTag Agent
|
||||
SUSPENDED_ :: ACommandTag Agent
|
||||
|
||||
deriving instance Show (ACommandTag p)
|
||||
|
||||
aCommandTag :: ACommand p -> ACommandTag p
|
||||
aCommandTag = \case
|
||||
NEW {} -> NEW_
|
||||
INV _ -> INV_
|
||||
JOIN {} -> JOIN_
|
||||
CONF {} -> CONF_
|
||||
LET {} -> LET_
|
||||
REQ {} -> REQ_
|
||||
ACPT {} -> ACPT_
|
||||
RJCT _ -> RJCT_
|
||||
INFO _ -> INFO_
|
||||
CON -> CON_
|
||||
SUB -> SUB_
|
||||
END -> END_
|
||||
CONNECT {} -> CONNECT_
|
||||
DISCONNECT {} -> DISCONNECT_
|
||||
DOWN {} -> DOWN_
|
||||
UP {} -> UP_
|
||||
SEND {} -> SEND_
|
||||
MID _ -> MID_
|
||||
SENT _ -> SENT_
|
||||
MERR {} -> MERR_
|
||||
MSG {} -> MSG_
|
||||
ACK _ -> ACK_
|
||||
OFF -> OFF_
|
||||
DEL -> DEL_
|
||||
CHK -> CHK_
|
||||
STAT _ -> STAT_
|
||||
OK -> OK_
|
||||
ERR _ -> ERR_
|
||||
SUSPENDED -> SUSPENDED_
|
||||
|
||||
data ConnectionStats = ConnectionStats
|
||||
{ rcvServers :: [SMPServer],
|
||||
sndServers :: [SMPServer]
|
||||
@@ -930,58 +1006,121 @@ networkCommandP = commandP A.takeByteString
|
||||
dbCommandP :: Parser ACmd
|
||||
dbCommandP = commandP $ A.take =<< (A.decimal <* "\n")
|
||||
|
||||
instance Encoding ACmdTag where
|
||||
smpEncode (ACmdTag _ cmd) = smpEncode cmd
|
||||
smpP =
|
||||
A.takeTill (== ' ') >>= \case
|
||||
"NEW" -> pure $ ACmdTag SClient NEW_
|
||||
"INV" -> pure $ ACmdTag SAgent INV_
|
||||
"JOIN" -> pure $ ACmdTag SClient JOIN_
|
||||
"CONF" -> pure $ ACmdTag SAgent CONF_
|
||||
"LET" -> pure $ ACmdTag SClient LET_
|
||||
"REQ" -> pure $ ACmdTag SAgent REQ_
|
||||
"ACPT" -> pure $ ACmdTag SClient ACPT_
|
||||
"RJCT" -> pure $ ACmdTag SClient RJCT_
|
||||
"INFO" -> pure $ ACmdTag SAgent INFO_
|
||||
"CON" -> pure $ ACmdTag SAgent CON_
|
||||
"SUB" -> pure $ ACmdTag SClient SUB_
|
||||
"END" -> pure $ ACmdTag SAgent END_
|
||||
"CONNECT" -> pure $ ACmdTag SAgent CONNECT_
|
||||
"DISCONNECT" -> pure $ ACmdTag SAgent DISCONNECT_
|
||||
"DOWN" -> pure $ ACmdTag SAgent DOWN_
|
||||
"UP" -> pure $ ACmdTag SAgent UP_
|
||||
"SEND" -> pure $ ACmdTag SClient SEND_
|
||||
"MID" -> pure $ ACmdTag SAgent MID_
|
||||
"SENT" -> pure $ ACmdTag SAgent SENT_
|
||||
"MERR" -> pure $ ACmdTag SAgent MERR_
|
||||
"MSG" -> pure $ ACmdTag SAgent MSG_
|
||||
"ACK" -> pure $ ACmdTag SClient ACK_
|
||||
"OFF" -> pure $ ACmdTag SClient OFF_
|
||||
"DEL" -> pure $ ACmdTag SClient DEL_
|
||||
"CHK" -> pure $ ACmdTag SClient CHK_
|
||||
"STAT" -> pure $ ACmdTag SAgent STAT_
|
||||
"OK" -> pure $ ACmdTag SAgent OK_
|
||||
"ERR" -> pure $ ACmdTag SAgent ERR_
|
||||
"SUSPENDED" -> pure $ ACmdTag SAgent SUSPENDED_
|
||||
_ -> fail "bad ACmdTag"
|
||||
|
||||
instance APartyI p => Encoding (ACommandTag p) where
|
||||
smpEncode = \case
|
||||
NEW_ -> "NEW"
|
||||
INV_ -> "INV"
|
||||
JOIN_ -> "JOIN"
|
||||
CONF_ -> "CONF"
|
||||
LET_ -> "LET"
|
||||
REQ_ -> "REQ"
|
||||
ACPT_ -> "ACPT"
|
||||
RJCT_ -> "RJCT"
|
||||
INFO_ -> "INFO"
|
||||
CON_ -> "CON"
|
||||
SUB_ -> "SUB"
|
||||
END_ -> "END"
|
||||
CONNECT_ -> "CONNECT"
|
||||
DISCONNECT_ -> "DISCONNECT"
|
||||
DOWN_ -> "DOWN"
|
||||
UP_ -> "UP"
|
||||
SEND_ -> "SEND"
|
||||
MID_ -> "MID"
|
||||
SENT_ -> "SENT"
|
||||
MERR_ -> "MERR"
|
||||
MSG_ -> "MSG"
|
||||
ACK_ -> "ACK"
|
||||
OFF_ -> "OFF"
|
||||
DEL_ -> "DEL"
|
||||
CHK_ -> "CHK"
|
||||
STAT_ -> "STAT"
|
||||
OK_ -> "OK"
|
||||
ERR_ -> "ERR"
|
||||
SUSPENDED_ -> "SUSPENDED"
|
||||
smpP = (\(ACmdTag _ t) -> checkParty t) <$?> smpP
|
||||
|
||||
checkParty :: forall t p p'. (APartyI p, APartyI p') => t p' -> Either String (t p)
|
||||
checkParty x = case testEquality (sAParty @p) (sAParty @p') of
|
||||
Just Refl -> Right x
|
||||
Nothing -> Left "bad party"
|
||||
|
||||
-- | SMP agent command and response parser
|
||||
commandP :: Parser ByteString -> Parser ACmd
|
||||
commandP parseByteString =
|
||||
"NEW " *> newCmd
|
||||
<|> "INV " *> invResp
|
||||
<|> "JOIN " *> joinCmd
|
||||
<|> "CONF " *> confMsg
|
||||
<|> "LET " *> letCmd
|
||||
<|> "REQ " *> reqMsg
|
||||
<|> "ACPT " *> acptCmd
|
||||
<|> "RJCT " *> rjctCmd
|
||||
<|> "INFO " *> infoCmd
|
||||
<|> "SUB" $> ACmd SClient SUB
|
||||
<|> "END" $> ACmd SAgent END
|
||||
<|> "CONNECT " *> connectResp
|
||||
<|> "DISCONNECT " *> disconnectResp
|
||||
<|> "DOWN " *> downResp
|
||||
<|> "UP " *> upResp
|
||||
<|> "SEND " *> sendCmd
|
||||
<|> "MID " *> msgIdResp
|
||||
<|> "SENT " *> sentResp
|
||||
<|> "MERR " *> msgErrResp
|
||||
<|> "MSG " *> message
|
||||
<|> "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
|
||||
commandP binaryP =
|
||||
smpP
|
||||
>>= \case
|
||||
ACmdTag SClient cmd ->
|
||||
ACmd SClient <$> case cmd of
|
||||
NEW_ -> s (NEW <$> strP_ <*> strP)
|
||||
JOIN_ -> s (JOIN <$> strP_ <*> strP_ <*> binaryP)
|
||||
LET_ -> s (LET <$> A.takeTill (== ' ') <* A.space <*> binaryP)
|
||||
ACPT_ -> s (ACPT <$> A.takeTill (== ' ') <* A.space <*> binaryP)
|
||||
RJCT_ -> s (RJCT <$> A.takeByteString)
|
||||
SUB_ -> pure SUB
|
||||
SEND_ -> s (SEND <$> smpP <* A.space <*> binaryP)
|
||||
ACK_ -> s (ACK <$> A.decimal)
|
||||
OFF_ -> pure OFF
|
||||
DEL_ -> pure DEL
|
||||
CHK_ -> pure CHK
|
||||
ACmdTag SAgent cmd ->
|
||||
ACmd SAgent <$> case cmd of
|
||||
INV_ -> s (INV <$> strP)
|
||||
CONF_ -> s (CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> binaryP)
|
||||
REQ_ -> s (REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> binaryP)
|
||||
INFO_ -> s (INFO <$> binaryP)
|
||||
CON_ -> pure CON
|
||||
END_ -> pure END
|
||||
CONNECT_ -> s (CONNECT <$> strP_ <*> strP)
|
||||
DISCONNECT_ -> s (DISCONNECT <$> strP_ <*> strP)
|
||||
DOWN_ -> s (DOWN <$> strP_ <*> connections)
|
||||
UP_ -> s (UP <$> strP_ <*> connections)
|
||||
MID_ -> s (MID <$> A.decimal)
|
||||
SENT_ -> s (SENT <$> A.decimal)
|
||||
MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
|
||||
MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP)
|
||||
STAT_ -> s (STAT <$> strP)
|
||||
OK_ -> pure OK
|
||||
ERR_ -> s (ERR <$> strP)
|
||||
SUSPENDED_ -> pure SUSPENDED
|
||||
where
|
||||
newCmd = ACmd SClient .: NEW <$> strP_ <*> strP
|
||||
invResp = ACmd SAgent . INV <$> strP
|
||||
joinCmd = ACmd SClient .:. JOIN <$> strP_ <*> strP_ <*> parseByteString
|
||||
confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> parseByteString
|
||||
letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> parseByteString
|
||||
reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> parseByteString
|
||||
acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> parseByteString
|
||||
rjctCmd = ACmd SClient . RJCT <$> A.takeByteString
|
||||
infoCmd = ACmd SAgent . INFO <$> parseByteString
|
||||
connectResp = ACmd SAgent .: CONNECT <$> strP_ <*> strP
|
||||
disconnectResp = ACmd SAgent .: DISCONNECT <$> strP_ <*> strP
|
||||
downResp = ACmd SAgent .: DOWN <$> strP_ <*> connections
|
||||
upResp = ACmd SAgent .: UP <$> strP_ <*> connections
|
||||
sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> parseByteString
|
||||
msgIdResp = ACmd SAgent . MID <$> A.decimal
|
||||
sentResp = ACmd SAgent . SENT <$> A.decimal
|
||||
msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP
|
||||
message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> parseByteString
|
||||
ackCmd = ACmd SClient . ACK <$> A.decimal
|
||||
statResp = ACmd SAgent . STAT <$> strP
|
||||
s :: Parser a -> Parser a
|
||||
s p = A.space *> p
|
||||
connections :: Parser [ConnId]
|
||||
connections = strP `A.sepBy'` A.char ','
|
||||
msgMetaP = do
|
||||
integrity <- strP
|
||||
@@ -990,7 +1129,6 @@ commandP parseByteString =
|
||||
sndMsgId <- " S=" *> A.decimal
|
||||
pure MsgMeta {integrity, recipient, broker, sndMsgId}
|
||||
partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
|
||||
agentError = ACmd SAgent . ERR <$> strP
|
||||
|
||||
parseCommand :: ByteString -> Either AgentErrorType ACmd
|
||||
parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX
|
||||
|
||||
@@ -695,19 +695,18 @@ updateRatchet db connId rc skipped = do
|
||||
forM_ (M.assocs mks) $ \(msgN, mk) ->
|
||||
DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk)
|
||||
|
||||
createCommand :: DB.Connection -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId
|
||||
createCommand db connId (Just (SMPServer host port _)) command = do
|
||||
createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId
|
||||
createCommand db corrId connId srv cmd = do
|
||||
DB.execute
|
||||
db
|
||||
"INSERT INTO commands (host, port, conn_id, command) VALUES (?, ?, ?, ?)"
|
||||
(host, port, connId, serializeCommand command)
|
||||
insertedRowId db
|
||||
createCommand db connId Nothing command = do
|
||||
DB.execute
|
||||
db
|
||||
"INSERT INTO commands (conn_id, command) VALUES (?, ?)"
|
||||
(connId, command)
|
||||
"INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)"
|
||||
(host_, port_, corrId, connId, aCommandTag cmd, cmd)
|
||||
insertedRowId db
|
||||
where
|
||||
(host_, port_) =
|
||||
case srv of
|
||||
Just (SMPServer host port _) -> (Just host, Just port)
|
||||
_ -> (Nothing, Nothing)
|
||||
|
||||
insertedRowId :: DB.Connection -> IO Int64
|
||||
insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
|
||||
@@ -728,16 +727,13 @@ getPendingCommands db connId = do
|
||||
where
|
||||
srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId)
|
||||
|
||||
getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ConnId, ACmd))
|
||||
getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, ACmd))
|
||||
getPendingCommand db msgId = do
|
||||
firstRow pendingCmd SECmdNotFound $
|
||||
firstRow id SECmdNotFound $
|
||||
DB.query
|
||||
db
|
||||
"SELECT conn_id, command FROM commands WHERE command_id = ?"
|
||||
"SELECT corr_id, conn_id, command FROM commands WHERE command_id = ?"
|
||||
(Only msgId)
|
||||
where
|
||||
pendingCmd :: (ConnId, ACmd) -> (ConnId, ACmd)
|
||||
pendingCmd (connId, commandStr) = (connId, commandStr)
|
||||
|
||||
deleteCommand :: DB.Connection -> AsyncCmdId -> IO ()
|
||||
deleteCommand db cmdId =
|
||||
@@ -1109,6 +1105,10 @@ instance ToField (ACommand p) where toField = toField . serializeCommand
|
||||
|
||||
instance FromField ACmd where fromField = blobFieldParser dbCommandP
|
||||
|
||||
instance APartyI p => ToField (ACommandTag p) where toField = toField . smpEncode
|
||||
|
||||
instance FromField ACmdTag where fromField = blobFieldParser smpP
|
||||
|
||||
listToEither :: e -> [a] -> Either e a
|
||||
listToEither _ (x : _) = Right x
|
||||
listToEither e _ = Left e
|
||||
|
||||
@@ -13,8 +13,10 @@ CREATE TABLE commands (
|
||||
conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
|
||||
host TEXT,
|
||||
port TEXT,
|
||||
command TEXT NOT NULL,
|
||||
command_version INTEGER NOT NULL DEFAULT 1,
|
||||
corr_id BLOB NOT NULL,
|
||||
command_tag BLOB NOT NULL,
|
||||
command BLOB NOT NULL,
|
||||
agent_version INTEGER NOT NULL DEFAULT 1,
|
||||
FOREIGN KEY (host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
);
|
||||
|
||||
@@ -199,8 +199,10 @@ CREATE TABLE commands(
|
||||
conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
|
||||
host TEXT,
|
||||
port TEXT,
|
||||
command TEXT NOT NULL,
|
||||
command_version INTEGER NOT NULL DEFAULT 1,
|
||||
corr_id BLOB NOT NULL,
|
||||
command_tag BLOB NOT NULL,
|
||||
command BLOB NOT NULL,
|
||||
agent_version INTEGER NOT NULL DEFAULT 1,
|
||||
FOREIGN KEY(host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
);
|
||||
|
||||
@@ -570,11 +570,15 @@ testAsyncCommands = do
|
||||
alice <- getSMPAgentClient agentCfg initAgentServers
|
||||
bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers
|
||||
Right () <- runExceptT $ do
|
||||
bobId <- createConnectionAsync alice True SCMInvitation
|
||||
("", _, INV (ACR _ qInfo)) <- get alice
|
||||
aliceId <- joinConnectionAsync bob True qInfo "bob's connInfo"
|
||||
bobId <- createConnectionAsync alice "1" True SCMInvitation
|
||||
("1", bobId', INV (ACR _ qInfo)) <- get alice
|
||||
liftIO $ bobId' `shouldBe` bobId
|
||||
aliceId <- joinConnectionAsync bob "2" True qInfo "bob's connInfo"
|
||||
("2", aliceId', OK) <- get bob
|
||||
liftIO $ aliceId' `shouldBe` aliceId
|
||||
("", _, CONF confId _ "bob's connInfo") <- get alice
|
||||
allowConnectionAsync alice bobId confId "alice's connInfo"
|
||||
allowConnectionAsync alice "3" bobId confId "alice's connInfo"
|
||||
("3", _, OK) <- get alice
|
||||
get alice ##> ("", bobId, CON)
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
@@ -584,17 +588,22 @@ testAsyncCommands = do
|
||||
2 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?"
|
||||
get alice ##> ("", bobId, SENT $ baseId + 2)
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
ackMessageAsync bob aliceId $ baseId + 1
|
||||
ackMessageAsync bob "4" aliceId $ baseId + 1
|
||||
("4", _, OK) <- get bob
|
||||
get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
|
||||
ackMessageAsync bob aliceId $ baseId + 2
|
||||
ackMessageAsync bob "5" aliceId $ baseId + 2
|
||||
("5", _, OK) <- get bob
|
||||
3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "hello too"
|
||||
get bob ##> ("", aliceId, SENT $ baseId + 3)
|
||||
4 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 1"
|
||||
get bob ##> ("", aliceId, SENT $ baseId + 4)
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
ackMessageAsync alice bobId $ baseId + 3
|
||||
ackMessageAsync alice "6" bobId $ baseId + 3
|
||||
("6", _, OK) <- get alice
|
||||
get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
|
||||
ackMessageAsync alice bobId $ baseId + 4
|
||||
ackMessageAsync alice "7" bobId $ baseId + 4
|
||||
("7", _, OK) <- get alice
|
||||
pure ()
|
||||
pure ()
|
||||
where
|
||||
baseId = 3
|
||||
@@ -603,14 +612,14 @@ testAsyncCommands = do
|
||||
testAsyncCommandsRestore :: ATransport -> IO ()
|
||||
testAsyncCommandsRestore t = do
|
||||
alice <- getSMPAgentClient agentCfg initAgentServers
|
||||
Right bobId <- runExceptT $ createConnectionAsync alice True SCMInvitation
|
||||
Right bobId <- runExceptT $ createConnectionAsync alice "1" True SCMInvitation
|
||||
liftIO $ noMessages alice "alice doesn't receive INV because server is down"
|
||||
disconnectAgentClient alice
|
||||
alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers
|
||||
withSmpServerStoreLogOn t testPort $ \_ -> do
|
||||
Right () <- runExceptT $ do
|
||||
subscribeConnection alice' bobId
|
||||
("", _, INV _) <- get alice'
|
||||
("1", _, INV _) <- get alice'
|
||||
pure ()
|
||||
pure ()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user