From 51e7f821e97c266df1895c6d0693800300751e58 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sat, 10 Oct 2026 01:15:32 +0200 Subject: [PATCH] Own Android file transfers and core activity independently of UI --- .../java/com/utun/chat/data/CallController.kt | 8 +- .../java/com/utun/chat/data/CoreActivity.kt | 24 ++ .../utun/chat/data/MemberFilesRepository.kt | 313 ++++++++++++++++++ 3 files changed, 341 insertions(+), 4 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/CoreActivity.kt create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/MemberFilesRepository.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallController.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallController.kt index 24398ef5..6c8c32a0 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallController.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallController.kt @@ -71,6 +71,7 @@ object CallController : TelecomCallManager.Listener { fun setActivityVisible(visible: Boolean) { checkMain() activityVisible = visible + CoreActivity.visible(visible) syncCoreActivity() if (visible && answerWhenVisible != 0L) { val id = answerWhenVisible @@ -80,10 +81,7 @@ object CallController : TelecomCallManager.Listener { } private fun syncCoreActivity() { - val ongoing = _state.value?.ongoing == true - val active = activityVisible || ongoing - LogManager.addLog("INFO", "Call", "core activity=$active visible=$activityVisible call=$ongoing") - if (coreReady && NativeLib.isReady()) NativeLib.setActive(active) + CoreActivity.call(_state.value?.ongoing == true) } fun start(channelId: String, peerNodeId: Long, peerName: String) { @@ -231,8 +229,10 @@ object CallController : TelecomCallManager.Listener { fun reset() { checkMain() coreReady = false + CoreActivity.reset() end() _state.value = null + CoreActivity.call(false) } fun toggleMute() { diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CoreActivity.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CoreActivity.kt new file mode 100644 index 00000000..c4f077d7 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CoreActivity.kt @@ -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) + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MemberFilesRepository.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MemberFilesRepository.kt new file mode 100644 index 00000000..40e15f4a --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/MemberFilesRepository.kt @@ -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) +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) + private data class Transfer(val location: FileLocation, val result: CompletableDeferred) + private val pending = mutableMapOf() + private val transfers = mutableMapOf() + private var task: Job? = null + private var retry: (() -> Unit)? = null + private var wakeLock: PowerManager.WakeLock? = null + private val _operation = MutableStateFlow(null) + val operation: StateFlow = _operation + private val _conflict = MutableStateFlow(null) + val conflict: StateFlow = _conflict + private val _export = MutableStateFlow(null) + val export: StateFlow = _export + private val _revision = MutableStateFlow(0L) + val revision: StateFlow = _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() + 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() + 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) { + 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() + _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") + } + } +}