refactor(sync): перевёл журнал на map по пути и убрал цикл debounce-sync
Журнал хранится как словарь path→entry, служебные пути исключены из sync. Apply пишет файлы без записи в журнал; bootstrap не триггерит sync во время работы.
This commit is contained in:
@@ -1,5 +1,6 @@
|
|||||||
package com.github.nullptroma.wallenc.app.sync
|
package com.github.nullptroma.wallenc.app.sync
|
||||||
|
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IVaultsManager
|
import com.github.nullptroma.wallenc.domain.interfaces.IVaultsManager
|
||||||
import com.github.nullptroma.wallenc.ui.R
|
import com.github.nullptroma.wallenc.ui.R
|
||||||
import com.github.nullptroma.wallenc.ui.resources.UiStringResolver
|
import com.github.nullptroma.wallenc.ui.resources.UiStringResolver
|
||||||
@@ -10,6 +11,7 @@ import kotlinx.coroutines.FlowPreview
|
|||||||
import kotlinx.coroutines.SupervisorJob
|
import kotlinx.coroutines.SupervisorJob
|
||||||
import kotlinx.coroutines.flow.collectLatest
|
import kotlinx.coroutines.flow.collectLatest
|
||||||
import kotlinx.coroutines.flow.debounce
|
import kotlinx.coroutines.flow.debounce
|
||||||
|
import kotlinx.coroutines.flow.filter
|
||||||
import kotlinx.coroutines.flow.map
|
import kotlinx.coroutines.flow.map
|
||||||
import kotlinx.coroutines.flow.merge
|
import kotlinx.coroutines.flow.merge
|
||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
@@ -35,14 +37,28 @@ class StorageSyncBootstrap @Inject constructor(
|
|||||||
}
|
}
|
||||||
val triggers = storages.flatMap { storage ->
|
val triggers = storages.flatMap { storage ->
|
||||||
listOf(
|
listOf(
|
||||||
storage.accessor.filesUpdates.map {},
|
storage.accessor.filesUpdates
|
||||||
storage.accessor.dirsUpdates.map {},
|
.filter { page ->
|
||||||
|
page.data.any { file ->
|
||||||
|
StorageSyncPaths.isSyncableUserPath(file.metaInfo.path)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
.map {},
|
||||||
|
storage.accessor.dirsUpdates
|
||||||
|
.filter { page ->
|
||||||
|
page.data.any { dir ->
|
||||||
|
StorageSyncPaths.isSyncableUserPath(dir.metaInfo.path)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
.map {},
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
merge(*triggers.toTypedArray())
|
merge(*triggers.toTypedArray())
|
||||||
.debounce(DEBOUNCE_AFTER_CHANGE_MS)
|
.debounce(DEBOUNCE_AFTER_CHANGE_MS)
|
||||||
.collect {
|
.collect {
|
||||||
// RunStorageSyncUseCase.enqueue отбрасывает повтор, пока sync уже в очереди/в работе.
|
if (syncRunner.syncRunning.value) {
|
||||||
|
return@collect
|
||||||
|
}
|
||||||
syncRunner.enqueue(
|
syncRunner.enqueue(
|
||||||
displayTitle = uiStrings(R.string.task_title_storage_sync_background),
|
displayTitle = uiStrings(R.string.task_title_storage_sync_background),
|
||||||
logReason = "debounce",
|
logReason = "debounce",
|
||||||
|
|||||||
@@ -0,0 +1,29 @@
|
|||||||
|
package com.github.nullptroma.wallenc.domain.vault.storages.common
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import java.util.LinkedHashMap
|
||||||
|
|
||||||
|
internal object StorageSyncJournalCodec {
|
||||||
|
private val journalJavaType = ObjectMapper().typeFactory.constructMapType(
|
||||||
|
LinkedHashMap::class.java,
|
||||||
|
String::class.java,
|
||||||
|
StorageSyncJournalEntry::class.java,
|
||||||
|
)
|
||||||
|
|
||||||
|
fun read(mapper: ObjectMapper, bytes: ByteArray): StorageSyncJournal {
|
||||||
|
if (bytes.isEmpty()) {
|
||||||
|
return emptyMap()
|
||||||
|
}
|
||||||
|
return runCatching {
|
||||||
|
@Suppress("UNCHECKED_CAST")
|
||||||
|
mapper.readValue(bytes, journalJavaType) as StorageSyncJournal
|
||||||
|
}.getOrElse {
|
||||||
|
emptyMap<String, StorageSyncJournalEntry>()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fun write(mapper: ObjectMapper, journal: StorageSyncJournal): ByteArray =
|
||||||
|
mapper.writeValueAsBytes(journal)
|
||||||
|
}
|
||||||
@@ -1,6 +1,8 @@
|
|||||||
package com.github.nullptroma.wallenc.domain.vault.storages.encrypt
|
package com.github.nullptroma.wallenc.domain.vault.storages.encrypt
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.module.kotlin.readValue
|
||||||
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
||||||
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.StorageSyncJournalCodec
|
||||||
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
||||||
import com.github.nullptroma.wallenc.domain.vault.utils.CloseHandledStreamExtension.Companion.onClosed
|
import com.github.nullptroma.wallenc.domain.vault.utils.CloseHandledStreamExtension.Companion.onClosed
|
||||||
import com.github.nullptroma.wallenc.domain.vault.utils.CloseHandledStreamExtension.Companion.onClosing
|
import com.github.nullptroma.wallenc.domain.vault.utils.CloseHandledStreamExtension.Companion.onClosing
|
||||||
@@ -15,7 +17,10 @@ import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
|||||||
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IMetaInfo
|
import com.github.nullptroma.wallenc.domain.interfaces.IMetaInfo
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
||||||
@@ -257,7 +262,8 @@ class EncryptedStorageAccessor(
|
|||||||
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE)
|
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE)
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun openWrite(path: String): OutputStream = openWriteInternal(path, recordJournal = true)
|
override suspend fun openWrite(path: String, recordSyncJournal: Boolean): OutputStream =
|
||||||
|
openWriteInternal(path, recordJournal = recordSyncJournal)
|
||||||
|
|
||||||
override suspend fun openRead(path: String): InputStream {
|
override suspend fun openRead(path: String): InputStream {
|
||||||
val stream = source.openRead(encryptPath(path))
|
val stream = source.openRead(encryptPath(path))
|
||||||
@@ -290,35 +296,22 @@ class EncryptedStorageAccessor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun readSyncJournal(): List<StorageSyncJournalEntry> {
|
override suspend fun readSyncJournal(): StorageSyncJournal {
|
||||||
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
||||||
if (bytes.isEmpty()) {
|
return StorageSyncJournalCodec.read(jackson, bytes)
|
||||||
return emptyList()
|
|
||||||
}
|
|
||||||
return runCatching {
|
|
||||||
val journalType = jackson.typeFactory.constructCollectionType(
|
|
||||||
List::class.java,
|
|
||||||
StorageSyncJournalEntry::class.java,
|
|
||||||
)
|
|
||||||
jackson.readValue<List<StorageSyncJournalEntry>>(bytes, journalType)
|
|
||||||
}.getOrElse {
|
|
||||||
emptyList()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun appendSyncJournal(entries: List<StorageSyncJournalEntry>) {
|
override suspend fun putSyncJournalEntries(entries: StorageSyncJournal) {
|
||||||
if (entries.isEmpty()) {
|
if (entries.isEmpty()) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
val next = readSyncJournal().toMutableList().apply { addAll(entries) }
|
val merged = StorageSyncJournalMerge.merge(readSyncJournal(), entries)
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
writeSyncJournal(merged)
|
||||||
jackson.writeValue(out, next)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun rewriteSyncJournal(entries: List<StorageSyncJournalEntry>) {
|
private suspend fun writeSyncJournal(journal: StorageSyncJournal) {
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
||||||
jackson.writeValue(out, entries)
|
jackson.writeValue(out, journal)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -376,15 +369,15 @@ class EncryptedStorageAccessor(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
||||||
val cleanedPath = if (path.startsWith("/")) path else "/$path"
|
val cleanedPath = StorageSyncPaths.normalize(path)
|
||||||
if (cleanedPath.startsWith("/$systemHiddenDirName/")) {
|
if (!StorageSyncPaths.isSyncableUserPath(cleanedPath)) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
val entries = readSyncJournal()
|
val journal = readSyncJournal()
|
||||||
val nextSequence = (entries.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
val nextSequence = (journal.values.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
||||||
appendSyncJournal(
|
putSyncJournalEntries(
|
||||||
listOf(
|
mapOf(
|
||||||
StorageSyncJournalEntry(
|
cleanedPath to StorageSyncJournalEntry(
|
||||||
path = cleanedPath,
|
path = cleanedPath,
|
||||||
operation = operation,
|
operation = operation,
|
||||||
revision = StorageSyncRevision(
|
revision = StorageSyncRevision(
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
package com.github.nullptroma.wallenc.domain.vault.storages.local
|
package com.github.nullptroma.wallenc.domain.vault.storages.local
|
||||||
|
|
||||||
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
||||||
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.StorageSyncJournalCodec
|
||||||
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.JacksonException
|
import com.fasterxml.jackson.core.JacksonException
|
||||||
@@ -15,9 +16,12 @@ import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
|||||||
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IMetaInfo
|
import com.github.nullptroma.wallenc.domain.interfaces.IMetaInfo
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
||||||
import kotlinx.coroutines.CoroutineDispatcher
|
import kotlinx.coroutines.CoroutineDispatcher
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
@@ -523,7 +527,7 @@ class LocalStorageAccessor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun openWrite(path: String): OutputStream = withContext(ioDispatcher) {
|
override suspend fun openWrite(path: String, recordSyncJournal: Boolean): OutputStream = withContext(ioDispatcher) {
|
||||||
touchFileInternal(path, recordJournal = false)
|
touchFileInternal(path, recordJournal = false)
|
||||||
val pair = LocalStorageFilePair.from(_filesystemBasePath, path)
|
val pair = LocalStorageFilePair.from(_filesystemBasePath, path)
|
||||||
?: throw WallencException.Storage.FileNotFound()
|
?: throw WallencException.Storage.FileNotFound()
|
||||||
@@ -531,7 +535,9 @@ class LocalStorageAccessor(
|
|||||||
CoroutineScope(ioDispatcher).launch {
|
CoroutineScope(ioDispatcher).launch {
|
||||||
touchFileInternal(path, recordJournal = false)
|
touchFileInternal(path, recordJournal = false)
|
||||||
scanSizeAndNumOfFiles()
|
scanSizeAndNumOfFiles()
|
||||||
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
|
if (recordSyncJournal) {
|
||||||
|
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -572,33 +578,22 @@ class LocalStorageAccessor(
|
|||||||
return@withContext file.outputStream()
|
return@withContext file.outputStream()
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun readSyncJournal(): List<StorageSyncJournalEntry> = withContext(ioDispatcher) {
|
override suspend fun readSyncJournal(): StorageSyncJournal = withContext(ioDispatcher) {
|
||||||
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
||||||
if (bytes.isEmpty()) {
|
StorageSyncJournalCodec.read(jackson, bytes)
|
||||||
return@withContext emptyList()
|
|
||||||
}
|
|
||||||
return@withContext runCatching {
|
|
||||||
jackson.readValue<List<StorageSyncJournalEntry>>(bytes)
|
|
||||||
}.getOrElse {
|
|
||||||
emptyList()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun appendSyncJournal(entries: List<StorageSyncJournalEntry>) = withContext(ioDispatcher) {
|
override suspend fun putSyncJournalEntries(entries: StorageSyncJournal) = withContext(ioDispatcher) {
|
||||||
if (entries.isEmpty()) {
|
if (entries.isEmpty()) {
|
||||||
return@withContext
|
return@withContext
|
||||||
}
|
}
|
||||||
val next = readSyncJournal().toMutableList().apply {
|
val merged = StorageSyncJournalMerge.merge(readSyncJournal(), entries)
|
||||||
addAll(entries)
|
writeSyncJournal(merged)
|
||||||
}
|
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
|
||||||
jackson.writeValue(out, next)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun rewriteSyncJournal(entries: List<StorageSyncJournalEntry>) = withContext(ioDispatcher) {
|
private suspend fun writeSyncJournal(journal: StorageSyncJournal) {
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
||||||
jackson.writeValue(out, entries)
|
jackson.writeValue(out, journal)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -722,19 +717,25 @@ class LocalStorageAccessor(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
||||||
val cleanedPath = if (path.startsWith("/")) path else "/$path"
|
val cleanedPath = StorageSyncPaths.normalize(path)
|
||||||
val entries = readSyncJournal()
|
if (!StorageSyncPaths.isSyncableUserPath(cleanedPath)) {
|
||||||
val nextSequence = (entries.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
return
|
||||||
val entry = StorageSyncJournalEntry(
|
}
|
||||||
path = cleanedPath,
|
val journal = readSyncJournal()
|
||||||
operation = operation,
|
val nextSequence = (journal.values.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
||||||
revision = StorageSyncRevision(
|
putSyncJournalEntries(
|
||||||
sequence = nextSequence,
|
mapOf(
|
||||||
actorId = syncActorId,
|
cleanedPath to StorageSyncJournalEntry(
|
||||||
createdAt = Instant.now(),
|
path = cleanedPath,
|
||||||
|
operation = operation,
|
||||||
|
revision = StorageSyncRevision(
|
||||||
|
sequence = nextSequence,
|
||||||
|
actorId = syncActorId,
|
||||||
|
createdAt = Instant.now(),
|
||||||
|
),
|
||||||
|
),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
appendSyncJournal(listOf(entry))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
companion object {
|
companion object {
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package com.github.nullptroma.wallenc.domain.vault.storages.yandex
|
|||||||
|
|
||||||
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
import com.github.nullptroma.wallenc.domain.errors.WallencException
|
||||||
import com.github.nullptroma.wallenc.domain.vault.errors.toVaultWallencException
|
import com.github.nullptroma.wallenc.domain.vault.errors.toVaultWallencException
|
||||||
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.StorageSyncJournalCodec
|
||||||
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
import com.github.nullptroma.wallenc.domain.vault.storages.common.readSystemFileBytesOrEmpty
|
||||||
|
|
||||||
import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
|
import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
|
||||||
@@ -17,7 +18,10 @@ import com.github.nullptroma.wallenc.domain.datatypes.DataPage
|
|||||||
import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
||||||
@@ -527,13 +531,13 @@ class YandexStorageAccessor(
|
|||||||
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE)
|
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE)
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun openWrite(path: String): OutputStream = withContext(ioDispatcher) {
|
override suspend fun openWrite(path: String, recordSyncJournal: Boolean): OutputStream = withContext(ioDispatcher) {
|
||||||
touchParentDirs(path)
|
touchParentDirs(path)
|
||||||
val tmp = File.createTempFile("wallenc-yandex-", ".upload")
|
val tmp = File.createTempFile("wallenc-yandex-", ".upload")
|
||||||
val fos = FileOutputStream(tmp)
|
val fos = FileOutputStream(tmp)
|
||||||
fos.onClosed {
|
fos.onClosed {
|
||||||
runCommitAfterStreamClosed {
|
runCommitAfterStreamClosed {
|
||||||
commitUploadedFile(path, tmp)
|
commitUploadedFile(path, tmp, recordSyncJournal)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -571,37 +575,22 @@ class YandexStorageAccessor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun readSyncJournal(): List<StorageSyncJournalEntry> = withContext(ioDispatcher) {
|
override suspend fun readSyncJournal(): StorageSyncJournal = withContext(ioDispatcher) {
|
||||||
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
val bytes = readSystemFileBytesOrEmpty { openReadSystemFile(SYNC_JOURNAL_FILENAME) }
|
||||||
if (bytes.isEmpty()) {
|
StorageSyncJournalCodec.read(statsMapper, bytes)
|
||||||
return@withContext emptyList()
|
|
||||||
}
|
|
||||||
return@withContext runCatching {
|
|
||||||
val journalType = statsMapper.typeFactory.constructCollectionType(
|
|
||||||
List::class.java,
|
|
||||||
StorageSyncJournalEntry::class.java,
|
|
||||||
)
|
|
||||||
statsMapper.readValue<List<StorageSyncJournalEntry>>(bytes, journalType)
|
|
||||||
}.getOrElse {
|
|
||||||
emptyList()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun appendSyncJournal(entries: List<StorageSyncJournalEntry>) = withContext(ioDispatcher) {
|
override suspend fun putSyncJournalEntries(entries: StorageSyncJournal) = withContext(ioDispatcher) {
|
||||||
if (entries.isEmpty()) {
|
if (entries.isEmpty()) {
|
||||||
return@withContext
|
return@withContext
|
||||||
}
|
}
|
||||||
val next = readSyncJournal().toMutableList().apply {
|
val merged = StorageSyncJournalMerge.merge(readSyncJournal(), entries)
|
||||||
addAll(entries)
|
writeSyncJournal(merged)
|
||||||
}
|
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
|
||||||
statsMapper.writeValue(out, next)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun rewriteSyncJournal(entries: List<StorageSyncJournalEntry>) = withContext(ioDispatcher) {
|
private suspend fun writeSyncJournal(journal: StorageSyncJournal) {
|
||||||
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
openWriteSystemFile(SYNC_JOURNAL_FILENAME).use { out ->
|
||||||
statsMapper.writeValue(out, entries)
|
statsMapper.writeValue(out, journal)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -684,7 +673,7 @@ class YandexStorageAccessor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun commitUploadedFile(path: String, tmp: File) {
|
private suspend fun commitUploadedFile(path: String, tmp: File, recordSyncJournal: Boolean) {
|
||||||
try {
|
try {
|
||||||
val diskPath = toDiskPath(path)
|
val diskPath = toDiskPath(path)
|
||||||
val prior = guard { repo.getOrNull(diskPath) }
|
val prior = guard { repo.getOrNull(diskPath) }
|
||||||
@@ -708,27 +697,35 @@ class YandexStorageAccessor(
|
|||||||
info?.let {
|
info?.let {
|
||||||
_filesUpdates.emit(DataPage(listOf(it), pageLength = 1, pageIndex = 0))
|
_filesUpdates.emit(DataPage(listOf(it), pageLength = 1, pageIndex = 0))
|
||||||
}
|
}
|
||||||
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
|
if (recordSyncJournal) {
|
||||||
|
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
|
||||||
|
}
|
||||||
} finally {
|
} finally {
|
||||||
tmp.delete()
|
tmp.delete()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
|
||||||
val cleanedPath = if (path.startsWith("/")) path else "/$path"
|
val cleanedPath = StorageSyncPaths.normalize(path)
|
||||||
val entries = readSyncJournal()
|
if (!StorageSyncPaths.isSyncableUserPath(cleanedPath)) {
|
||||||
val nextSequence = (entries.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
return
|
||||||
val entry = StorageSyncJournalEntry(
|
}
|
||||||
path = cleanedPath,
|
val journal = readSyncJournal()
|
||||||
operation = operation,
|
val nextSequence = (journal.values.maxOfOrNull { it.revision.sequence } ?: 0L) + 1L
|
||||||
revision = StorageSyncRevision(
|
putSyncJournalEntries(
|
||||||
sequence = nextSequence,
|
mapOf(
|
||||||
actorId = syncActorId,
|
cleanedPath to StorageSyncJournalEntry(
|
||||||
createdAt = Instant.now(),
|
path = cleanedPath,
|
||||||
|
operation = operation,
|
||||||
|
revision = StorageSyncRevision(
|
||||||
|
sequence = nextSequence,
|
||||||
|
actorId = syncActorId,
|
||||||
|
createdAt = Instant.now(),
|
||||||
|
),
|
||||||
|
originStorageUuid = storageUuid,
|
||||||
|
),
|
||||||
),
|
),
|
||||||
originStorageUuid = storageUuid,
|
|
||||||
)
|
)
|
||||||
appendSyncJournal(listOf(entry))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun touchParentDirs(path: String) {
|
private suspend fun touchParentDirs(path: String) {
|
||||||
|
|||||||
@@ -0,0 +1,59 @@
|
|||||||
|
package com.github.nullptroma.wallenc.domain.datatypes
|
||||||
|
|
||||||
|
/** Журнал синхронизации: один актуальный [StorageSyncJournalEntry] на нормализованный путь. */
|
||||||
|
typealias StorageSyncJournal = Map<String, StorageSyncJournalEntry>
|
||||||
|
|
||||||
|
object StorageSyncPaths {
|
||||||
|
fun normalize(path: String): String = if (path.startsWith("/")) path else "/$path"
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Пути, которые участвуют в sync и попадают в журнал при пользовательских операциях.
|
||||||
|
* Служебные каталоги, lock/journal/meta и файлы внутри *-enc-dir исключены.
|
||||||
|
*/
|
||||||
|
fun isSyncableUserPath(path: String): Boolean {
|
||||||
|
val p = normalize(path)
|
||||||
|
if (p == "/" || p.isBlank()) return false
|
||||||
|
if (p == "/wallenc-yandex-system" || p.startsWith("/wallenc-yandex-system/")) return false
|
||||||
|
if (p.contains("-enc-dir/") || p.endsWith("-enc-dir")) return false
|
||||||
|
val name = p.substringAfterLast('/')
|
||||||
|
if (name == "sync-journal.json" || name == "sync-lock.json") return false
|
||||||
|
if (name.endsWith(".enc-meta") || name.endsWith(".storage-info")) return false
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
object StorageSyncJournalMerge {
|
||||||
|
fun merge(into: StorageSyncJournal, entries: Map<String, StorageSyncJournalEntry>): StorageSyncJournal {
|
||||||
|
if (entries.isEmpty()) return into
|
||||||
|
val result = into.toMutableMap()
|
||||||
|
for ((rawPath, entry) in entries) {
|
||||||
|
val path = StorageSyncPaths.normalize(rawPath)
|
||||||
|
if (!StorageSyncPaths.isSyncableUserPath(path)) continue
|
||||||
|
val normalizedEntry = entry.copy(path = path)
|
||||||
|
val current = result[path]
|
||||||
|
if (current == null || compareEntries(normalizedEntry, current) > 0) {
|
||||||
|
result[path] = normalizedEntry
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
fun merge(into: StorageSyncJournal, entry: StorageSyncJournalEntry): StorageSyncJournal =
|
||||||
|
merge(into, mapOf(entry.path to entry))
|
||||||
|
|
||||||
|
fun mergeAll(journals: Collection<StorageSyncJournal>): StorageSyncJournal {
|
||||||
|
var acc: StorageSyncJournal = emptyMap()
|
||||||
|
for (journal in journals) {
|
||||||
|
acc = merge(acc, journal)
|
||||||
|
}
|
||||||
|
return acc
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun compareEntries(a: StorageSyncJournalEntry, b: StorageSyncJournalEntry): Int {
|
||||||
|
val seqCmp = a.revision.sequence.compareTo(b.revision.sequence)
|
||||||
|
if (seqCmp != 0) return seqCmp
|
||||||
|
val actorCmp = a.revision.actorId.compareTo(b.revision.actorId)
|
||||||
|
if (actorCmp != 0) return actorCmp
|
||||||
|
return a.revision.createdAt.compareTo(b.revision.createdAt)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
package com.github.nullptroma.wallenc.domain.interfaces
|
package com.github.nullptroma.wallenc.domain.interfaces
|
||||||
|
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.DataPage
|
import com.github.nullptroma.wallenc.domain.datatypes.DataPage
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
||||||
import kotlinx.coroutines.flow.Flow
|
import kotlinx.coroutines.flow.Flow
|
||||||
import kotlinx.coroutines.flow.SharedFlow
|
import kotlinx.coroutines.flow.SharedFlow
|
||||||
@@ -40,7 +40,7 @@ interface IStorageAccessor {
|
|||||||
suspend fun touchFile(path: String)
|
suspend fun touchFile(path: String)
|
||||||
suspend fun touchDir(path: String)
|
suspend fun touchDir(path: String)
|
||||||
suspend fun delete(path: String)
|
suspend fun delete(path: String)
|
||||||
suspend fun openWrite(path: String): OutputStream
|
suspend fun openWrite(path: String, recordSyncJournal: Boolean = true): OutputStream
|
||||||
suspend fun openRead(path: String): InputStream
|
suspend fun openRead(path: String): InputStream
|
||||||
suspend fun moveToTrash(path: String)
|
suspend fun moveToTrash(path: String)
|
||||||
|
|
||||||
@@ -52,9 +52,8 @@ interface IStorageAccessor {
|
|||||||
suspend fun openReadSystemFile(name: String): InputStream
|
suspend fun openReadSystemFile(name: String): InputStream
|
||||||
suspend fun openWriteSystemFile(name: String): OutputStream
|
suspend fun openWriteSystemFile(name: String): OutputStream
|
||||||
|
|
||||||
suspend fun readSyncJournal(): List<StorageSyncJournalEntry>
|
suspend fun readSyncJournal(): StorageSyncJournal
|
||||||
suspend fun appendSyncJournal(entries: List<StorageSyncJournalEntry>)
|
suspend fun putSyncJournalEntries(entries: StorageSyncJournal)
|
||||||
suspend fun rewriteSyncJournal(entries: List<StorageSyncJournalEntry>)
|
|
||||||
|
|
||||||
suspend fun readSyncLock(): StorageSyncLock?
|
suspend fun readSyncLock(): StorageSyncLock?
|
||||||
suspend fun tryAcquireSyncLock(holderId: String, leaseUntil: Instant): Boolean
|
suspend fun tryAcquireSyncLock(holderId: String, leaseUntil: Instant): Boolean
|
||||||
|
|||||||
@@ -1,7 +1,10 @@
|
|||||||
package com.github.nullptroma.wallenc.usecases
|
package com.github.nullptroma.wallenc.usecases
|
||||||
|
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorage
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorage
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageSyncEngine
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageSyncEngine
|
||||||
@@ -113,8 +116,8 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
lockedAccessors.add(storage.accessor)
|
lockedAccessors.add(storage.accessor)
|
||||||
}
|
}
|
||||||
|
|
||||||
val latestByPath = mutableMapOf<String, StorageSyncJournalEntry>()
|
val mergedByPath = mutableMapOf<String, StorageSyncJournalEntry>()
|
||||||
val entriesByStorage = mutableMapOf<UUID, Map<String, StorageSyncJournalEntry>>()
|
val entriesByStorage = mutableMapOf<UUID, StorageSyncJournal>()
|
||||||
|
|
||||||
reportProgress(null, TaskProgressLabel.SyncGroupReadingJournals(groupId))
|
reportProgress(null, TaskProgressLabel.SyncGroupReadingJournals(groupId))
|
||||||
for ((journalIndex, storage) in storages.withIndex()) {
|
for ((journalIndex, storage) in storages.withIndex()) {
|
||||||
@@ -132,17 +135,14 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
null,
|
null,
|
||||||
TaskProgressLabel.SyncGroupJournalProgress(groupId, journalIndex + 1, storages.size),
|
TaskProgressLabel.SyncGroupJournalProgress(groupId, journalIndex + 1, storages.size),
|
||||||
)
|
)
|
||||||
val latestEntries = latestByPath(storage.accessor.readSyncJournal())
|
val journal = filterSyncableJournal(storage.accessor.readSyncJournal())
|
||||||
entriesByStorage[storage.uuid] = latestEntries
|
entriesByStorage[storage.uuid] = journal
|
||||||
for ((path, entry) in latestEntries) {
|
mergedByPath.putAll(
|
||||||
val current = latestByPath[path]
|
StorageSyncJournalMerge.merge(mergedByPath, journal),
|
||||||
if (current == null || compareEntries(entry, current) > 0) {
|
)
|
||||||
latestByPath[path] = entry
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
val mergedEntries = latestByPath.entries.toList()
|
val mergedEntries = mergedByPath.entries.toList()
|
||||||
if (mergedEntries.isEmpty()) {
|
if (mergedEntries.isEmpty()) {
|
||||||
reportProgress(null, TaskProgressLabel.SyncGroupNoJournalEntries(groupId))
|
reportProgress(null, TaskProgressLabel.SyncGroupNoJournalEntries(groupId))
|
||||||
return
|
return
|
||||||
@@ -191,13 +191,12 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
entry = winnerEntry,
|
entry = winnerEntry,
|
||||||
)
|
)
|
||||||
if (applied) {
|
if (applied) {
|
||||||
target.accessor.appendSyncJournal(listOf(winnerEntry))
|
target.accessor.putSyncJournalEntries(mapOf(path to winnerEntry))
|
||||||
} else {
|
} else {
|
||||||
applyFailures++
|
applyFailures++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
compactSyncJournals(storages)
|
|
||||||
if (applyFailures > 0) {
|
if (applyFailures > 0) {
|
||||||
reportProgress(
|
reportProgress(
|
||||||
null,
|
null,
|
||||||
@@ -215,6 +214,10 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun filterSyncableJournal(journal: StorageSyncJournal): StorageSyncJournal {
|
||||||
|
return journal.filterKeys { StorageSyncPaths.isSyncableUserPath(it) }
|
||||||
|
}
|
||||||
|
|
||||||
private suspend fun renewLocksIfNeeded(
|
private suspend fun renewLocksIfNeeded(
|
||||||
groupId: String,
|
groupId: String,
|
||||||
lockedAccessors: List<IStorageAccessor>,
|
lockedAccessors: List<IStorageAccessor>,
|
||||||
@@ -243,20 +246,9 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
return uuids.mapNotNull(findStorageUseCase::find)
|
return uuids.mapNotNull(findStorageUseCase::find)
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun latestByPath(entries: List<StorageSyncJournalEntry>): Map<String, StorageSyncJournalEntry> {
|
|
||||||
val map = mutableMapOf<String, StorageSyncJournalEntry>()
|
|
||||||
for (entry in entries) {
|
|
||||||
val current = map[entry.path]
|
|
||||||
if (current == null || compareEntries(entry, current) > 0) {
|
|
||||||
map[entry.path] = entry
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return map
|
|
||||||
}
|
|
||||||
|
|
||||||
private fun findSourceStorage(
|
private fun findSourceStorage(
|
||||||
storages: List<IStorage>,
|
storages: List<IStorage>,
|
||||||
entriesByStorage: Map<UUID, Map<String, StorageSyncJournalEntry>>,
|
entriesByStorage: Map<UUID, StorageSyncJournal>,
|
||||||
path: String,
|
path: String,
|
||||||
winnerEntry: StorageSyncJournalEntry,
|
winnerEntry: StorageSyncJournalEntry,
|
||||||
): IStorage? {
|
): IStorage? {
|
||||||
@@ -273,16 +265,6 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun compactSyncJournals(storages: List<IStorage>) {
|
|
||||||
for (storage in storages) {
|
|
||||||
val entries = storage.accessor.readSyncJournal()
|
|
||||||
val compacted = latestByPath(entries).values.toList()
|
|
||||||
if (compacted.size < entries.size) {
|
|
||||||
storage.accessor.rewriteSyncJournal(compacted)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private suspend fun applyEntry(
|
private suspend fun applyEntry(
|
||||||
source: IStorage?,
|
source: IStorage?,
|
||||||
target: IStorage,
|
target: IStorage,
|
||||||
@@ -305,7 +287,7 @@ class StorageSyncEngine @Inject constructor(
|
|||||||
val sourceAccessor = source?.accessor ?: return false
|
val sourceAccessor = source?.accessor ?: return false
|
||||||
runCatching {
|
runCatching {
|
||||||
sourceAccessor.openRead(entry.path).use { input ->
|
sourceAccessor.openRead(entry.path).use { input ->
|
||||||
target.accessor.openWrite(entry.path).use { output ->
|
target.accessor.openWrite(entry.path, recordSyncJournal = false).use { output ->
|
||||||
input.copyTo(output)
|
input.copyTo(output)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
package com.github.nullptroma.wallenc.usecases
|
||||||
|
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncOperation
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncPaths
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncRevision
|
||||||
|
import org.junit.Assert.assertEquals
|
||||||
|
import org.junit.Assert.assertFalse
|
||||||
|
import org.junit.Assert.assertTrue
|
||||||
|
import org.junit.Test
|
||||||
|
import java.time.Instant
|
||||||
|
|
||||||
|
class StorageSyncJournalMergeTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun mergeKeepsSingleEntryPerPath() {
|
||||||
|
val older = entry(path = "/a.txt", sequence = 1L)
|
||||||
|
val newer = entry(path = "/a.txt", sequence = 2L)
|
||||||
|
val merged = StorageSyncJournalMerge.merge(
|
||||||
|
mapOf("/a.txt" to older),
|
||||||
|
mapOf("/a.txt" to newer),
|
||||||
|
)
|
||||||
|
assertEquals(1, merged.size)
|
||||||
|
assertEquals(2L, merged["/a.txt"]?.revision?.sequence)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun isSyncableUserPathExcludesEncDirAndJournal() {
|
||||||
|
assertFalse(StorageSyncPaths.isSyncableUserPath("/88a048ed-enc-dir/sync-journal.json"))
|
||||||
|
assertFalse(StorageSyncPaths.isSyncableUserPath("/wallenc-yandex-system/sync-journal.json"))
|
||||||
|
assertTrue(StorageSyncPaths.isSyncableUserPath("/wallenc-data/text-secrets.json"))
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun entry(path: String, sequence: Long): StorageSyncJournalEntry =
|
||||||
|
StorageSyncJournalEntry(
|
||||||
|
path = path,
|
||||||
|
operation = StorageSyncOperation.UPSERT,
|
||||||
|
revision = StorageSyncRevision(
|
||||||
|
sequence = sequence,
|
||||||
|
actorId = "actor",
|
||||||
|
createdAt = Instant.EPOCH,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -3,6 +3,7 @@ package com.github.nullptroma.wallenc.usecases.fakes
|
|||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageEncryptionInfo
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageEncryptionInfo
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageMetaLoadState
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageMetaLoadState
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorage
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorage
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageAccessor
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IStorageMetaInfo
|
import com.github.nullptroma.wallenc.domain.interfaces.IStorageMetaInfo
|
||||||
@@ -46,10 +47,13 @@ class FakeStorage(
|
|||||||
fun fileBytes(path: String): ByteArray? = accessorImpl.dataFiles[path]
|
fun fileBytes(path: String): ByteArray? = accessorImpl.dataFiles[path]
|
||||||
|
|
||||||
fun addSyncJournalEntry(entry: StorageSyncJournalEntry) {
|
fun addSyncJournalEntry(entry: StorageSyncJournalEntry) {
|
||||||
accessorImpl.syncJournal.add(entry)
|
accessorImpl.syncJournal = StorageSyncJournalMerge.merge(
|
||||||
|
accessorImpl.syncJournal,
|
||||||
|
mapOf(entry.path to entry),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
fun syncJournalEntries(): List<StorageSyncJournalEntry> = accessorImpl.syncJournal.toList()
|
fun syncJournalEntries(): List<StorageSyncJournalEntry> = accessorImpl.syncJournal.values.toList()
|
||||||
|
|
||||||
fun setAcquireLockResult(result: Boolean) {
|
fun setAcquireLockResult(result: Boolean) {
|
||||||
accessorImpl.acquireLockResult = result
|
accessorImpl.acquireLockResult = result
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package com.github.nullptroma.wallenc.usecases.fakes
|
package com.github.nullptroma.wallenc.usecases.fakes
|
||||||
|
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.DataPage
|
import com.github.nullptroma.wallenc.domain.datatypes.DataPage
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournal
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalEntry
|
||||||
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncJournalMerge
|
||||||
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
import com.github.nullptroma.wallenc.domain.datatypes.StorageSyncLock
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
import com.github.nullptroma.wallenc.domain.interfaces.IDirectory
|
||||||
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
import com.github.nullptroma.wallenc.domain.interfaces.IFile
|
||||||
@@ -25,7 +27,7 @@ class FakeStorageAccessor : IStorageAccessor {
|
|||||||
private val systemFiles: MutableMap<String, ByteArray> = mutableMapOf()
|
private val systemFiles: MutableMap<String, ByteArray> = mutableMapOf()
|
||||||
private val _filesUpdates = MutableSharedFlow<DataPage<IFile>>(extraBufferCapacity = 16)
|
private val _filesUpdates = MutableSharedFlow<DataPage<IFile>>(extraBufferCapacity = 16)
|
||||||
|
|
||||||
var syncJournal: MutableList<StorageSyncJournalEntry> = mutableListOf()
|
var syncJournal: StorageSyncJournal = emptyMap()
|
||||||
var syncLock: StorageSyncLock? = null
|
var syncLock: StorageSyncLock? = null
|
||||||
var acquireLockResult: Boolean = true
|
var acquireLockResult: Boolean = true
|
||||||
|
|
||||||
@@ -65,7 +67,7 @@ class FakeStorageAccessor : IStorageAccessor {
|
|||||||
dataFiles.remove(path)
|
dataFiles.remove(path)
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun openWrite(path: String): OutputStream {
|
override suspend fun openWrite(path: String, recordSyncJournal: Boolean): OutputStream {
|
||||||
return object : ByteArrayOutputStream() {
|
return object : ByteArrayOutputStream() {
|
||||||
override fun close() {
|
override fun close() {
|
||||||
dataFiles[path] = toByteArray()
|
dataFiles[path] = toByteArray()
|
||||||
@@ -104,15 +106,10 @@ class FakeStorageAccessor : IStorageAccessor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun readSyncJournal(): List<StorageSyncJournalEntry> = syncJournal.toList()
|
override suspend fun readSyncJournal(): StorageSyncJournal = syncJournal
|
||||||
|
|
||||||
override suspend fun appendSyncJournal(entries: List<StorageSyncJournalEntry>) {
|
override suspend fun putSyncJournalEntries(entries: StorageSyncJournal) {
|
||||||
syncJournal.addAll(entries)
|
syncJournal = StorageSyncJournalMerge.merge(syncJournal, entries)
|
||||||
}
|
|
||||||
|
|
||||||
override suspend fun rewriteSyncJournal(entries: List<StorageSyncJournalEntry>) {
|
|
||||||
syncJournal.clear()
|
|
||||||
syncJournal.addAll(entries)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun readSyncLock(): StorageSyncLock? = syncLock
|
override suspend fun readSyncLock(): StorageSyncLock? = syncLock
|
||||||
|
|||||||
Reference in New Issue
Block a user