evgeny
2 days ago
3 changed files with 341 additions and 4 deletions
@ -0,0 +1,24 @@
|
||||
package com.utun.chat.data |
||||
|
||||
import android.os.Looper |
||||
|
||||
/** Один расчёт активности ядра: завершение одного владельца не усыпляет остальных. */ |
||||
object CoreActivity { |
||||
private var visible = false |
||||
private var call = false |
||||
private var transfer = false |
||||
private var applied: Boolean? = null |
||||
fun visible(value: Boolean) { visible = value; sync() } |
||||
fun call(value: Boolean) { call = value; sync() } |
||||
fun transfer(value: Boolean) { transfer = value; sync() } |
||||
fun reset() { applied = null } |
||||
fun sync() { |
||||
check(Looper.myLooper() == Looper.getMainLooper()) |
||||
if (!NativeLib.isReady()) { applied = null; return } |
||||
val active = visible || call || transfer |
||||
if (applied == active) return |
||||
applied = active |
||||
LogManager.addLog("INFO", "Activity", "core active=$active visible=$visible call=$call transfer=$transfer") |
||||
NativeLib.setActive(active) |
||||
} |
||||
} |
||||
@ -0,0 +1,313 @@
|
||||
package com.utun.chat.data |
||||
|
||||
import android.content.Context |
||||
import android.net.Uri |
||||
import android.os.Looper |
||||
import android.os.PowerManager |
||||
import android.provider.OpenableColumns |
||||
import android.provider.DocumentsContract |
||||
import com.utun.chat.R |
||||
import com.utun.chat.headless.HeadlessService |
||||
import kotlinx.coroutines.* |
||||
import kotlinx.coroutines.flow.MutableStateFlow |
||||
import kotlinx.coroutines.flow.StateFlow |
||||
import java.io.File |
||||
import java.io.IOException |
||||
|
||||
data class FileLocation(val group: Long, val server: Long, val owner: Long, val path: String) |
||||
data class FileOperation(val token: Long, val location: FileLocation, val name: String = "", val index: Int = 1, |
||||
val count: Int = 1, val done: Long = 0, val total: Long = 0, val stage: Int = R.string.files_preparing, |
||||
val status: Int? = null, val active: Boolean = true, val cancelled: Boolean = false, val retryable: Boolean = true, |
||||
val upload: Boolean = false, val uploaded: Int = 0, val skipped: Int = 0) |
||||
data class FileConflict(val token: Long, val name: String, val decision: CompletableDeferred<Boolean?>) |
||||
data class FileExport(val token: Long, val location: FileLocation, val name: String, val file: File, val action: Int) |
||||
class FileStatusException(val status: Int) : IOException("member_files status=$status") |
||||
|
||||
/** Владеет передачами независимо от экранов. Все состояния/ожидания изменяются только на main. */ |
||||
class MemberFilesRepository(context: Context) { |
||||
companion object { const val OPEN = 0; const val SAVE = 1; const val SHARE = 2 } |
||||
private val app = context.applicationContext |
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate) |
||||
private data class Pending(val location: FileLocation, val command: Int, val result: CompletableDeferred<MemberFilesProtocol.Reply>) |
||||
private data class Transfer(val location: FileLocation, val result: CompletableDeferred<MemberFilesProtocol.Progress>) |
||||
private val pending = mutableMapOf<Long, Pending>() |
||||
private val transfers = mutableMapOf<Long, Transfer>() |
||||
private var task: Job? = null |
||||
private var retry: (() -> Unit)? = null |
||||
private var wakeLock: PowerManager.WakeLock? = null |
||||
private val _operation = MutableStateFlow<FileOperation?>(null) |
||||
val operation: StateFlow<FileOperation?> = _operation |
||||
private val _conflict = MutableStateFlow<FileConflict?>(null) |
||||
val conflict: StateFlow<FileConflict?> = _conflict |
||||
private val _export = MutableStateFlow<FileExport?>(null) |
||||
val export: StateFlow<FileExport?> = _export |
||||
private val _revision = MutableStateFlow(0L) |
||||
val revision: StateFlow<Long> = _revision |
||||
private fun log(level: String, message: String) = LogManager.addLog(level, "Files", message) |
||||
private fun main() = check(Looper.myLooper() == Looper.getMainLooper()) |
||||
|
||||
init { |
||||
// URI для внешнего приложения сохраняется сутки; старые экспорты и брошенные источники очищаются при запуске. |
||||
scope.launch(Dispatchers.IO) { |
||||
try { |
||||
val root = File(app.cacheDir, "member_files") |
||||
val cutoff = System.currentTimeMillis() - 24 * 60 * 60 * 1000 |
||||
root.listFiles()?.filter { it.lastModified() < cutoff }?.forEach { file -> |
||||
if (file.isDirectory) file.listFiles()?.forEach { remove(it) } |
||||
remove(file) |
||||
} |
||||
} catch (e: Exception) { log("WARN", "Cache cleanup failed: $e") } |
||||
} |
||||
} |
||||
|
||||
private fun protect(active: Boolean) { |
||||
if (active && wakeLock == null) { |
||||
val manager = app.getSystemService(Context.POWER_SERVICE) as PowerManager |
||||
wakeLock = manager.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:member_files").apply { acquire() } |
||||
} else if (!active) { |
||||
wakeLock?.let { if (it.isHeld) it.release() }; wakeLock = null |
||||
} |
||||
CoreActivity.transfer(active) |
||||
} |
||||
|
||||
fun onEvent(type: Int, bytes: ByteArray?) { |
||||
main() |
||||
if (type == 13) { coreStopped(); return } |
||||
if (type !in 49..50 || bytes == null) return |
||||
try { |
||||
if (type == 49) { |
||||
val r = MemberFilesProtocol.response(bytes); val p = pending[r.id] ?: return |
||||
require(r.group == p.location.group && r.server == p.location.server && r.owner == p.location.owner && |
||||
r.command == p.command && r.path == p.location.path) { "request context mismatch id=${r.id}" } |
||||
log("DEBUG", "response id=${r.id} command=${r.command} location=${p.location} status=${r.status}") |
||||
p.result.complete(r) |
||||
} else { |
||||
val r = MemberFilesProtocol.progress(bytes); val p = transfers[r.id] ?: return |
||||
require(r.group == p.location.group && r.server == p.location.server && r.owner == p.location.owner) { |
||||
"transfer context mismatch id=${r.id}" |
||||
} |
||||
if (r.finished) { |
||||
log("INFO", "transfer finished id=${r.id} location=${p.location} status=${r.status} bytes=${r.done}/${r.total}") |
||||
p.result.complete(r) |
||||
} else _operation.value?.let { publish(it.copy(done = r.done, total = r.total, stage = R.string.files_transferring)) } |
||||
} |
||||
} catch (e: Exception) { log("ERROR", "Rejected event type=$type bytes=${bytes.size}: $e") } |
||||
} |
||||
|
||||
suspend fun request(location: FileLocation, command: Int, target: String = "", cursor: Long = 0, |
||||
version: ByteArray = ByteArray(16)): MemberFilesProtocol.Reply { |
||||
main() |
||||
val id = MemberFilesProtocol.nextId(); val result = CompletableDeferred<MemberFilesProtocol.Reply>() |
||||
pending[id] = Pending(location, command, result) |
||||
var finished = false |
||||
try { |
||||
log("DEBUG", "request id=$id command=$command location=$location cursor=$cursor") |
||||
val wire = MemberFilesProtocol.request(id, location.owner, command, location.path, target, cursor, version) |
||||
if (!NativeLib.fileRequest(location.group, location.server, wire)) throw IOException("request enqueue failed id=$id") |
||||
return withTimeout(60_000) { result.await() }.also { finished = true } |
||||
} catch (e: TimeoutCancellationException) { |
||||
log("WARN", "request timed out id=$id command=$command location=$location") |
||||
throw FileStatusException(13) |
||||
} finally { |
||||
pending.remove(id) |
||||
if (!finished && NativeLib.isReady()) NativeLib.fileCancel(id) |
||||
} |
||||
} |
||||
|
||||
fun check(reply: MemberFilesProtocol.Reply): MemberFilesProtocol.Reply { |
||||
if (reply.status != 0) throw FileStatusException(reply.status) |
||||
return reply |
||||
} |
||||
|
||||
private suspend fun transfer(location: FileLocation, local: File, upload: Boolean, version: ByteArray) { |
||||
val id = MemberFilesProtocol.nextId(); val result = CompletableDeferred<MemberFilesProtocol.Progress>() |
||||
transfers[id] = Transfer(location, result) |
||||
var queued = false |
||||
try { |
||||
queued = NativeLib.fileTransfer(id, location.group, location.server, location.owner, location.path, local.path, upload, version) |
||||
if (!queued) throw IOException("transfer enqueue failed id=$id") |
||||
val r = result.await() |
||||
if (r.status != 0) throw FileStatusException(r.status) |
||||
} finally { |
||||
// Временный источник/приёмник освобождаем только после закрытия native handle. |
||||
if (queued && !result.isCompleted) { |
||||
if (NativeLib.isReady()) NativeLib.fileCancel(id) |
||||
withContext(NonCancellable) { runCatching { result.await() } } |
||||
} |
||||
transfers.remove(id) |
||||
} |
||||
} |
||||
|
||||
private fun publish(operation: FileOperation) { |
||||
val previous = _operation.value |
||||
_operation.value = operation |
||||
// Уведомление обновляется по процентам; UI получает каждый прогресс без спама в логах. |
||||
fun percent(op: FileOperation?) = if (op == null || op.total == 0L) -1 else (op.done.toDouble() * 100 / op.total).toInt() |
||||
if (previous?.name != operation.name || previous?.stage != operation.stage || previous?.active != operation.active || |
||||
percent(previous) != percent(operation)) HeadlessService.refreshFiles() |
||||
} |
||||
|
||||
private fun launch(location: FileLocation, count: Int, retryable: Boolean = true, work: suspend (Long) -> Unit) { |
||||
main() |
||||
if (task?.isCompleted == false || _export.value != null) { |
||||
log("WARN", "operation rejected while busy location=$location"); return |
||||
} |
||||
val token = MemberFilesProtocol.nextId() |
||||
retry = if (retryable) ({ launch(location, count, work = work) }) else null |
||||
publish(FileOperation(token, location, count = count, retryable = retryable)) |
||||
task = scope.launch { |
||||
try { |
||||
protect(true) |
||||
log("INFO", "operation started token=$token location=$location count=$count") |
||||
work(token) |
||||
publish(requireNotNull(_operation.value).copy(active = false, stage = R.string.files_completed)) |
||||
_revision.value++ |
||||
} catch (e: CancellationException) { |
||||
log("INFO", "operation cancelled token=$token location=$location") |
||||
_operation.value?.let { publish(it.copy(active = false, cancelled = true, stage = R.string.files_cancelled)) } |
||||
_revision.value++ |
||||
} catch (e: Exception) { |
||||
log("ERROR", "operation failed token=$token location=$location: $e") |
||||
_operation.value?.let { publish(it.copy(active = false, status = (e as? FileStatusException)?.status ?: |
||||
if (e is IllegalArgumentException) 3 else 9, |
||||
stage = R.string.files_failed)) } |
||||
_revision.value++ |
||||
} finally { |
||||
_conflict.value = null |
||||
protect(false) |
||||
log("DEBUG", "operation released token=$token wake=false") |
||||
} |
||||
} |
||||
} |
||||
|
||||
fun upload(location: FileLocation, uris: List<Uri>) { |
||||
if (uris.isEmpty()) return |
||||
val sources = uris.distinct() |
||||
launch(location, sources.size) { token -> |
||||
require(location.owner == NativeLib.getMyNodeId() && !MemberFilesProtocol.servicePath(location.path)) |
||||
var uploaded = 0 |
||||
var skipped = 0 |
||||
for ((index, uri) in sources.withIndex()) { |
||||
retry = { upload(location, sources.drop(index)) } |
||||
protect(true) |
||||
var source: File? = null |
||||
try { |
||||
val name = withContext(Dispatchers.IO) { |
||||
app.contentResolver.query(uri, arrayOf(OpenableColumns.DISPLAY_NAME), null, null, null)?.use { |
||||
if (it.moveToFirst()) it.getString(0) else null |
||||
} ?: throw IOException("document has no display name uri=$uri") |
||||
} |
||||
val remote = location.copy(path = MemberFilesProtocol.child(location.path, name)) |
||||
require(!MemberFilesProtocol.servicePath(remote.path)) { "reserved avatar directory" } |
||||
publish(FileOperation(token, location, name, index + 1, sources.size, upload = true, uploaded = uploaded, skipped = skipped)) |
||||
val stat = request(remote, MemberFilesProtocol.STAT) |
||||
var version = ByteArray(16) |
||||
if (stat.status == 0) { |
||||
if (stat.type != MemberFilesProtocol.FILE) throw FileStatusException(5) |
||||
val answer = CompletableDeferred<Boolean?>() |
||||
_conflict.value = FileConflict(token, name, answer) |
||||
publish(requireNotNull(_operation.value).copy(stage = R.string.files_replace_title)) |
||||
protect(false) |
||||
val replace = answer.await() |
||||
currentCoroutineContext().ensureActive() |
||||
when (replace) { |
||||
null -> throw CancellationException("queue cancelled") |
||||
false -> { |
||||
skipped++; publish(requireNotNull(_operation.value).copy(skipped = skipped)) |
||||
retry = { upload(location, sources.drop(index + 1)) } |
||||
log("INFO", "upload skipped token=$token path=${remote.path}"); continue |
||||
} |
||||
true -> version = stat.version |
||||
} |
||||
protect(true) |
||||
_conflict.value = null |
||||
} else if (stat.status != MemberFilesProtocol.NOT_FOUND) throw FileStatusException(stat.status) |
||||
// Владение фиксируется внутри IO до возврата на main, в том числе при отмене. |
||||
withContext(Dispatchers.IO) { |
||||
val file = File.createTempFile("upload-", ".tmp", exportDirectory()); source = file |
||||
app.contentResolver.openInputStream(uri)?.use { input -> file.outputStream().use { output -> |
||||
val bytes = ByteArray(64 * 1024) |
||||
while (true) { ensureActive(); val size = input.read(bytes); if (size < 0) break; output.write(bytes, 0, size) } |
||||
} } ?: throw IOException("document cannot be read uri=$uri") |
||||
} |
||||
transfer(remote, requireNotNull(source), true, version) |
||||
uploaded++; publish(requireNotNull(_operation.value).copy(uploaded = uploaded)) |
||||
retry = { upload(location, sources.drop(index + 1)) } |
||||
_revision.value++ |
||||
} finally { |
||||
_conflict.value = null |
||||
withContext(NonCancellable + Dispatchers.IO) { source?.let { remove(it) } } |
||||
} |
||||
} |
||||
} |
||||
} |
||||
|
||||
private fun exportDirectory(): File { |
||||
val directory = File(app.cacheDir, "member_files") |
||||
if (!directory.isDirectory && !directory.mkdirs()) throw IOException("cannot create file cache") |
||||
return directory |
||||
} |
||||
private fun remove(file: File) { if (file.exists() && !file.delete()) log("WARN", "Cannot remove temporary file=$file") } |
||||
|
||||
fun download(location: FileLocation, entry: MemberFilesProtocol.Entry, action: Int) { |
||||
launch(location, 1) { token -> |
||||
publish(FileOperation(token, location, entry.name)) |
||||
val remote = location.copy(path = MemberFilesProtocol.child(location.path, entry.name)) |
||||
var directory: File? = null |
||||
var file: File? = null |
||||
var retained = false |
||||
try { |
||||
withContext(Dispatchers.IO) { |
||||
val dir = File(exportDirectory(), token.toString()); directory = dir |
||||
if (!dir.mkdir()) throw IOException("cannot create export directory") |
||||
file = File(dir, entry.name) |
||||
} |
||||
transfer(remote, requireNotNull(file), false, entry.version) |
||||
_export.value = FileExport(token, location, entry.name, requireNotNull(file), action); retained = true |
||||
} finally { |
||||
if (!retained) withContext(NonCancellable + Dispatchers.IO) { file?.let { remove(it) }; directory?.let { remove(it) } } |
||||
} |
||||
} |
||||
} |
||||
|
||||
fun consumeExport(token: Long) { if (_export.value?.token == token) _export.value = null } |
||||
fun decideReplace(token: Long, replace: Boolean?) { |
||||
_conflict.value?.takeIf { it.token == token }?.decision?.complete(replace) |
||||
} |
||||
fun cancel() { |
||||
main() |
||||
log("INFO", "cancel requested token=${_operation.value?.token}") |
||||
task?.cancel() |
||||
} |
||||
fun retry() { if (_operation.value?.active == false) retry?.invoke() } |
||||
fun coreStopped() { |
||||
main(); cancel() |
||||
val error = CancellationException("core stopped") |
||||
pending.values.toList().forEach { it.result.completeExceptionally(error) } |
||||
transfers.values.toList().forEach { it.result.completeExceptionally(error) } |
||||
CoreActivity.reset() |
||||
} |
||||
fun dismiss() { if (_operation.value?.active == false) _operation.value = null; HeadlessService.refreshFiles() } |
||||
|
||||
fun save(export: FileExport, uri: Uri) { |
||||
launch(export.location, 1, retryable = false) { token -> |
||||
publish(FileOperation(token, export.location, export.name, stage = R.string.files_saving, retryable = false)) |
||||
try { |
||||
withContext(Dispatchers.IO) { |
||||
app.contentResolver.openOutputStream(uri, "w")?.use { output -> export.file.inputStream().use { input -> |
||||
val bytes = ByteArray(64 * 1024) |
||||
while (true) { ensureActive(); val n = input.read(bytes); if (n < 0) break; output.write(bytes, 0, n) } |
||||
} } ?: throw IOException("cannot open output document uri=$uri") |
||||
} |
||||
} catch (e: Exception) { |
||||
withContext(NonCancellable + Dispatchers.IO) { |
||||
try { |
||||
if (!DocumentsContract.deleteDocument(app.contentResolver, uri)) log("WARN", "Cannot delete incomplete document uri=$uri") |
||||
} catch (cleanup: Exception) { log("WARN", "Incomplete document cleanup failed uri=$uri: $cleanup") } |
||||
} |
||||
throw e |
||||
} |
||||
log("INFO", "document saved file=${export.file} uri=$uri") |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue