From e3e05d474d6b1c7461057020cd0c8f67080129fd Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sat, 17 Sep 2022 14:08:49 +0100 Subject: [PATCH] wrap agent command for async command processing, to allow extension with internal commands (#528) * wrap agent command for async command processing, to allow extension with internal commands * change encoding for constructor --- .gitignore | 3 ++- src/Simplex/Messaging/Agent.hs | 6 ++--- src/Simplex/Messaging/Agent/Protocol.hs | 17 ++++++------ src/Simplex/Messaging/Agent/Store.hs | 29 +++++++++++++++++++++ src/Simplex/Messaging/Agent/Store/SQLite.hs | 16 ++++++------ 5 files changed, 51 insertions(+), 20 deletions(-) diff --git a/.gitignore b/.gitignore index 5d4ac9f5d..3e7898442 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,5 @@ tests/tmp dist-newstyle/ -src/tor +cabal.project.local +cabal.project.local~ diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index f002f36a0..5c055b2e2 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -641,7 +641,7 @@ sendMessage' c connId msgFlags msg = 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 -> createCommand db corrId connId server aCommand + commandId <- withStore' c $ \db -> createCommand db corrId connId server $ AClientCommand aCommand queuePendingCommands c server [commandId] resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m () @@ -689,9 +689,9 @@ runCommandProcessing c@AgentClient {subQ} server = do atomically $ beginAgentOperation c AOSndNetwork E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e) - Right (corrId, connId, ACmd _ cmd) -> processCmd ri corrId connId cmdId cmd + Right (corrId, connId, AClientCommand cmd) -> processCmd ri corrId connId cmdId cmd where - processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand p -> m () + processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand 'Client -> m () processCmd ri corrId connId cmdId = \case NEW enableNtfs (ACM cMode) -> do usedSrvs <- newTVarIO ([] :: [SMPServer]) diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 5b0370d0b..5a13cc42e 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -105,6 +105,7 @@ module Simplex.Messaging.Agent.Protocol agentMessageType, extraSMPServerHosts, updateSMPServerHosts, + checkParty, -- * TCP transport functions tPut, @@ -222,7 +223,7 @@ instance APartyI Agent where sAParty = SAgent instance APartyI Client where sAParty = SClient -data ACmd = forall p. ACmd (SAParty p) (ACommand p) +data ACmd = forall p. APartyI p => ACmd (SAParty p) (ACommand p) deriving instance Show ACmd @@ -1008,9 +1009,9 @@ networkCommandP = commandP A.takeByteString dbCommandP :: Parser ACmd dbCommandP = commandP $ A.take =<< (A.decimal <* "\n") -instance Encoding ACmdTag where - smpEncode (ACmdTag _ cmd) = smpEncode cmd - smpP = +instance StrEncoding ACmdTag where + strEncode (ACmdTag _ cmd) = strEncode cmd + strP = A.takeTill (== ' ') >>= \case "NEW" -> pure $ ACmdTag SClient NEW_ "INV" -> pure $ ACmdTag SAgent INV_ @@ -1043,8 +1044,8 @@ instance Encoding ACmdTag where "SUSPENDED" -> pure $ ACmdTag SAgent SUSPENDED_ _ -> fail "bad ACmdTag" -instance APartyI p => Encoding (ACommandTag p) where - smpEncode = \case +instance APartyI p => StrEncoding (ACommandTag p) where + strEncode = \case NEW_ -> "NEW" INV_ -> "INV" JOIN_ -> "JOIN" @@ -1074,7 +1075,7 @@ instance APartyI p => Encoding (ACommandTag p) where OK_ -> "OK" ERR_ -> "ERR" SUSPENDED_ -> "SUSPENDED" - smpP = (\(ACmdTag _ t) -> checkParty t) <$?> smpP + strP = (\(ACmdTag _ t) -> checkParty t) <$?> strP checkParty :: forall t p p'. (APartyI p, APartyI p') => t p' -> Either String (t p) checkParty x = case testEquality (sAParty @p) (sAParty @p') of @@ -1084,7 +1085,7 @@ checkParty x = case testEquality (sAParty @p) (sAParty @p') of -- | SMP agent command and response parser commandP :: Parser ByteString -> Parser ACmd commandP binaryP = - smpP + strP >>= \case ACmdTag SClient cmd -> ACmd SClient <$> case cmd of diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 7bc8a6b54..fb13fac06 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -3,13 +3,16 @@ {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE KindSignatures #-} +{-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE StandaloneDeriving #-} {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-} module Simplex.Messaging.Agent.Store where import Control.Exception (Exception) +import qualified Data.Attoparsec.ByteString.Char8 as A import Data.ByteString.Char8 (ByteString) import Data.Int (Int64) import Data.Kind (Type) @@ -18,6 +21,7 @@ import Data.Type.Equality import Simplex.Messaging.Agent.Protocol import qualified Simplex.Messaging.Crypto as C import Simplex.Messaging.Crypto.Ratchet (RatchetX448) +import Simplex.Messaging.Encoding.String import Simplex.Messaging.Protocol ( MsgBody, MsgFlags, @@ -31,6 +35,7 @@ import Simplex.Messaging.Protocol SndPrivateSignKey, ) import qualified Simplex.Messaging.Protocol as SMP +import Simplex.Messaging.Util ((<$?>)) import Simplex.Messaging.Version -- * Queue types @@ -159,6 +164,30 @@ data ConnData = ConnData } deriving (Eq, Show) +data AgentCommand = AClientCommand (ACommand 'Client) + +instance StrEncoding AgentCommand where + strEncode = \case + AClientCommand cmd -> "CLIENT " <> serializeCommand cmd + strP = + A.takeTill (== ' ') >>= \case + "CLIENT" -> AClientCommand <$> (A.space *> ((\(ACmd _ cmd) -> checkParty cmd) <$?> dbCommandP)) + _ -> fail "bad AgentCommand" + +data AgentCommandTag = AClientCommandTag (ACommandTag 'Client) + +instance StrEncoding AgentCommandTag where + strEncode = \case + AClientCommandTag t -> "CLIENT " <> strEncode t + strP = + A.takeTill (== ' ') >>= \case + "CLIENT" -> AClientCommandTag <$> (A.space *> strP) + _ -> fail "bad AgentCommandTag" + +agentCommandTag :: AgentCommand -> AgentCommandTag +agentCommandTag = \case + AClientCommand cmd -> AClientCommandTag $ aCommandTag cmd + -- * Confirmation types data NewConfirmation = NewConfirmation diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 3def1d5a2..a2ba013e2 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -118,7 +118,7 @@ import Data.Char (toLower) import Data.Function (on) import Data.Functor (($>)) import Data.Int (Int64) -import Data.List (find, foldl', groupBy) +import Data.List (foldl', groupBy) import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.Map.Strict as M import Data.Maybe (fromMaybe, listToMaybe) @@ -702,12 +702,12 @@ 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 -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId +createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> AgentCommand -> IO AsyncCmdId createCommand db corrId connId srv cmd = do DB.execute db "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)" - (host_, port_, corrId, connId, aCommandTag cmd, cmd) + (host_, port_, corrId, connId, agentCommandTag cmd, cmd) insertedRowId db where (host_, port_) = @@ -734,7 +734,7 @@ getPendingCommands db connId = do where srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId) -getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, ACmd)) +getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, AgentCommand)) getPendingCommand db msgId = do firstRow id SECmdNotFound $ DB.query @@ -1108,13 +1108,13 @@ instance ToField (NonEmpty TransportHost) where toField = toField . decodeLatin1 instance FromField (NonEmpty TransportHost) where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8 -instance ToField (ACommand p) where toField = toField . serializeCommand +instance ToField AgentCommand where toField = toField . strEncode -instance FromField ACmd where fromField = blobFieldParser dbCommandP +instance FromField AgentCommand where fromField = blobFieldParser strP -instance APartyI p => ToField (ACommandTag p) where toField = toField . smpEncode +instance ToField AgentCommandTag where toField = toField . strEncode -instance FromField ACmdTag where fromField = blobFieldParser smpP +instance FromField AgentCommandTag where fromField = blobFieldParser strP listToEither :: e -> [a] -> Either e a listToEither _ (x : _) = Right x