diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 12ac35548..399b4e6be 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -7,12 +7,15 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} module Simplex.FileTransfer.Agent - ( -- Receiving files + ( startWorkers, + closeXFTPAgent, + toFSFilePath, + -- Receiving files receiveFile, - addXFTPWorker, deleteRcvFile, -- Sending files sendFileExperimental, @@ -33,6 +36,7 @@ import qualified Data.ByteString.Char8 as B import Data.List (isSuffixOf, partition) import Data.List.NonEmpty (nonEmpty) import qualified Data.List.NonEmpty as L +import qualified Data.Map.Strict as M import Data.Time.Clock (getCurrentTime) import Data.Time.Format (defaultTimeLocale, formatTime) import Simplex.FileTransfer.Client.Main (CLIError, SendOptions (..), cliSendFile) @@ -57,19 +61,39 @@ import UnliftIO.Concurrent import UnliftIO.Directory import qualified UnliftIO.Exception as E -receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId -receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpWorkPath = do +startWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m () +startWorkers c workDir = do + wd <- asks $ xftpWorkDir . xftpAgent + atomically $ writeTVar wd workDir + startFiles + where + startFiles = do + pendingRcvServers <- withStore' c getPendingRcvFilesServers + forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s) + -- start local worker for files pending decryption, + -- no need to make an extra query for the check + -- as the worker will check the store anyway + addXFTPWorker c Nothing + +closeXFTPAgent :: MonadUnliftIO m => XFTPAgent -> m () +closeXFTPAgent XFTPAgent {xftpWorkers} = do + ws <- atomically $ stateTVar xftpWorkers (,M.empty) + mapM_ (uninterruptibleCancel . snd) ws + +receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId +receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) = do g <- asks idsDrg - workPath <- maybe getTemporaryDirectory pure xftpWorkPath + workPath <- getWorkPath ts <- liftIO getCurrentTime let isoTime = formatTime defaultTimeLocale "%Y%m%d_%H%M%S_%6q" ts prefixPath <- uniqueCombine workPath (isoTime <> "_rcv.xftp") createDirectory prefixPath - let tmpPath = prefixPath "xftp.encrypted" - createDirectory tmpPath - let savePath = prefixPath "xftp.decrypted" - createEmptyFile savePath - fId <- withStore c $ \db -> createRcvFile db g userId fd prefixPath tmpPath savePath + let relPrefixPath = takeFileName prefixPath + relTmpPath = relPrefixPath "xftp.encrypted" + relSavePath = relPrefixPath "xftp.decrypted" + createDirectory =<< toFSFilePath relTmpPath + createEmptyFile =<< toFSFilePath relSavePath + fId <- withStore c $ \db -> createRcvFile db g userId fd relPrefixPath relTmpPath relSavePath forM_ chunks downloadChunk pure fId where @@ -78,6 +102,14 @@ receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) xftpWork addXFTPWorker c (Just server) downloadChunk _ = throwError $ INTERNAL "no replicas" +getWorkPath :: AgentMonad m => m FilePath +getWorkPath = do + workDir <- readTVarIO =<< asks (xftpWorkDir . xftpAgent) + maybe getTemporaryDirectory pure workDir + +toFSFilePath :: AgentMonad m => FilePath -> m FilePath +toFSFilePath f = ( f) <$> getWorkPath + createEmptyFile :: AgentMonad m => FilePath -> m () createEmptyFile fPath = do h <- openFile fPath AppendMode @@ -101,7 +133,7 @@ runXFTPWorker :: forall m. AgentMonad m => AgentClient -> XFTPServer -> TMVar () runXFTPWorker c srv doWork = do forever $ do void . atomically $ readTMVar doWork - agentOperationBracket c AORcvNetwork throwWhenInactive runXftpOperation + agentOperationBracket c AORcvNetwork waitUntilActive runXftpOperation where noWorkToDo = void . atomically $ tryTakeTMVar doWork runXftpOperation :: m () @@ -133,12 +165,14 @@ runXFTPWorker c srv doWork = do loop downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m () downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do - chunkPath <- uniqueCombine fileTmpPath $ show chunkNo + fsFileTmpPath <- toFSFilePath fileTmpPath + chunkPath <- uniqueCombine fsFileTmpPath $ show chunkNo let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest) + relChunkPath = fileTmpPath takeFileName chunkPath agentXFTPDownloadChunk c userId replica chunkSpec fileReceived <- withStore c $ \db -> runExceptT $ do -- both actions can be done in a single store method - f <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId chunkPath + f <- ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId relChunkPath let fileReceived = allChunksReceived f when fileReceived $ liftIO $ updateRcvFileStatus db rcvFileId RFSReceived @@ -151,7 +185,7 @@ runXFTPWorker c srv doWork = do workerInternalError :: AgentMonad m => AgentClient -> DBRcvFileId -> RcvFileId -> Maybe FilePath -> String -> m () workerInternalError c rcvFileId rcvFileEntityId tmpPath internalErrStr = do - forM_ tmpPath removePath + forM_ tmpPath (removePath <=< toFSFilePath) withStore' c $ \db -> updateRcvFileError db rcvFileId internalErrStr notifyInternalError c rcvFileEntityId internalErrStr @@ -162,6 +196,7 @@ runXFTPLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m () runXFTPLocalWorker c@AgentClient {subQ} doWork = do forever $ do void . atomically $ readTMVar doWork + -- TODO agentOperationBracket? runXftpOperation where runXftpOperation :: m () @@ -174,16 +209,17 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do noWorkToDo = void . atomically $ tryTakeTMVar doWork decryptFile :: RcvFile -> m () decryptFile RcvFile {rcvFileId, rcvFileEntityId, key, nonce, tmpPath, savePath, chunks} = do + fsSavePath <- toFSFilePath savePath -- TODO test; recreate file if it's in status RFSDecrypting -- when (status == RFSDecrypting) $ - -- whenM (doesFileExist savePath) (removeFile savePath >> createEmptyFile savePath) + -- whenM (doesFileExist fsSavePath) (removeFile fsSavePath >> createEmptyFile fsSavePath) withStore' c $ \db -> updateRcvFileStatus db rcvFileId RFSDecrypting chunkPaths <- getChunkPaths chunks encSize <- liftIO $ foldM (\s path -> (s +) . fromIntegral <$> getFileSize path) 0 chunkPaths - void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure savePath - forM_ tmpPath removePath + void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure fsSavePath + forM_ tmpPath (removePath <=< toFSFilePath) withStore' c (`updateRcvFileComplete` rcvFileId) - notify $ RFDONE savePath + notify $ RFDONE fsSavePath where notify :: forall e. AEntityI e => ACommand 'Agent e -> m () notify cmd = atomically $ writeTBQueue subQ ("", rcvFileEntityId, APC (sAEntity @e) cmd) @@ -191,7 +227,8 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do getChunkPaths [] = pure [] getChunkPaths (RcvFileChunk {chunkTmpPath = Just path} : cs) = do ps <- getChunkPaths cs - pure $ path : ps + fsPath <- toFSFilePath path + pure $ fsPath : ps getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) = throwError $ INTERNAL "no chunk path" @@ -204,8 +241,8 @@ deleteRcvFile c userId rcvFileEntityId = do withStore' c (`deleteRcvFile'` rcvFileId) else withStore' c (`updateRcvFileDeleted` rcvFileId) -sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId -sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipients xftpWorkPath = do +sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId +sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipients = do g <- asks idsDrg sndFileId <- liftIO $ randomId g 12 xftpSrvs <- atomically $ TM.lookup userId xftpServers @@ -217,7 +254,7 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien sendCLI :: SndFileId -> [XFTPServerWithAuth] -> m () sendCLI sndFileId xftpSrvs = do let fileName = takeFileName filePath - workPath <- maybe getTemporaryDirectory pure xftpWorkPath + workPath <- getWorkPath outputDir <- uniqueCombine workPath $ fileName <> ".descr" createDirectory outputDir let tempPath = workPath "snd" @@ -234,6 +271,8 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien } liftCLI $ cliSendFile sendOptions (sndDescr, rcvDescrs) <- readDescrs outputDir fileName + removePath tempPath + removePath outputDir notify sndFileId $ SFDONE sndDescr rcvDescrs liftCLI :: ExceptT CLIError IO () -> m () liftCLI = either (throwError . INTERNAL . show) pure <=< liftIO . runExceptT @@ -250,9 +289,9 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien notify :: forall e. AEntityI e => SndFileId -> ACommand 'Agent e -> m () notify sndFileId cmd = atomically $ writeTBQueue subQ ("", sndFileId, APC (sAEntity @e) cmd) --- _sendFile :: AgentMonad m => AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId -_sendFile :: AgentClient -> UserId -> Int -> FilePath -> FilePath -> m SndFileId -_sendFile _c _userId _numRecipients _xftpPath _filePath = do +-- _sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId +_sendFile :: AgentClient -> UserId -> FilePath -> Int -> m SndFileId +_sendFile _c _userId _filePath _numRecipients = do -- db: create file in status New without chunks -- add local snd worker for encryption -- return file id to client diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 5b6d475b5..5c342664c 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -80,6 +80,7 @@ module Simplex.Messaging.Agent getNtfToken, getNtfTokenData, toggleConnectionNtfs, + xftpStartWorkers, xftpReceiveFile, xftpDeleteRcvFile, xftpSendFile, @@ -116,7 +117,7 @@ import qualified Data.Text as T import Data.Time.Clock import Data.Time.Clock.System (systemToUTCTime) import qualified Database.SQLite.Simple as DB -import Simplex.FileTransfer.Agent (addXFTPWorker, deleteRcvFile, receiveFile, sendFileExperimental) +import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteRcvFile, receiveFile, sendFileExperimental, startWorkers, toFSFilePath) import Simplex.FileTransfer.Description (ValidFileDescription) import Simplex.FileTransfer.Protocol (FileParty (..)) import Simplex.FileTransfer.Util (removePath) @@ -157,22 +158,13 @@ getSMPAgentClient cfg initServers = newSMPAgentEnv cfg >>= runReaderT runAgent runAgent = do c <- getAgentClient initServers void $ raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c] `forkFinally` const (disconnectAgentClient c) - runExceptT (startFiles c) >>= \case - Left e -> liftIO $ print e - Right _ -> pure () pure c - startFiles c = do - pendingRcvServers <- withStore' c getPendingRcvFilesServers - forM_ pendingRcvServers $ \s -> addXFTPWorker c (Just s) - -- start local worker for files pending decryption, - -- no need to make an extra query for the check - -- as the worker will check the store anyway - addXFTPWorker c Nothing disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m () -disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns}} = do +disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do closeAgentClient c - liftIO $ closeNtfSupervisor ns + closeNtfSupervisor ns + closeXFTPAgent xa logConnection c False resumeAgentClient :: MonadIO m => AgentClient -> m () @@ -339,17 +331,20 @@ getNtfTokenData c = withAgentEnv c $ getNtfTokenData' c toggleConnectionNtfs :: AgentErrorMonad m => AgentClient -> ConnId -> Bool -> m () toggleConnectionNtfs c = withAgentEnv c .: toggleConnectionNtfs' c +xftpStartWorkers :: AgentErrorMonad m => AgentClient -> Maybe FilePath -> m () +xftpStartWorkers c = withAgentEnv c . startWorkers c + -- | Receive XFTP file -xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe FilePath -> m RcvFileId -xftpReceiveFile c = withAgentEnv c .:. receiveFile c +xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId +xftpReceiveFile c = withAgentEnv c .: receiveFile c -- | Delete XFTP rcv file (deletes work files from file system and db records) xftpDeleteRcvFile :: AgentErrorMonad m => AgentClient -> UserId -> RcvFileId -> m () xftpDeleteRcvFile c = withAgentEnv c .: deleteRcvFile c -- | Send XFTP file -xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> Maybe FilePath -> m SndFileId -xftpSendFile c = withAgentEnv c .:: sendFileExperimental c +xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId +xftpSendFile c = withAgentEnv c .:. sendFileExperimental c -- | Activate operations activateAgent :: MonadUnliftIO m => AgentClient -> m () @@ -1625,12 +1620,12 @@ cleanupManager c = do -- cleanup rcv files marked for deletion rcvDeleted <- withStore' c getCleanupRcvFilesDeleted forM_ rcvDeleted $ \(fId, p) -> do - removePath p + removePath =<< toFSFilePath p withStore' c (`deleteRcvFile'` fId) -- cleanup rcv tmp paths rcvTmpPaths <- withStore' c getCleanupRcvFilesTmpPaths forM_ rcvTmpPaths $ \(fId, p) -> do - removePath p + removePath =<< toFSFilePath p withStore' c (`updateRcvFileNoTmpPath` fId) processSMPTransmission :: forall m. AgentMonad m => AgentClient -> ServerTransmission BrokerMsg -> m () diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 27b460cf6..812431510 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -35,14 +35,11 @@ import Control.Monad.Reader import Crypto.Random import Data.List.NonEmpty (NonEmpty) import Data.Map (Map) -import Data.Set (Set) -import qualified Data.Set as S import Data.Time.Clock (NominalDiffTime, nominalDay) import Data.Word (Word16) import Network.Socket import Numeric.Natural import Simplex.FileTransfer.Client (XFTPClientConfig (..), defaultXFTPClientConfig) -import Simplex.FileTransfer.Types (DBSndFileId) import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store.SQLite @@ -220,18 +217,19 @@ newNtfSubSupervisor qSize = do pure NtfSupervisor {ntfTkn, ntfSubQ, ntfWorkers, ntfSMPWorkers} data XFTPAgent = XFTPAgent - { xftpWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()), + { -- if set, XFTP file paths will be considered as relative to this directory + xftpWorkDir :: TVar (Maybe FilePath), + xftpWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()) -- separate send workers for unhindered concurrency between download and upload, -- clients can also be separate by passing direction to withXFTPClient, and differentiating by it - xftpSndWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()), + -- xftpSndWorkers :: TMap (Maybe XFTPServer) (TMVar (), Async ()), -- files currently in upload - to throttle upload of other files' chunks, -- this optimization can be dropped for the MVP - xftpSndFiles :: TVar (Set DBSndFileId) + -- xftpSndFiles :: TVar (Set DBSndFileId) } newXFTPAgent :: STM XFTPAgent newXFTPAgent = do + xftpWorkDir <- newTVar Nothing xftpWorkers <- TM.empty - xftpSndWorkers <- TM.empty - xftpSndFiles <- newTVar S.empty - pure XFTPAgent {xftpWorkers, xftpSndWorkers, xftpSndFiles} + pure XFTPAgent {xftpWorkDir, xftpWorkers} diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index ebd4a9937..8ba0ea863 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -17,12 +17,10 @@ module Simplex.Messaging.Agent.NtfSubSupervisor ) where -import Control.Concurrent.Async (Async, uninterruptibleCancel) import Control.Concurrent.STM (stateTVar) import Control.Logger.Simple (logError, logInfo) import Control.Monad import Control.Monad.Except -import Control.Monad.IO.Unlift (MonadUnliftIO) import Control.Monad.Reader import Data.Bifunctor (first) import Data.Fixed (Fixed (MkFixed), Pico) @@ -44,10 +42,9 @@ import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Util (tshow, unlessM) import System.Random (randomR) -import UnliftIO (async) +import UnliftIO import UnliftIO.Concurrent (forkIO, threadDelay) import qualified UnliftIO.Exception as E -import UnliftIO.STM runNtfSupervisor :: forall m. AgentMonad' m => AgentClient -> m () runNtfSupervisor c = do @@ -349,12 +346,12 @@ instantNotifications = \case Just NtfToken {ntfTknStatus = NTActive, ntfMode = NMInstant} -> True _ -> False -closeNtfSupervisor :: NtfSupervisor -> IO () +closeNtfSupervisor :: MonadUnliftIO m => NtfSupervisor -> m () closeNtfSupervisor ns = do cancelNtfWorkers_ $ ntfWorkers ns cancelNtfWorkers_ $ ntfSMPWorkers ns -cancelNtfWorkers_ :: TMap (ProtocolServer s) (TMVar (), Async ()) -> IO () +cancelNtfWorkers_ :: MonadUnliftIO m => TMap (ProtocolServer s) (TMVar (), Async ()) -> m () cancelNtfWorkers_ wsVar = do ws <- atomically $ stateTVar wsVar (,M.empty) mapM_ (uninterruptibleCancel . snd) ws diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index ba87160da..52e234b5a 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -13,7 +13,7 @@ import qualified Data.ByteString.Char8 as B import SMPAgentClient (agentCfg, initAgentServers) import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..)) -import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpDeleteRcvFile, xftpReceiveFile, xftpSendFile) +import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpDeleteRcvFile, xftpReceiveFile, xftpSendFile, xftpStartWorkers) import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..)) import Simplex.Messaging.Encoding.String (StrEncoding (..)) import System.Directory (doesDirectoryExist, getFileSize, listDirectory) @@ -49,8 +49,9 @@ testXFTPAgentReceive = withXFTPServer $ do -- receive file using agent rcp <- getSMPAgentClient agentCfg initAgentServers runRight_ $ do + xftpStartWorkers rcp (Just recipientFiles) fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv - fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles) + fId <- xftpReceiveFile rcp 1 fd ("", fId', RFDONE path) <- rfGet rcp liftIO $ do fId' `shouldBe` fId @@ -87,8 +88,9 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do -- receive file using agent - should not succeed due to server being down rcp <- getSMPAgentClient agentCfg initAgentServers fId <- runRight $ do + xftpStartWorkers rcp (Just recipientFiles) fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv - fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles) + fId <- xftpReceiveFile rcp 1 fd liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt pure fId disconnectAgentClient rcp @@ -97,9 +99,10 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do let tmpPath = recipientFiles prefixDir "xftp.encrypted" doesDirectoryExist tmpPath `shouldReturn` True - rcp' <- getSMPAgentClient agentCfg initAgentServers withXFTPServerStoreLogOn $ \_ -> do -- receive file using agent - should succeed with server up + rcp' <- getSMPAgentClient agentCfg initAgentServers + runRight_ $ xftpStartWorkers rcp' (Just recipientFiles) ("", fId', RFDONE path) <- rfGet rcp' liftIO $ do fId' `shouldBe` fId @@ -130,8 +133,9 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do -- receive file using agent - should not succeed due to server being down rcp <- getSMPAgentClient agentCfg initAgentServers fId <- runRight $ do + xftpStartWorkers rcp (Just recipientFiles) fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv - fId <- xftpReceiveFile rcp 1 fd (Just recipientFiles) + fId <- xftpReceiveFile rcp 1 fd liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt pure fId disconnectAgentClient rcp @@ -140,9 +144,10 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do let tmpPath = recipientFiles prefixDir "xftp.encrypted" doesDirectoryExist tmpPath `shouldReturn` True - -- receive file using agent - should fail with AUTH error - rcp' <- getSMPAgentClient agentCfg initAgentServers withXFTPServerThreadOn $ \_ -> do + -- receive file using agent - should fail with AUTH error + rcp' <- getSMPAgentClient agentCfg initAgentServers + runRight_ $ xftpStartWorkers rcp' (Just recipientFiles) ("", fId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp' fId' `shouldBe` fId @@ -160,20 +165,17 @@ testXFTPAgentSendExperimental = withXFTPServer $ do -- send file using experimental agent API sndr <- getSMPAgentClient agentCfg initAgentServers rfd <- runRight $ do - sfId <- xftpSendFile sndr 1 filePath 2 $ Just senderFiles - ("", sfId', SFDONE sndDescr rcvDescrs) <- sfGet sndr - liftIO $ do - sfId' `shouldBe` sfId - strDecode <$> B.readFile (senderFiles "testfile.descr/testfile.xftp/snd.xftp.private") `shouldReturn` Right sndDescr - Right rfd1 <- strDecode <$> B.readFile (senderFiles "testfile.descr/testfile.xftp/rcv1.xftp") - Right rfd2 <- strDecode <$> B.readFile (senderFiles "testfile.descr/testfile.xftp/rcv2.xftp") - rcvDescrs `shouldMatchList` [rfd1, rfd2] - pure rfd1 + xftpStartWorkers sndr (Just senderFiles) + sfId <- xftpSendFile sndr 1 filePath 2 + ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr + liftIO $ sfId' `shouldBe` sfId + pure rfd1 -- receive file using agent rcp <- getSMPAgentClient agentCfg initAgentServers runRight_ $ do - rfId <- xftpReceiveFile rcp 1 rfd (Just recipientFiles) + xftpStartWorkers rcp (Just recipientFiles) + rfId <- xftpReceiveFile rcp 1 rfd ("", rfId', RFDONE path) <- rfGet rcp liftIO $ do rfId' `shouldBe` rfId