dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-09-01 20:18:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into notifications-server

Browse Source
This commit is contained in:
Evgeny Poberezkin
2022-04-13 08:52:45 +01:00
parent 65c77c78f3 d38303d5f1
commit 327f7112d0
7 changed files with 54 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Agent.hs
+16 -7
View File
@@ -82,7 +82,7 @@ import Simplex.Messaging.Agent.Protocol
import Simplex.Messaging.Agent.RetryInterval
import Simplex.Messaging.Agent.Store
import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
import Simplex.Messaging.Client (ServerTransmission)
import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
import qualified Simplex.Messaging.Crypto as C
import qualified Simplex.Messaging.Crypto.Ratchet as CR
import Simplex.Messaging.Encoding
@@ -512,7 +512,7 @@ deleteConnection' c connId =
delete :: RcvQueue -> m ()
delete rq = do
deleteQueue c rq
removeSubscription c connId
atomically $ removeSubscription c connId
withStore (`deleteConn` connId)
-- | Change servers to be used for creating new queues, in Reader monad
@@ -617,7 +617,7 @@ subscriber c@AgentClient {msgQ} = forever $ do
Right _ -> return ()
processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m ()
processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd) = do
withStore (\st -> getRcvConn st srv rId) >>= \case
SomeConn SCDuplex (DuplexConnection cData rq _) -> processSMP SCDuplex cData rq
SomeConn SCRcv (RcvConnection cData rq) -> processSMP SCRcv cData rq
@@ -658,10 +658,19 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
_ -> prohibited >> ack
_ -> prohibited >> ack
_ -> prohibited >> ack
SMP.END -> do
removeSubscription c connId
logServer "<--" c srv rId "END"
notify END
SMP.END ->
atomically (TM.lookup srv smpClients >>= fmap join . mapM tryReadTMVar >>= processEND)
>>= logServer "<--" c srv rId
where
processEND = \case
Just (Right clnt)
| sessId == sessionId clnt -> do
removeSubscription c connId
writeTBQueue subQ ("", connId, END)
pure "END"
| otherwise -> ignored
_ -> ignored
ignored = pure "END from disconnected client - ignored"
_ -> do
logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
notify . ERR $ BROKER UNEXPECTED
src/Simplex/Messaging/Agent/Client.hs
+2 -2
View File
@@ -399,8 +399,8 @@ addSubs_ ss rq@RcvQueue {server} connId =
Just m -> TM.insert connId rq m
_ -> TM.singleton connId rq >>= \m -> TM.insert server m ss
removeSubscription :: MonadIO m => AgentClient -> ConnId -> m ()
removeSubscription c@AgentClient {subscrConns} connId = atomically $ do
removeSubscription :: AgentClient -> ConnId -> STM ()
removeSubscription c@AgentClient {subscrConns} connId = do
server_ <- TM.lookupDelete connId subscrConns
mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
src/Simplex/Messaging/Client.hs
+24 -37
View File
@@ -24,7 +24,7 @@
-- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md
module Simplex.Messaging.Client
( -- * Connect (disconnect) client to (from) SMP server
ProtocolClient,
ProtocolClient (sessionId),
SMPClient,
getProtocolClient,
closeProtocolClient,
@@ -65,7 +65,7 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Protocol as SMP
import Simplex.Messaging.TMap (TMap)
import qualified Simplex.Messaging.TMap as TM
import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError)
import Simplex.Messaging.Transport
import Simplex.Messaging.Transport.Client (runTransportClient)
import Simplex.Messaging.Transport.KeepAlive
import Simplex.Messaging.Transport.WebSockets (WS)
@@ -78,10 +78,9 @@ import System.Timeout (timeout)
data ProtocolClient msg = ProtocolClient
{ action :: Async (),
connected :: TVar Bool,
sessionId :: ByteString,
sessionId :: SessionId,
protocolServer :: ProtocolServer,
tcpTimeout :: Int,
smpPingFailures :: TVar Int,
clientCorrId :: TVar Natural,
sentCommands :: TMap CorrId (Request msg),
sndQ :: TBQueue SentRawTransmission,
@@ -92,7 +91,7 @@ data ProtocolClient msg = ProtocolClient
type SMPClient = ProtocolClient SMP.BrokerMsg
-- | Type synonym for transmission from some SPM server queue.
type ServerTransmission msg = (ProtocolServer, QueueId, msg)
type ServerTransmission msg = (ProtocolServer, SessionId, QueueId, msg)
-- | protocol client configuration.
data ProtocolClientConfig = ProtocolClientConfig
@@ -105,11 +104,7 @@ data ProtocolClientConfig = ProtocolClientConfig
-- | TCP keep-alive options, Nothing to skip enabling keep-alive
tcpKeepAlive :: Maybe KeepAliveOpts,
-- | period for SMP ping commands (microseconds)
smpPing :: Int,
-- | timeout for SMP pings (microseconds)
smpPingTimeout :: Int,
-- | failed pings count
smpPingFailLimit :: Int
smpPing :: Int
}
-- | Default protocol client configuration.
@@ -120,9 +115,7 @@ defaultClientConfig =
defaultTransport = ("443", transport @TLS),
tcpTimeout = 5_000_000,
tcpKeepAlive = Just defaultKeepAliveOpts,
smpPing = 300_000_000, -- 5 min,
smpPingTimeout = 10_000_000,
smpPingFailLimit = 3
smpPing = 600_000_000 -- 10min
}
data Request msg = Request
@@ -138,14 +131,13 @@ type Response msg = Either ProtocolClientError msg
-- A single queue can be used for multiple 'SMPClient' instances,
-- as 'SMPServerTransmission' includes server information.
getProtocolClient :: forall msg. Protocol msg => ProtocolServer -> ProtocolClientConfig -> Maybe (TBQueue (ServerTransmission msg)) -> IO () -> IO (Either ProtocolClientError (ProtocolClient msg))
getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, tcpTimeout, smpPingTimeout, tcpKeepAlive, smpPing, smpPingFailLimit} msgQ disconnected =
getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, tcpTimeout, tcpKeepAlive, smpPing} msgQ disconnected =
(atomically mkProtocolClient >>= runClient useTransport)
`catch` \(e :: IOException) -> pure . Left $ PCEIOError e
where
mkProtocolClient :: STM (ProtocolClient msg)
mkProtocolClient = do
connected <- newTVar False
smpPingFailures <- newTVar smpPingFailLimit
clientCorrId <- newTVar 0
sentCommands <- TM.empty
sndQ <- newTBQueue qSize
@@ -157,7 +149,6 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, tcpTimeout, sm
connected,
protocolServer,
tcpTimeout,
smpPingFailures,
clientCorrId,
sentCommands,
sndQ,
@@ -204,16 +195,12 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, tcpTimeout, sm
receive ProtocolClient {rcvQ} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
ping :: ProtocolClient msg -> IO ()
ping c@ProtocolClient {smpPingFailures} = forever $ do
ping c = forever $ do
threadDelay smpPing
runExceptT (sendProtocolCommand c Nothing "" protocolPing $ Just smpPingTimeout) >>= \case
Right _ -> atomically $ writeTVar smpPingFailures smpPingFailLimit
Left e -> do
n <- atomically $ stateTVar smpPingFailures $ \n -> (n - 1, n - 1)
when (n == 0) $ throwIO e
runExceptT $ sendProtocolCommand c Nothing "" protocolPing
process :: ProtocolClient msg -> IO ()
process ProtocolClient {rcvQ, sentCommands} = forever $ do
process ProtocolClient {sessionId, rcvQ, sentCommands} = forever $ do
(_, _, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ
if B.null $ bs corrId
then sendMsg qId respOrErr
@@ -230,12 +217,12 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, tcpTimeout, sm
Just e -> Left $ PCEProtocolError e
_ -> Right r
else Left PCEUnexpectedResponse
sendMsg :: QueueId -> Either ErrorType msg -> IO ()
sendMsg qId = \case
Right cmd -> atomically $ mapM_ (`writeTBQueue` (protocolServer, qId, cmd)) msgQ
-- TODO send everything else to errQ and log in agent
_ -> return ()
where
sendMsg :: QueueId -> Either ErrorType msg -> IO ()
sendMsg qId = \case
Right cmd -> atomically $ mapM_ (`writeTBQueue` (protocolServer, sessionId, qId, cmd)) msgQ
-- TODO send everything else to errQ and log in agent
_ -> return ()
-- | Disconnects client from the server and terminates client threads.
closeProtocolClient :: ProtocolClient msg -> IO ()
@@ -287,11 +274,11 @@ createSMPQueue c rpKey rKey dhKey =
--
-- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue
subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT ProtocolClientError IO ()
subscribeSMPQueue c@ProtocolClient {protocolServer, msgQ} rpKey rId =
subscribeSMPQueue c@ProtocolClient {protocolServer, sessionId, msgQ} rpKey rId =
sendSMPCommand c (Just rpKey) rId SUB >>= \case
OK -> return ()
cmd@MSG {} ->
lift . atomically $ mapM_ (`writeTBQueue` (protocolServer, rId, cmd)) msgQ
lift . atomically $ mapM_ (`writeTBQueue` (protocolServer, sessionId, rId, cmd)) msgQ
_ -> throwE PCEUnexpectedResponse
-- | Subscribe to the SMP queue notifications.
@@ -328,11 +315,11 @@ sendSMPMessage c spKey sId msg =
--
-- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery
ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT ProtocolClientError IO ()
ackSMPMessage c@ProtocolClient {protocolServer, msgQ} rpKey rId =
ackSMPMessage c@ProtocolClient {protocolServer, sessionId, msgQ} rpKey rId =
sendSMPCommand c (Just rpKey) rId ACK >>= \case
OK -> return ()
cmd@MSG {} ->
lift . atomically $ mapM_ (`writeTBQueue` (protocolServer, rId, cmd)) msgQ
lift . atomically $ mapM_ (`writeTBQueue` (protocolServer, sessionId, rId, cmd)) msgQ
_ -> throwE PCEUnexpectedResponse
-- | Irreversibly suspend SMP queue.
@@ -356,11 +343,11 @@ okSMPCommand cmd c pKey qId =
-- | Send SMP command
sendSMPCommand :: PartyI p => SMPClient -> Maybe C.APrivateSignKey -> QueueId -> Command p -> ExceptT ProtocolClientError IO BrokerMsg
sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd) Nothing
sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
-- | Send Protocol command
sendProtocolCommand :: forall msg. ProtocolEncoding (ProtocolCommand msg) => ProtocolClient msg -> Maybe C.APrivateSignKey -> QueueId -> ProtocolCommand msg -> Maybe Int -> ExceptT ProtocolClientError IO msg
sendProtocolCommand ProtocolClient {sndQ, sentCommands, clientCorrId, sessionId, tcpTimeout} pKey qId cmd cmdTimeout_ = do
sendProtocolCommand :: forall msg. ProtocolEncoding (ProtocolCommand msg) => ProtocolClient msg -> Maybe C.APrivateSignKey -> QueueId -> ProtocolCommand msg -> ExceptT ProtocolClientError IO msg
sendProtocolCommand ProtocolClient {sndQ, sentCommands, clientCorrId, sessionId, tcpTimeout} pKey qId cmd = do
corrId <- lift_ getNextCorrId
t <- signTransmission $ encodeTransmission sessionId (corrId, qId, cmd)
ExceptT $ sendRecv corrId t
@@ -384,7 +371,7 @@ sendProtocolCommand ProtocolClient {sndQ, sentCommands, clientCorrId, sessionId,
sendRecv :: CorrId -> SentRawTransmission -> IO (Response msg)
sendRecv corrId t = atomically (send corrId t) >>= withTimeout . atomically . takeTMVar
where
withTimeout a = fromMaybe (Left PCEResponseTimeout) <$> timeout (fromMaybe tcpTimeout cmdTimeout_) a
withTimeout a = fromMaybe (Left PCEResponseTimeout) <$> timeout tcpTimeout a
send :: CorrId -> SentRawTransmission -> STM (TMVar (Response msg))
send corrId t = do
src/Simplex/Messaging/Notifications/Client.hs
+1 -1
View File
@@ -56,7 +56,7 @@ ntfDeleteSubscription = okNtfCommand SDEL
-- | Send notification server command
sendNtfCommand :: NtfEntityI e => NtfClient -> Maybe C.APrivateSignKey -> NtfEntityId -> NtfCommand e -> ExceptT ProtocolClientError IO NtfResponse
sendNtfCommand c pKey entId cmd = sendProtocolCommand c pKey entId (NtfCmd sNtfEntity cmd) Nothing
sendNtfCommand c pKey entId cmd = sendProtocolCommand c pKey entId (NtfCmd sNtfEntity cmd)
okNtfCommand :: NtfEntityI e => NtfCommand e -> NtfClient -> C.APrivateSignKey -> NtfEntityId -> ExceptT ProtocolClientError IO ()
okNtfCommand cmd c pKey entId =
src/Simplex/Messaging/Notifications/Server.hs
+1 -1
View File
@@ -75,7 +75,7 @@ ntfSubscriber NtfSubscriber {subQ, smpAgent = ca@SMPClientAgent {msgQ, agentQ}}
receiveSMP :: m ()
receiveSMP = forever $ do
(srv, ntfId, msg) <- atomically $ readTBQueue msgQ
(srv, sessId, ntfId, msg) <- atomically $ readTBQueue msgQ
case msg of
SMP.NMSG -> do
-- check when the last NMSG was received from this queue
src/Simplex/Messaging/Protocol.hs
+3 -3
View File
@@ -116,7 +116,7 @@ import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding
import Simplex.Messaging.Encoding.String
import Simplex.Messaging.Parsers
import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), smpClientHandshake, tGetBlock, tPutBlock)
import Simplex.Messaging.Transport
import Simplex.Messaging.Util (bshow, (<$?>))
import Simplex.Messaging.Version
import Test.QuickCheck (Arbitrary (..))
@@ -180,7 +180,7 @@ type Signed = ByteString
data RawTransmission = RawTransmission
{ signature :: ByteString,
signed :: ByteString,
sessId :: ByteString,
sessId :: SessionId,
corrId :: ByteString,
entityId :: ByteString,
command :: ByteString
@@ -188,7 +188,7 @@ data RawTransmission = RawTransmission
deriving (Show)
-- | unparsed sent SMP transmission with signature, without session ID.
type SignedRawTransmission = (Maybe C.ASignature, ByteString, ByteString, ByteString)
type SignedRawTransmission = (Maybe C.ASignature, SessionId, ByteString, ByteString)
-- | unparsed sent SMP transmission with signature.
type SentRawTransmission = (Maybe C.ASignature, ByteString)
src/Simplex/Messaging/Transport.hs
+7 -3
View File
@@ -37,6 +37,7 @@ module Simplex.Messaging.Transport
-- * TLS Transport
TLS (..),
SessionId,
connectTLS,
closeTLS,
supportedParameters,
@@ -114,7 +115,7 @@ class Transport c where
getClientConnection :: T.Context -> IO c
-- | tls-unique channel binding per RFC5929
tlsUnique :: c -> ByteString
tlsUnique :: c -> SessionId
-- | Close connection
closeConnection :: c -> IO ()
@@ -249,15 +250,18 @@ trimCR s = if B.last s == '\r' then B.init s else s
-- | The handle for SMP encrypted transport connection over Transport .
data THandle c = THandle
{ connection :: c,
sessionId :: ByteString,
sessionId :: SessionId,
blockSize :: Int,
-- | agreed server protocol version
thVersion :: Version
}
-- | TLS-unique channel binding
type SessionId = ByteString
data ServerHandshake = ServerHandshake
{ smpVersionRange :: VersionRange,
sessionId :: ByteString
sessionId :: SessionId
}
data ClientHandshake = ClientHandshake