diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index b1814b95d..700459cc2 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -432,7 +432,7 @@ newProtocolClient c srv clients connectClient reconnectClient clientVar = tryCon withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop hostEvent :: forall msg. ProtocolTypeI (ProtoType msg) => (AProtocolType -> TransportHost -> ACommand 'Agent) -> ProtocolClient msg -> ACommand 'Agent -hostEvent event client = event (AProtocolType $ protocolTypeI @(ProtoType msg)) $ transportHost client +hostEvent event client = event (AProtocolType $ protocolTypeI @(ProtoType msg)) $ transportHost' client getClientConfig :: AgentMonad m => AgentClient -> (AgentConfig -> ProtocolClientConfig) -> m ProtocolClientConfig getClientConfig AgentClient {useNetworkConfig} cfgSel = do @@ -965,4 +965,4 @@ incClientStatN :: AgentClient -> ProtocolClient msg -> Int -> ByteString -> Byte incClientStatN c pc n cmd res = do atomically $ incStat c n statsKey where - statsKey = AgentStatsKey {host = strEncode $ transportHost pc, clientTs = strEncode $ sessionTs pc, cmd, res} + statsKey = AgentStatsKey {host = strEncode $ transportHost' pc, clientTs = strEncode $ sessionTs pc, cmd, res} diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 981d2cda8..4b45c07bd 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -494,8 +494,6 @@ data StoreError SERatchetNotFound | -- | connection does not have associated x3dh keys SEX3dhKeysNotFound - | -- | Used in `getMsg` that is not implemented/used. TODO remove. - SENotImplemented | -- | Used to wrap agent errors inside store operations to avoid race conditions SEAgentError AgentErrorType deriving (Eq, Show, Exception) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220101_initial.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220101_initial.hs index 93a93a858..07c0499de 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220101_initial.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220101_initial.hs @@ -107,8 +107,8 @@ CREATE TABLE snd_messages ( CREATE TABLE conn_confirmations ( confirmation_id BLOB NOT NULL PRIMARY KEY, conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE, - e2e_snd_pub_key BLOB NOT NULL, -- TODO per-queue key. Split? - sender_key BLOB NOT NULL, -- TODO per-queue key. Split? + e2e_snd_pub_key BLOB NOT NULL, + sender_key BLOB NOT NULL, ratchet_state BLOB NOT NULL, sender_conn_info BLOB NOT NULL, accepted INTEGER NOT NULL, diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 693f8a73f..9c39ffa56 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -114,8 +114,8 @@ CREATE TABLE snd_messages( CREATE TABLE conn_confirmations( confirmation_id BLOB NOT NULL PRIMARY KEY, conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE, - e2e_snd_pub_key BLOB NOT NULL, -- TODO per-queue key. Split? - sender_key BLOB NOT NULL, -- TODO per-queue key. Split? + e2e_snd_pub_key BLOB NOT NULL, + sender_key BLOB NOT NULL, ratchet_state BLOB NOT NULL, sender_conn_info BLOB NOT NULL, accepted INTEGER NOT NULL, diff --git a/src/Simplex/Messaging/Agent/TRcvQueues.hs b/src/Simplex/Messaging/Agent/TRcvQueues.hs index 1cc66690f..bed4138a2 100644 --- a/src/Simplex/Messaging/Agent/TRcvQueues.hs +++ b/src/Simplex/Messaging/Agent/TRcvQueues.hs @@ -23,11 +23,9 @@ clear (TRcvQueues qs) = TM.clear qs deleteConn :: ConnId -> TRcvQueues -> STM () deleteConn cId (TRcvQueues qs) = modifyTVar' qs $ M.filter (\rq -> cId /= connId rq) --- TODO possibly, should be limited to active queues hasConn :: ConnId -> TRcvQueues -> STM Bool hasConn cId (TRcvQueues qs) = any (\rq -> cId == connId rq) <$> readTVar qs --- TODO possibly, should be limited to active queues getConns :: TRcvQueues -> STM (Set ConnId) getConns (TRcvQueues qs) = M.foldr' (S.insert . connId) S.empty <$> readTVar qs diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 7d043ee0a..2918d5d29 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -1,6 +1,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE DeriveAnyClass #-} {-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DisambiguateRecordFields #-} {-# LANGUAGE DuplicateRecordFields #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE GADTs #-} @@ -26,11 +27,12 @@ -- 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 (thVersion, sessionId, sessionTs, transportHost), + ProtocolClient (thVersion, sessionId, sessionTs), SMPClient, getProtocolClient, closeProtocolClient, clientServer, + transportHost', -- * SMP protocol command functions createSMPQueue, @@ -99,11 +101,15 @@ import System.Timeout (timeout) -- -- Use 'getSMPClient' to connect to an SMP server and create a client handle. data ProtocolClient msg = ProtocolClient - { action :: Async (), - connected :: TVar Bool, + { action :: Maybe (Async ()), sessionId :: SessionId, sessionTs :: UTCTime, thVersion :: Version, + client_ :: PClient msg + } + +data PClient msg = PClient + { connected :: TVar Bool, protocolServer :: ProtoServer msg, transportHost :: TransportHost, tcpTimeout :: Int, @@ -217,7 +223,10 @@ chooseTransportHost NetworkConfig {socksProxy, hostMode, requiredHostMode} hosts publicHost = find (not . isOnionHost) hosts clientServer :: ProtocolTypeI (ProtoType msg) => ProtocolClient msg -> String -clientServer = B.unpack . strEncode . protocolServer +clientServer = B.unpack . strEncode . protocolServer . client_ + +transportHost' :: ProtocolClient msg -> TransportHost +transportHost' = transportHost . client_ -- | Connects to 'ProtocolServer' using passed client configuration -- and queue for messages and notifications. @@ -233,7 +242,7 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, Left e -> pure $ Left e where NetworkConfig {tcpConnectTimeout, tcpTimeout, tcpKeepAlive, socksProxy, smpPingInterval} = networkConfig - mkProtocolClient :: TransportHost -> STM (ProtocolClient msg) + mkProtocolClient :: TransportHost -> STM (PClient msg) mkProtocolClient transportHost = do connected <- newTVar False clientCorrId <- newTVar 0 @@ -241,12 +250,8 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, sndQ <- newTBQueue qSize rcvQ <- newTBQueue qSize return - ProtocolClient - { action = undefined, - sessionId = undefined, - sessionTs = undefined, - thVersion = undefined, - connected, + PClient + { connected, protocolServer, transportHost, tcpTimeout, @@ -257,17 +262,16 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, msgQ } - runClient :: (ServiceName, ATransport) -> TransportHost -> ProtocolClient msg -> IO (Either ProtocolClientError (ProtocolClient msg)) + runClient :: (ServiceName, ATransport) -> TransportHost -> PClient msg -> IO (Either ProtocolClientError (ProtocolClient msg)) runClient (port', ATransport t) useHost c = do - thVar <- newEmptyTMVarIO + cVar <- newEmptyTMVarIO action <- async $ - runTransportClient socksProxy useHost port' (Just $ keyHash protocolServer) tcpKeepAlive (client t c thVar) - `finally` atomically (putTMVar thVar $ Left PCENetworkError) - th_ <- tcpConnectTimeout `timeout` atomically (takeTMVar thVar) - sessionTs <- getCurrentTime - pure $ case th_ of - Just (Right THandle {sessionId, thVersion}) -> Right c {action, sessionId, sessionTs, thVersion} + runTransportClient socksProxy useHost port' (Just $ keyHash protocolServer) tcpKeepAlive (client t c cVar) + `finally` atomically (putTMVar cVar $ Left PCENetworkError) + c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar) + pure $ case c_ of + Just (Right c') -> Right c' {action = Just action} Just (Left e) -> Left e Nothing -> Left PCENetworkError @@ -277,25 +281,24 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, "80" -> ("80", transport @WS) p -> (p, transport @TLS) - client :: forall c. Transport c => TProxy c -> ProtocolClient msg -> TMVar (Either ProtocolClientError (THandle c)) -> c -> IO () - client _ c thVar h = + client :: forall c. Transport c => TProxy c -> PClient msg -> TMVar (Either ProtocolClientError (ProtocolClient msg)) -> c -> IO () + client _ c cVar h = runExceptT (protocolClientHandshake @msg h (keyHash protocolServer) smpServerVRange) >>= \case - Left e -> atomically . putTMVar thVar . Left $ PCETransportError e + Left e -> atomically . putTMVar cVar . Left $ PCETransportError e Right th@THandle {sessionId, thVersion} -> do + sessionTs <- getCurrentTime + let c' = ProtocolClient {action = Nothing, client_ = c, sessionId, thVersion, sessionTs} atomically $ do writeTVar (connected c) True - putTMVar thVar $ Right th - sessionTs <- getCurrentTime - let c' = c {sessionId, thVersion, sessionTs} :: ProtocolClient msg - -- TODO remove ping if 0 is passed (or Nothing?) - raceAny_ [send c' th, process c', receive c' th, ping c'] + putTMVar cVar $ Right c' + raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0]) `finally` disconnected c' send :: Transport c => ProtocolClient msg -> THandle c -> IO () - send ProtocolClient {sndQ} h = forever $ atomically (readTBQueue sndQ) >>= tPut h + send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPut h receive :: Transport c => ProtocolClient msg -> THandle c -> IO () - receive ProtocolClient {rcvQ} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ + receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ ping :: ProtocolClient msg -> IO () ping c = forever $ do @@ -303,10 +306,10 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, runExceptT $ sendProtocolCommand c Nothing "" protocolPing process :: ProtocolClient msg -> IO () - process c = forever $ atomically (readTBQueue $ rcvQ c) >>= mapM_ (processMsg c) + process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c) processMsg :: ProtocolClient msg -> SignedTransmission msg -> IO () - processMsg c@ProtocolClient {sentCommands} (_, _, (corrId, qId, respOrErr)) = + processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, qId, respOrErr)) = if B.null $ bs corrId then sendMsg respOrErr else do @@ -326,12 +329,11 @@ getProtocolClient protocolServer cfg@ProtocolClientConfig {qSize, networkConfig, sendMsg :: Either ErrorType msg -> IO () sendMsg = \case Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c qId msg) msgQ - -- TODO send everything else to errQ and log in agent - _ -> return () + Left e -> putStrLn $ "SMP client error: " <> show e -- | Disconnects client from the server and terminates client threads. closeProtocolClient :: ProtocolClient msg -> IO () -closeProtocolClient = uninterruptibleCancel . action +closeProtocolClient = mapM_ uninterruptibleCancel . action -- | SMP client error type. data ProtocolClientError @@ -400,10 +402,10 @@ subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM response . L.zip qs Left e -> pure $ Left e writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO () -writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ c) +writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c) serverTransmission :: ProtocolClient msg -> RecipientId -> msg -> ServerTransmission msg -serverTransmission ProtocolClient {protocolServer, thVersion, sessionId} entityId message = +serverTransmission ProtocolClient {thVersion, sessionId, client_ = PClient {protocolServer}} entityId message = (protocolServer, thVersion, sessionId, entityId, message) -- | Get message from SMP queue. The server returns ERR PROHIBITED if a client uses SUB and GET via the same transport connection for the same queue @@ -507,7 +509,7 @@ sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd) -- | Send multiple commands with batching and collect responses sendProtocolCommands :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either ProtocolClientError msg)) -sendProtocolCommands c@ProtocolClient {sndQ, tcpTimeout} cs = do +sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} cs = do ts <- mapM (runExceptT . mkTransmission c) cs mapM_ (atomically . writeTBQueue sndQ . L.map fst) . L.nonEmpty . rights $ L.toList ts forConcurrently ts $ \case @@ -518,7 +520,7 @@ sendProtocolCommands c@ProtocolClient {sndQ, tcpTimeout} cs = do -- | Send Protocol command sendProtocolCommand :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT ProtocolClientError IO msg -sendProtocolCommand c@ProtocolClient {sndQ, tcpTimeout} pKey qId cmd = do +sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} pKey qId cmd = do (t, r) <- mkTransmission c (pKey, qId, cmd) ExceptT $ sendRecv t r where @@ -529,7 +531,7 @@ sendProtocolCommand c@ProtocolClient {sndQ, tcpTimeout} pKey qId cmd = do withTimeout a = fromMaybe (Left PCEResponseTimeout) <$> timeout tcpTimeout a mkTransmission :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> ClientCommand msg -> ExceptT ProtocolClientError IO (SentRawTransmission, TMVar (Response msg)) -mkTransmission ProtocolClient {clientCorrId, sessionId, thVersion, sentCommands} (pKey, qId, cmd) = do +mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do corrId <- liftIO $ atomically getNextCorrId t <- signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd) r <- liftIO . atomically $ mkRequest corrId diff --git a/src/Simplex/Messaging/Crypto/Ratchet.hs b/src/Simplex/Messaging/Crypto/Ratchet.hs index 5f6e9c08b..9fc619cad 100644 --- a/src/Simplex/Messaging/Crypto/Ratchet.hs +++ b/src/Simplex/Messaging/Crypto/Ratchet.hs @@ -472,7 +472,6 @@ rcDecrypt rc@Ratchet {rcRcv, rcAD = Str rcAD} rcMKSkipped msg' = do decryptMessage :: MessageKey -> EncRatchetMessage -> ExceptT CryptoError IO (Either CryptoError ByteString) decryptMessage (MessageKey mk iv) EncRatchetMessage {emHeader, emBody, emAuthTag} = -- DECRYPT(mk, cipher-text, CONCAT(AD, enc_header)) - -- TODO add associated data tryE $ decryptAEAD mk iv (rcAD <> emHeader) emBody emAuthTag rootKdf :: (AlgorithmI a, DhAlgorithm a) => RatchetKey -> PublicKey a -> PrivateKey a -> (RatchetKey, RatchetKey, Key) diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index f92c97a50..18ff71370 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -283,7 +283,6 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do Left e -> case e of PPConnection _ -> retryDeliver PPRetryLater -> retryDeliver - -- TODO alert PPCryptoError _ -> err e PPResponseError _ _ -> err e PPTokenInvalid -> updateTknStatus NTInvalid >> err e diff --git a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs index c0c39d665..01a41c1ec 100644 --- a/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs +++ b/src/Simplex/Messaging/Notifications/Server/Push/APNS.hs @@ -90,7 +90,7 @@ signedJWTToken pk (JWTToken hdr claims) = do readECPrivateKey :: FilePath -> IO EC.PrivateKey readECPrivateKey f = do - -- TODO this is specific to APNS key + -- this pattern match is specific to APNS key type, it may need to be extended for other push providers [PK.Unprotected (X.PrivKeyEC X.PrivKeyEC_Named {privkeyEC_name, privkeyEC_priv})] <- PK.readKeyFile f pure EC.PrivateKey {private_curve = ECT.getCurveByName privkeyEC_name, private_d = privkeyEC_priv} @@ -357,7 +357,7 @@ apnsPushProviderClient c@APNSPushClient {nonceDrg, apnsCfg} tkn@NtfTknData {toke _ -> err status reason' | status == Just N.gone410 = throwError PPTokenInvalid | status == Just N.serviceUnavailable503 = liftIO (disconnectApnsHTTP2Client c) >> throwError PPRetryLater - -- Just tooManyRequests429 -> TODO TooManyRequests - too many requests for the same token + -- Just tooManyRequests429 -> TooManyRequests - too many requests for the same token | otherwise = err status reason' err :: Maybe Status -> Text -> ExceptT PushProviderError IO () err s r = throwError $ PPResponseError s r diff --git a/src/Simplex/Messaging/Notifications/Server/Store.hs b/src/Simplex/Messaging/Notifications/Server/Store.hs index d9af0fb29..ee413cab1 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store.hs +++ b/src/Simplex/Messaging/Notifications/Server/Store.hs @@ -106,7 +106,7 @@ removeInactiveTokenRegistrations st NtfTknData {ntfTknId = tId, token} = forM_ tIds $ \(regKey, tId') -> do TM.delete regKey tknRegs TM.delete tId' $ tokens st - -- TODO remove token subscriptions as in deleteNtfToken + void $ deleteTokenSubs st tId' pure $ map snd tIds removeTokenRegistration :: NtfStore -> NtfTknData -> STM () @@ -120,37 +120,28 @@ removeTokenRegistration st NtfTknData {ntfTknId = tId, token, tknVerifyKey} = deleteNtfToken :: NtfStore -> NtfTokenId -> STM [SMPQueueNtf] deleteNtfToken st tknId = do - TM.lookupDelete tknId (tokens st) - >>= mapM_ - ( \NtfTknData {token, tknVerifyKey} -> - TM.lookup token regs - >>= mapM_ - ( \tIds -> do - TM.delete (regKey tknVerifyKey) tIds - whenM (TM.null tIds) $ TM.delete token regs - ) - ) - - -- TODO refactor - qs <- - TM.lookupDelete tknId (tokenSubscriptions st) - >>= mapM - ( readTVar - >=> mapM - ( \subId -> do - TM.lookupDelete subId (subscriptions st) - >>= mapM - ( \NtfSubData {smpQueue} -> - TM.delete smpQueue (subscriptionLookup st) $> smpQueue - ) - ) - . S.toList - ) - pure $ maybe [] catMaybes qs + void $ + TM.lookupDelete tknId (tokens st) $>>= \NtfTknData {token, tknVerifyKey} -> + TM.lookup token regs $>>= \tIds -> + TM.delete (regKey tknVerifyKey) tIds + >> whenM (TM.null tIds) (TM.delete token regs) $> Just () + deleteTokenSubs st tknId where regs = tokenRegistrations st regKey = C.toPubKey C.pubKeyBytes +deleteTokenSubs :: NtfStore -> NtfTokenId -> STM [SMPQueueNtf] +deleteTokenSubs st tknId = do + qs <- + TM.lookupDelete tknId (tokenSubscriptions st) + >>= mapM (readTVar >=> mapM deleteSub . S.toList) + pure $ maybe [] catMaybes qs + where + deleteSub subId = do + TM.lookupDelete subId (subscriptions st) + $>>= \NtfSubData {smpQueue} -> + TM.delete smpQueue (subscriptionLookup st) $> Just smpQueue + getNtfSubscription :: NtfStore -> NtfSubscriptionId -> STM (Maybe NtfSubData) getNtfSubscription st subId = TM.lookup subId (subscriptions st) diff --git a/src/Simplex/Messaging/Server/CLI.hs b/src/Simplex/Messaging/Server/CLI.hs index 3e597a84d..462274b47 100644 --- a/src/Simplex/Messaging/Server/CLI.hs +++ b/src/Simplex/Messaging/Server/CLI.hs @@ -111,10 +111,6 @@ createServerX509 cfgPath x509cfg = do \subjectKeyIdentifier = hash\n\ \authorityKeyIdentifier = keyid:always\n\ \basicConstraints = critical,CA:true\n" - -- TODO revise https://www.rfc-editor.org/rfc/rfc5280#section-4.2.1.3, https://www.rfc-editor.org/rfc/rfc3279#section-2.3.5 - -- IP and FQDN can't both be used as server address interchangeably even if IP is added - -- as Subject Alternative Name, unless the following validation hook is disabled: - -- https://hackage.haskell.org/package/x509-validation-1.6.10/docs/src/Data-X509-Validation.html#validateCertificateName createOpensslServerConf = writeFile (c opensslServerConfFile) diff --git a/src/Simplex/Messaging/Transport.hs b/src/Simplex/Messaging/Transport.hs index ad9caa42c..053a85cae 100644 --- a/src/Simplex/Messaging/Transport.hs +++ b/src/Simplex/Messaging/Transport.hs @@ -28,6 +28,7 @@ module Simplex.Messaging.Transport ( -- * SMP transport parameters supportedSMPServerVRange, simplexMQVersion, + smpBlockSize, -- * Transport connection class Transport (..), diff --git a/src/Simplex/Messaging/Transport/HTTP2/Client.hs b/src/Simplex/Messaging/Transport/HTTP2/Client.hs index c3e96e9d9..428eee214 100644 --- a/src/Simplex/Messaging/Transport/HTTP2/Client.hs +++ b/src/Simplex/Messaging/Transport/HTTP2/Client.hs @@ -26,7 +26,7 @@ import UnliftIO.STM import UnliftIO.Timeout data HTTP2Client = HTTP2Client - { action :: Async (), + { action :: Maybe (Async ()), connected :: TVar Bool, host :: HostName, port :: ServiceName, @@ -71,7 +71,7 @@ getHTTP2Client host port config@HTTP2ClientConfig {tcpKeepAlive, connTimeout, ca mkHTTPS2Client = do connected <- newTVar False reqQ <- newTBQueue $ qSize config - pure HTTP2Client {action = undefined, connected, host, port, config, reqQ} + pure HTTP2Client {action = Nothing, connected, host, port, config, reqQ} runClient :: HTTP2Client -> IO (Either HTTP2ClientError HTTP2Client) runClient c = do @@ -84,7 +84,7 @@ getHTTP2Client host port config@HTTP2ClientConfig {tcpKeepAlive, connTimeout, ca `E.finally` atomically (putTMVar cVar $ Left HCNetworkError) conn_ <- connTimeout `timeout` atomically (takeTMVar cVar) pure $ case conn_ of - Just (Right ()) -> Right c {action} + Just (Right ()) -> Right c {action = Just action} Just (Left e) -> Left e Nothing -> Left HCNetworkError1 @@ -111,8 +111,7 @@ getHTTP2Client host port config@HTTP2ClientConfig {tcpKeepAlive, connTimeout, ca -- | Disconnects client from the server and terminates client threads. closeHTTP2Client :: HTTP2Client -> IO () --- TODO disconnect -closeHTTP2Client = uninterruptibleCancel . action +closeHTTP2Client = mapM_ uninterruptibleCancel . action sendRequest :: HTTP2Client -> Request -> IO (Either HTTP2ClientError HTTP2Response) sendRequest HTTP2Client {reqQ, config} req = do diff --git a/src/Simplex/Messaging/Transport/WebSockets.hs b/src/Simplex/Messaging/Transport/WebSockets.hs index 962cae7a9..69e137c7e 100644 --- a/src/Simplex/Messaging/Transport/WebSockets.hs +++ b/src/Simplex/Messaging/Transport/WebSockets.hs @@ -18,6 +18,7 @@ import Simplex.Messaging.Transport TransportError (..), TransportPeer (..), closeTLS, + smpBlockSize, trimCR, withTlsUnique, ) @@ -33,7 +34,7 @@ websocketsOpts :: ConnectionOptions websocketsOpts = defaultConnectionOptions { connectionCompressionOptions = NoCompression, - connectionFramePayloadSizeLimit = SizeLimit $ 16 * 1024, -- TODO move to Protocol + connectionFramePayloadSizeLimit = SizeLimit $ fromIntegral smpBlockSize, connectionMessageDataSizeLimit = SizeLimit 65536 }