diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 399b4e6be..9e091b9f1 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -33,13 +33,15 @@ import Data.Bifunctor (first) import Data.ByteString (ByteString) import qualified Data.ByteString.Base64.URL as U import qualified Data.ByteString.Char8 as B -import Data.List (isSuffixOf, partition) +import Data.Composition ((.:)) +import Data.Int (Int64) +import Data.List (foldl', 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) +import Simplex.FileTransfer.Client.Main (CLIError, SendOptions (..), cliSendFileOpts) import Simplex.FileTransfer.Crypto import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI) @@ -52,7 +54,7 @@ import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Agent.RetryInterval import Simplex.Messaging.Agent.Store.SQLite import Simplex.Messaging.Encoding.String -import Simplex.Messaging.Protocol (XFTPServer, XFTPServerWithAuth) +import Simplex.Messaging.Protocol (EntityId, XFTPServer, XFTPServerWithAuth) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Util (liftError, liftIOEither, tshow) import System.FilePath (takeFileName, ()) @@ -164,24 +166,28 @@ runXFTPWorker c srv doWork = do atomically $ beginAgentOperation c AORcvNetwork loop downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> m () - downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do + downloadFileChunk RcvFileChunk {userId, rcvFileId, rcvFileEntityId, rcvChunkId, chunkNo, chunkSize, digest, fileTmpPath} replica = do 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 relChunkPath - let fileReceived = allChunksReceived f - when fileReceived $ - liftIO $ updateRcvFileStatus db rcvFileId RFSReceived - pure fileReceived - when fileReceived $ addXFTPWorker c Nothing + (complete, progress) <- withStore c $ \db -> runExceptT $ do + RcvFile {size = FileSize total, chunks} <- + ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId relChunkPath + let rcvd = receivedSize chunks + complete = all chunkReceived chunks + liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived + pure (complete, RFPROG rcvd total) + liftIO $ notify c rcvFileEntityId progress + when complete $ addXFTPWorker c Nothing where - allChunksReceived :: RcvFile -> Bool - allChunksReceived RcvFile {chunks} = - all (\RcvFileChunk {replicas} -> any received replicas) chunks + receivedSize :: [RcvFileChunk] -> Int64 + receivedSize = foldl' (\sz ch -> sz + receivedChunkSize ch) 0 + receivedChunkSize ch@RcvFileChunk {chunkSize = s} + | chunkReceived ch = fromIntegral (unFileSize s) + | otherwise = 0 + chunkReceived RcvFileChunk {replicas} = any received replicas workerInternalError :: AgentMonad m => AgentClient -> DBRcvFileId -> RcvFileId -> Maybe FilePath -> String -> m () workerInternalError c rcvFileId rcvFileEntityId tmpPath internalErrStr = do @@ -193,7 +199,7 @@ notifyInternalError :: (MonadUnliftIO m) => AgentClient -> RcvFileId -> String - notifyInternalError AgentClient {subQ} rcvFileEntityId internalErrStr = atomically $ writeTBQueue subQ ("", rcvFileEntityId, APC SAERcvFile $ RFERR $ INTERNAL internalErrStr) runXFTPLocalWorker :: forall m. AgentMonad m => AgentClient -> TMVar () -> m () -runXFTPLocalWorker c@AgentClient {subQ} doWork = do +runXFTPLocalWorker c doWork = do forever $ do void . atomically $ readTMVar doWork -- TODO agentOperationBracket? @@ -219,10 +225,8 @@ runXFTPLocalWorker c@AgentClient {subQ} doWork = do void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure fsSavePath forM_ tmpPath (removePath <=< toFSFilePath) withStore' c (`updateRcvFileComplete` rcvFileId) - notify $ RFDONE fsSavePath + liftIO $ notify c rcvFileEntityId $ RFDONE fsSavePath where - notify :: forall e. AEntityI e => ACommand 'Agent e -> m () - notify cmd = atomically $ writeTBQueue subQ ("", rcvFileEntityId, APC (sAEntity @e) cmd) getChunkPaths :: [RcvFileChunk] -> m [FilePath] getChunkPaths [] = pure [] getChunkPaths (RcvFileChunk {chunkTmpPath = Just path} : cs) = do @@ -242,7 +246,7 @@ deleteRcvFile c userId rcvFileEntityId = do else withStore' c (`updateRcvFileDeleted` rcvFileId) sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId -sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipients = do +sendFileExperimental c@AgentClient {xftpServers} userId filePath numRecipients = do g <- asks idsDrg sndFileId <- liftIO $ randomId g 12 xftpSrvs <- atomically $ TM.lookup userId xftpServers @@ -269,11 +273,11 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien tempPath = Just tempPath, verbose = False } - liftCLI $ cliSendFile sendOptions + liftCLI $ cliSendFileOpts sendOptions False $ notify c sndFileId .: SFPROG (sndDescr, rcvDescrs) <- readDescrs outputDir fileName removePath tempPath removePath outputDir - notify sndFileId $ SFDONE sndDescr rcvDescrs + liftIO $ notify c sndFileId $ SFDONE sndDescr rcvDescrs liftCLI :: ExceptT CLIError IO () -> m () liftCLI = either (throwError . INTERNAL . show) pure <=< liftIO . runExceptT readDescrs :: FilePath -> FilePath -> m (ValidFileDescription 'FSender, [ValidFileDescription 'FRecipient]) @@ -286,8 +290,9 @@ sendFileExperimental AgentClient {subQ, xftpServers} userId filePath numRecipien (,) <$> readDescr sdFile <*> mapM readDescr rdFiles' readDescr :: FilePartyI p => FilePath -> m (ValidFileDescription p) readDescr f = liftIOEither $ first INTERNAL . strDecode <$> B.readFile f - notify :: forall e. AEntityI e => SndFileId -> ACommand 'Agent e -> m () - notify sndFileId cmd = atomically $ writeTBQueue subQ ("", sndFileId, APC (sAEntity @e) cmd) + +notify :: forall e. AEntityI e => AgentClient -> EntityId -> ACommand 'Agent e -> IO () +notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, APC (sAEntity @e) cmd) -- _sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId _sendFile :: AgentClient -> UserId -> FilePath -> Int -> m SndFileId diff --git a/src/Simplex/FileTransfer/Client/Main.hs b/src/Simplex/FileTransfer/Client/Main.hs index 5ed5d5eba..7f9186c73 100644 --- a/src/Simplex/FileTransfer/Client/Main.hs +++ b/src/Simplex/FileTransfer/Client/Main.hs @@ -14,6 +14,7 @@ module Simplex.FileTransfer.Client.Main CLIError (..), xftpClientCLI, cliSendFile, + cliSendFileOpts, prepareChunkSizes, ) where @@ -257,11 +258,14 @@ runE a = _ -> pure () cliSendFile :: SendOptions -> ExceptT CLIError IO () -cliSendFile SendOptions {filePath, outputDir, numRecipients, xftpServers, retryCount, tempPath, verbose} = do +cliSendFile opts = cliSendFileOpts opts True $ printProgress "Uploaded" + +cliSendFileOpts :: SendOptions -> Bool -> (Int64 -> Int64 -> IO ()) -> ExceptT CLIError IO () +cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, retryCount, tempPath, verbose} printInfo notifyProgress = do let (_, fileName) = splitFileName filePath - liftIO $ printNoNewLine "Encrypting file..." + liftIO $ when printInfo $ printNoNewLine "Encrypting file..." (encPath, fdRcv, fdSnd, chunkSpecs, encSize) <- encryptFileForUpload fileName - liftIO $ printNoNewLine "Uploading file..." + liftIO $ when printInfo $ printNoNewLine "Uploading file..." uploadedChunks <- newTVarIO [] sentChunks <- uploadFile chunkSpecs uploadedChunks encSize whenM (doesFileExist encPath) $ removeFile encPath @@ -270,9 +274,10 @@ cliSendFile SendOptions {filePath, outputDir, numRecipients, xftpServers, retryC let fdRcvs = createRcvFileDescriptions fdRcv sentChunks fdSnd' = createSndFileDescription fdSnd sentChunks (fdRcvPaths, fdSndPath) <- writeFileDescriptions fileName fdRcvs fdSnd' - printNoNewLine "File uploaded!" - putStrLn $ "\nSender file description: " <> fdSndPath - putStrLn "Pass file descriptions to the recipient(s):" + when printInfo $ do + printNoNewLine "File uploaded!" + putStrLn $ "\nSender file description: " <> fdSndPath + putStrLn "Pass file descriptions to the recipient(s):" forM_ fdRcvPaths putStrLn where encryptFileForUpload :: String -> ExceptT CLIError IO (FilePath, FileDescription 'FRecipient, FileDescription 'FSender, [XFTPChunkSpec], Int64) @@ -322,7 +327,7 @@ cliSendFile SendOptions {filePath, outputDir, numRecipients, xftpServers, retryC uploaded <- atomically . stateTVar uploadedChunks $ \cs -> let cs' = fromIntegral chunkSize : cs in (sum cs', cs') liftIO $ do - printProgress "Uploaded" uploaded encSize + notifyProgress uploaded encSize when verbose $ putStrLn "" let recipients = L.toList $ L.map ChunkReplicaId rIds `L.zip` L.map snd rKeys replicas = [SentFileChunkReplica {server = xftpServer, recipients}] diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 7d7d1180b..25d0a6187 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -336,10 +336,10 @@ data ACommand (p :: AParty) (e :: AEntity) where ERR :: AgentErrorType -> ACommand Agent AEConn SUSPENDED :: ACommand Agent AENone -- XFTP commands and responses - RFPROG :: Int -> Int -> ACommand Agent AERcvFile + RFPROG :: Int64 -> Int64 -> ACommand Agent AERcvFile RFDONE :: FilePath -> ACommand Agent AERcvFile RFERR :: AgentErrorType -> ACommand Agent AERcvFile - SFPROG :: Int -> Int -> ACommand Agent AESndFile + SFPROG :: Int64 -> Int64 -> ACommand Agent AESndFile SFDONE :: ValidFileDescription 'FSender -> [ValidFileDescription 'FRecipient] -> ACommand Agent AESndFile deriving instance Eq (ACommand p e) diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index 52e234b5a..bd4f3beee 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -10,10 +10,11 @@ import Control.Logger.Simple import Control.Monad.Except import Data.Bifunctor (first) import qualified Data.ByteString.Char8 as B +import Data.Int (Int64) import SMPAgentClient (agentCfg, initAgentServers) import Simplex.FileTransfer.Description import Simplex.FileTransfer.Protocol (FileParty (..)) -import Simplex.Messaging.Agent (disconnectAgentClient, getSMPAgentClient, xftpDeleteRcvFile, xftpReceiveFile, xftpSendFile, xftpStartWorkers) +import Simplex.Messaging.Agent (AgentClient, 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) @@ -30,6 +31,29 @@ xftpAgentTests = around_ testBracket . describe "Functional API" $ do it "should cleanup tmp path after permanent error" testXFTPAgentReceiveCleanup it "should send file using experimental api" testXFTPAgentSendExperimental +rfProgress :: (MonadIO m, MonadFail m) => AgentClient -> Int64 -> m () +rfProgress c expected = loop 0 + where + loop prev = do + (_, _, RFPROG rcvd total) <- rfGet c + checkProgress (prev, expected) (rcvd, total) loop + +sfProgress :: (MonadIO m, MonadFail m) => AgentClient -> Int64 -> m () +sfProgress c expected = loop 0 + where + loop prev = do + (_, _, SFPROG sent total) <- sfGet c + checkProgress (prev, expected) (sent, total) loop + +-- checks that progress increases till it reaches total +checkProgress :: MonadIO m => (Int64, Int64) -> (Int64, Int64) -> (Int64 -> m ()) -> m () +checkProgress (prev, expected) (progress, total) loop + | total /= expected = error "total /= expected" + | progress <= prev = error "progress <= prev" + | progress > total = error "progress > total" + | progress < total = loop progress + | otherwise = pure () + testXFTPAgentReceive :: IO () testXFTPAgentReceive = withXFTPServer $ do -- send file using CLI @@ -52,6 +76,7 @@ testXFTPAgentReceive = withXFTPServer $ do xftpStartWorkers rcp (Just recipientFiles) fd :: ValidFileDescription 'FRecipient <- getFileDescription fdRcv fId <- xftpReceiveFile rcp 1 fd + rfProgress rcp $ mb 18 ("", fId', RFDONE path) <- rfGet rcp liftIO $ do fId' `shouldBe` fId @@ -103,6 +128,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do -- receive file using agent - should succeed with server up rcp' <- getSMPAgentClient agentCfg initAgentServers runRight_ $ xftpStartWorkers rcp' (Just recipientFiles) + rfProgress rcp' $ mb 18 ("", fId', RFDONE path) <- rfGet rcp' liftIO $ do fId' `shouldBe` fId @@ -167,6 +193,7 @@ testXFTPAgentSendExperimental = withXFTPServer $ do rfd <- runRight $ do xftpStartWorkers sndr (Just senderFiles) sfId <- xftpSendFile sndr 1 filePath 2 + sfProgress sndr $ mb 18 ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr liftIO $ sfId' `shouldBe` sfId pure rfd1 @@ -176,6 +203,7 @@ testXFTPAgentSendExperimental = withXFTPServer $ do runRight_ $ do xftpStartWorkers rcp (Just recipientFiles) rfId <- xftpReceiveFile rcp 1 rfd + rfProgress rcp $ mb 18 ("", rfId', RFDONE path) <- rfGet rcp liftIO $ do rfId' `shouldBe` rfId