From d1774e5b5608a13c9d37ccbc9c4772eed51def68 Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Tue, 11 Apr 2023 19:36:51 +0400 Subject: [PATCH] xftp: send progress events (#717) --- src/Simplex/FileTransfer/Agent.hs | 19 ++++++++++++++----- src/Simplex/Messaging/Agent/Store/SQLite.hs | 10 +++++----- tests/XFTPAgent.hs | 6 +++--- 3 files changed, 22 insertions(+), 13 deletions(-) diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index 340cfe8bf..75624426f 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -195,8 +195,8 @@ runXFTPRcvWorker c srv doWork = do relChunkPath = fileTmpPath takeFileName chunkPath agentXFTPDownloadChunk c userId rcvChunkId replica chunkSpec (complete, progress) <- withStore c $ \db -> runExceptT $ do - RcvFile {size = FileSize total, chunks} <- - ExceptT $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId rcvFileId relChunkPath + liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath + RcvFile {size = FileSize total, chunks} <- ExceptT $ getRcvFile db rcvFileId let rcvd = receivedSize chunks complete = all chunkReceived chunks liftIO . when complete $ updateRcvFileStatus db rcvFileId RFSReceived @@ -474,13 +474,15 @@ runXFTPSndWorker c srv doWork = do sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded getSndFile db sndFileId - let complete = all chunkUploaded chunks - -- TODO calculate progress, notify SFPROG + let uploaded = uploadedSize chunks + total = totalSize chunks + complete = all chunkUploaded chunks + notify c sndFileEntityId $ SFPROG uploaded total when complete $ do (sndDescr, rcvDescrs) <- sndFileToDescrs sf notify c sndFileEntityId $ SFDONE sndDescr rcvDescrs forM_ prefixPath (removePath <=< toFSFilePath) - withStore' c $ \db -> updateSndFileStatus db sndFileId SFSComplete + withStore' c $ \db -> updateSndFileComplete db sndFileId where addRecipients :: SndFileChunk -> SndFileChunkReplica -> m SndFileChunkReplica addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {rcvIdsKeys} @@ -547,6 +549,13 @@ runXFTPSndWorker c srv doWork = do Just ch@FileChunk {replicas} -> ch {replicas = replica' : replicas} _ -> FileChunk {chunkNo, digest, chunkSize, replicas = [replica']} replica' = FileChunkReplica {server, replicaId, replicaKey} + uploadedSize :: [SndFileChunk] -> Int64 + uploadedSize = foldl' (\sz ch -> sz + uploadedChunkSize ch) 0 + uploadedChunkSize ch + | chunkUploaded ch = fromIntegral (sndChunkSize ch) + | otherwise = 0 + totalSize :: [SndFileChunk] -> Int64 + totalSize = foldl' (\sz ch -> sz + fromIntegral (sndChunkSize ch)) 0 chunkUploaded :: SndFileChunk -> Bool chunkUploaded SndFileChunk {replicas} = any (\SndFileChunkReplica {replicaStatus} -> replicaStatus == SFRSUploaded) replicas diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 7da038e01..480a477c0 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -134,6 +134,7 @@ module Simplex.Messaging.Agent.Store.SQLite -- Rcv files createRcvFile, + getRcvFile, getRcvFileByEntityId, updateRcvChunkReplicaDelay, updateRcvFileChunkReceived, @@ -1947,12 +1948,11 @@ updateRcvChunkReplicaDelay db replicaId delay = do updatedAt <- getCurrentTime DB.execute db "UPDATE rcv_file_chunk_replicas SET delay = ?, retries = retries + 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (delay, updatedAt, replicaId) -updateRcvFileChunkReceived :: DB.Connection -> Int64 -> Int64 -> DBRcvFileId -> FilePath -> IO (Either StoreError RcvFile) -updateRcvFileChunkReceived db rId cId fId chunkTmpPath = do +updateRcvFileChunkReceived :: DB.Connection -> Int64 -> Int64 -> FilePath -> IO () +updateRcvFileChunkReceived db replicaId chunkId chunkTmpPath = do updatedAt <- getCurrentTime - DB.execute db "UPDATE rcv_file_chunk_replicas SET received = 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (updatedAt, rId) - DB.execute db "UPDATE rcv_file_chunks SET tmp_path = ?, updated_at = ? WHERE rcv_file_chunk_id = ?" (chunkTmpPath, updatedAt, cId) - getRcvFile db fId + DB.execute db "UPDATE rcv_file_chunk_replicas SET received = 1, updated_at = ? WHERE rcv_file_chunk_replica_id = ?" (updatedAt, replicaId) + DB.execute db "UPDATE rcv_file_chunks SET tmp_path = ?, updated_at = ? WHERE rcv_file_chunk_id = ?" (chunkTmpPath, updatedAt, chunkId) updateRcvFileStatus :: DB.Connection -> DBRcvFileId -> RcvFileStatus -> IO () updateRcvFileStatus db rcvFileId status = do diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index b80c7c510..8c2b2f814 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -81,7 +81,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do rfd <- runRight $ do xftpStartWorkers sndr (Just senderFiles) sfId <- xftpSendFile sndr 1 filePath 2 - -- sfProgress sndr $ mb 18 + sfProgress sndr $ mb 18 ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr liftIO $ sfId' `shouldBe` sfId pure rfd1 @@ -121,7 +121,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do runRight $ do xftpStartWorkers sndr (Just senderFiles) sfId <- xftpSendFile sndr 1 filePath 2 - -- sfProgress sndr $ mb 18 + sfProgress sndr $ mb 18 ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr liftIO $ sfId' `shouldBe` sfId pure rfd1 @@ -166,7 +166,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do runRight $ do xftpStartWorkers sndr (Just senderFiles) sfId <- xftpSendFile sndr 1 filePath 2 - -- sfProgress sndr $ mb 18 + sfProgress sndr $ mb 18 ("", sfId', SFDONE _sndDescr [rfd1, _rfd2]) <- sfGet sndr liftIO $ sfId' `shouldBe` sfId pure rfd1