diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 64eb7e7e1..d1dc9a440 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -352,6 +352,14 @@ android { manifest.srcFile("src/androidFull/AndroidManifest.xml") java.srcDir(fullCommonSourceDir) } + splits { + abi { + isEnable = true + reset() + include("arm64-v8a", "armeabi-v7a", "x86", "x86_64") + isUniversalApk = false + } + } packaging { resources { excludes += "/META-INF/{AL2.0,LGPL2.1}" diff --git a/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so b/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so index 12994d13e..844e339ed 100755 Binary files a/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so and b/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so differ diff --git a/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so b/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so index 6e7bc6bcb..3c9935edd 100755 Binary files a/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so and b/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so differ diff --git a/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so b/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so index 6309221c4..edc2bf081 100755 Binary files a/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so and b/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so differ diff --git a/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so b/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so index 039eba7d5..5fc0b7a13 100755 Binary files a/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so and b/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so differ diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt index 7497c0607..d9e952b9e 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt @@ -7,12 +7,15 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import okhttp3.MediaType.Companion.toMediaType import okhttp3.OkHttpClient @@ -21,10 +24,29 @@ import okhttp3.RequestBody.Companion.toRequestBody import org.json.JSONArray import org.json.JSONObject import java.io.File +import java.net.URLDecoder import java.net.URLEncoder +import java.util.Locale import java.util.concurrent.TimeUnit private const val TAG = "P2pStreamingEngine" +private const val IDLE_TORRENT_TTL_MS = 120_000L +private const val FILE_INDEX_METADATA_TIMEOUT_MS = 15_000L +private const val FILE_INDEX_FAST_VALIDATION_TIMEOUT_MS = 10_000L +private const val FILE_INDEX_POLL_INTERVAL_MS = 250L +private const val STREAMING_CACHE_SIZE_BYTES = 128L * 1024L * 1024L +private const val STREAMING_CONNECTION_LIMIT = 160 +private const val STREAMING_HALF_OPEN_CONNECTION_LIMIT = 120 +private const val STREAMING_TOTAL_HALF_OPEN_CONNECTION_LIMIT = 500 +private const val STREAMING_PEERS_HIGH_WATER = 900 +private const val STREAMING_PEERS_LOW_WATER = 120 +private const val STREAMING_NOMINAL_DIAL_TIMEOUT_MS = 8_000 +private const val STREAMING_MIN_DIAL_TIMEOUT_MS = 1_500 +private const val STREAMING_HANDSHAKE_TIMEOUT_MS = 3_000 +private const val STREAMING_DISCONNECT_TIMEOUT_SECONDS = 120 +private const val STREAMING_READ_AHEAD_PERCENT = 95 +private const val STREAMING_PRELOAD_CACHE_PERCENT = 50 +private const val STREAM_SCREEN_WARMUP_COOLDOWN_MS = 10_000L private val VIDEO_EXTENSIONS = setOf("mkv", "mp4", "avi", "webm", "ts", "m4v", "mov", "wmv", "flv") actual object P2pStreamingEngine { @@ -34,10 +56,13 @@ actual object P2pStreamingEngine { private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) private val lifecycleLock = Any() private var statsJob: Job? = null - private var cleanupJob: Job? = null + private var preloadJob: Job? = null + private var warmupJob: Job? = null + private var warmupCooldownJob: Job? = null private var currentHash: String? = null private var streamGeneration = 0L private var appContext: Context? = null + private val idleDropJobs = mutableMapOf() private val binary = TorrServerBinary() private val api = TorrServerApi(binary) @@ -46,36 +71,134 @@ actual object P2pStreamingEngine { binary.initialize(context.applicationContext) } + actual fun warmup() { + if (appContext == null) return + synchronized(lifecycleLock) { + warmupCooldownJob?.cancel() + warmupCooldownJob = null + if (warmupJob?.isActive == true) return + warmupJob = scope.launch { + try { + binary.start() + if (api.ensureStreamingSettings().changed) { + binary.stop() + binary.start() + api.ensureStreamingSettings() + } + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + Log.w(TAG, "TorrServer warmup failed", e) + } + } + } + } + + actual fun cooldownWarmup() { + synchronized(lifecycleLock) { + if (currentHash != null) { + return + } + warmupCooldownJob?.cancel() + warmupCooldownJob = scope.launch { + delay(STREAM_SCREEN_WARMUP_COOLDOWN_MS) + val warmup = synchronized(lifecycleLock) { + if (currentHash != null) { + warmupCooldownJob = null + return@launch + } + val job = warmupJob + warmupJob = null + warmupCooldownJob = null + job + } + try { + warmup?.join() + } catch (_: CancellationException) { + } + val shouldStop = synchronized(lifecycleLock) { currentHash == null } + if (!shouldStop) { + return@launch + } + try { + binary.stop() + } catch (e: Exception) { + Log.w(TAG, "Failed to stop idle TorrServer", e) + } + } + } + } + actual suspend fun startStream(request: P2pStreamRequest): String = withContext(Dispatchers.IO) { - stopStreamNow(stopBinary = false) - val generation = nextStreamGeneration() + val requestedHash = request.infoHash.trim().takeIf { it.isNotEmpty() } + ?: throw P2pStreamingException("Missing torrent info hash") + val detached = beginStreamGeneration() + val generation = detached.generation + detached.hash?.let(::scheduleIdleDrop) _state.value = P2pStreamingState.Connecting + var attachedHash: String? = null try { + cancelWarmupCooldown() + awaitWarmup() binary.start() ensureCurrentGeneration(generation) + val settingsResult = api.ensureStreamingSettings() + if (settingsResult.changed) { + binary.stop() + binary.start() + api.ensureStreamingSettings() + } + ensureCurrentGeneration(generation) - val magnetLink = buildMagnetUri(request.infoHash, request.trackers) - Log.d(TAG, "Starting stream: $magnetLink") + val magnetLink = buildMagnetUri( + infoHash = requestedHash, + magnetUri = request.magnetUri, + extraTrackers = request.trackers, + ) val hash = api.addTorrent(magnetLink) ?: throw P2pStreamingException("Failed to add torrent") + attachedHash = hash + cancelIdleDrop(hash) if (!attachTorrentIfCurrent(generation, hash)) { - api.dropTorrent(hash) + scheduleIdleDrop(hash) throw CancellationException("P2P stream start was cancelled") } - val resolvedIdx = resolveFileIndex( - hash = hash, - requestedIdx = request.fileIdx, - filename = request.filename, - ) + val requestedName = request.filename?.trim()?.takeIf { it.isNotEmpty() } + val useEngineFileSelector = requestedName != null || request.fileIdx != null + val resolvedIdx = if (useEngineFileSelector) { + null + } else { + resolveFileIndex( + hash = hash, + requestedIdx = request.fileIdx, + filename = request.filename, + ) + } ensureCurrentGeneration(generation) - val streamUrl = api.getStreamUrl(magnetLink, resolvedIdx) - Log.d(TAG, "Stream URL: $streamUrl") + val streamSelector = TorrServerStreamSelector( + legacyIndex = resolvedIdx, + fileIdx = request.fileIdx, + filename = requestedName, + ) + val streamUrl = api.getStreamUrl( + magnetLink = magnetLink, + selector = streamSelector, + ) - startStatsPolling(hash, generation) + startPreload( + hash = hash, + generation = generation, + magnetLink = magnetLink, + selector = streamSelector, + ) + startStatsPolling( + hash = hash, + generation = generation, + ) ensureCurrentGeneration(generation) _state.value = P2pStreamingState.Streaming( @@ -90,8 +213,11 @@ actual object P2pStreamingEngine { streamUrl } catch (e: CancellationException) { + attachedHash?.takeUnless(::isHashCurrent)?.let(::scheduleIdleDrop) throw e } catch (e: Exception) { + Log.w(TAG, "Failed to start P2P stream", e) + attachedHash?.takeUnless(::isHashCurrent)?.let(::scheduleIdleDrop) if (isCurrentGeneration(generation)) { _state.value = P2pStreamingState.Error(e.message ?: "Unknown torrent error") } @@ -100,52 +226,36 @@ actual object P2pStreamingEngine { } actual fun stopStream() { - scheduleStop(stopBinary = false) + detachActiveStream()?.let(::scheduleIdleDrop) } actual fun shutdown() { - scheduleStop(stopBinary = true) - } - - private fun scheduleStop(stopBinary: Boolean) { val hash = detachActiveStream() - val previousCleanup = cleanupJob - cleanupJob = scope.launch { - previousCleanup?.join() - cleanupDetachedStream(hash, stopBinary) + val idleHashes = cancelScheduledIdleDrops() + val warmup = synchronized(lifecycleLock) { + warmupCooldownJob?.cancel() + warmupCooldownJob = null + val job = warmupJob + warmupJob = null + job } - } - - private suspend fun stopStreamNow(stopBinary: Boolean) { - cleanupJob?.join() - val hash = detachActiveStream() - cleanupDetachedStream(hash, stopBinary) - } - - private fun detachActiveStream(): String? { - val detached = synchronized(lifecycleLock) { - streamGeneration += 1 - val hash = currentHash - val job = statsJob - currentHash = null - statsJob = null - hash to job - } - detached.second?.cancel() - _state.value = P2pStreamingState.Idle - return detached.first - } - - private suspend fun cleanupDetachedStream(hash: String?, stopBinary: Boolean) { - hash?.let { + warmup?.cancel() + scope.launch { try { - api.dropTorrent(it) - } catch (e: Exception) { - Log.w(TAG, "Error dropping torrent", e) + warmup?.join() + } catch (_: CancellationException) { } - } - if (stopBinary) { + (listOfNotNull(hash) + idleHashes) + .distinctBy { hashKey(it) } + .forEach { + try { + api.dropTorrent(it) + } catch (e: Exception) { + Log.w(TAG, "Error dropping torrent", e) + } + } + try { binary.stop() } catch (e: Exception) { @@ -154,11 +264,49 @@ actual object P2pStreamingEngine { } } - private fun nextStreamGeneration(): Long = - synchronized(lifecycleLock) { + private data class DetachedStream( + val generation: Long, + val hash: String?, + val statsJob: Job?, + val preloadJob: Job?, + ) + + private fun beginStreamGeneration(): DetachedStream { + val detached = synchronized(lifecycleLock) { streamGeneration += 1 - streamGeneration + val detached = DetachedStream( + generation = streamGeneration, + hash = currentHash, + statsJob = statsJob, + preloadJob = preloadJob, + ) + currentHash = null + statsJob = null + preloadJob = null + detached } + detached.statsJob?.cancel() + detached.preloadJob?.cancel() + return detached + } + + private fun detachActiveStream(): String? { + val detached = beginStreamGeneration() + _state.value = P2pStreamingState.Idle + return detached.hash + } + + private suspend fun awaitWarmup() { + val job = synchronized(lifecycleLock) { warmupJob?.takeIf { it.isActive } } + job?.join() + } + + private fun cancelWarmupCooldown() { + synchronized(lifecycleLock) { + warmupCooldownJob?.cancel() + warmupCooldownJob = null + } + } private fun attachTorrentIfCurrent(generation: Long, hash: String): Boolean = synchronized(lifecycleLock) { @@ -176,59 +324,171 @@ actual object P2pStreamingEngine { } } - private fun buildMagnetUri(infoHash: String, extraTrackers: List): String { - val trackers = (DEFAULT_TRACKERS + extraTrackers).distinct() - val trackerParams = trackers.joinToString("") { "&tr=$it" } - return "magnet:?xt=urn:btih:$infoHash$trackerParams" + private fun isHashCurrent(hash: String): Boolean = + synchronized(lifecycleLock) { hashMatches(currentHash, hash) } + + private fun scheduleIdleDrop(hash: String, delayMs: Long = IDLE_TORRENT_TTL_MS) { + val key = hashKey(hash) + if (key.isBlank()) return + val job = scope.launch { + delay(delayMs) + val shouldDrop = synchronized(lifecycleLock) { + if (hashMatches(currentHash, hash)) { + idleDropJobs.remove(key) + false + } else { + idleDropJobs.remove(key) + true + } + } + if (shouldDrop) { + api.dropTorrent(hash) + } + } + synchronized(lifecycleLock) { + if (hashMatches(currentHash, hash)) { + job.cancel() + return + } + idleDropJobs.remove(key)?.cancel() + idleDropJobs[key] = job + } } + private fun cancelIdleDrop(hash: String) { + synchronized(lifecycleLock) { + idleDropJobs.remove(hashKey(hash))?.let { + it.cancel() + } + } + } + + private fun cancelScheduledIdleDrops(): List = + synchronized(lifecycleLock) { + val hashes = idleDropJobs.keys.toList() + idleDropJobs.values.forEach { it.cancel() } + idleDropJobs.clear() + hashes + } + + private fun hashMatches(left: String?, right: String?): Boolean { + if (left.isNullOrBlank() || right.isNullOrBlank()) return false + return hashKey(left) == hashKey(right) + } + + private fun hashKey(hash: String): String = + hash.trim().lowercase(Locale.US) + + private fun buildMagnetUri( + infoHash: String, + magnetUri: String?, + extraTrackers: List, + ): String { + val parsedMagnet = parseMagnetUri(magnetUri) + val trackers = (DEFAULT_TRACKERS + parsedMagnet.trackers + extraTrackers) + .asSequence() + .mapNotNull(::normalizeTracker) + .distinctBy { it.lowercase(Locale.US) } + .toList() + + return buildString { + append("magnet:?xt=urn:btih:") + append(infoHash.trim()) + parsedMagnet.passthroughParams.forEach { param -> + append('&') + append(param) + } + trackers.forEach { tracker -> + append("&tr=") + append(URLEncoder.encode(tracker, "UTF-8")) + } + } + } + + private data class ParsedMagnet( + val trackers: List, + val passthroughParams: List, + ) + + private fun parseMagnetUri(magnetUri: String?): ParsedMagnet { + val raw = magnetUri + ?.trim() + ?.takeIf { it.startsWith("magnet:", ignoreCase = true) } + ?: return ParsedMagnet(trackers = emptyList(), passthroughParams = emptyList()) + val query = raw.substringAfter('?', missingDelimiterValue = "") + if (query.isBlank()) return ParsedMagnet(trackers = emptyList(), passthroughParams = emptyList()) + + val trackers = mutableListOf() + val passthroughParams = mutableListOf() + query.split('&') + .filter { it.isNotBlank() } + .forEach { param -> + val key = param.substringBefore('=').lowercase(Locale.US) + when (key) { + "tr" -> trackers += decodeQueryValue(param.substringAfter('=', missingDelimiterValue = "")) + "xt" -> Unit + else -> passthroughParams += param + } + } + return ParsedMagnet( + trackers = trackers, + passthroughParams = passthroughParams.distinct(), + ) + } + + private fun decodeQueryValue(value: String): String = + runCatching { URLDecoder.decode(value, "UTF-8") } + .getOrDefault(value) + + private fun normalizeTracker(value: String): String? = + value + .trim() + .removePrefix("tracker:") + .trim() + .takeIf { it.isNotEmpty() } + private suspend fun resolveFileIndex(hash: String, requestedIdx: Int?, filename: String?): Int { - val deadline = System.currentTimeMillis() + 15_000L - var files: List = emptyList() - - while (System.currentTimeMillis() < deadline) { - files = api.getTorrentStats(hash)?.files ?: emptyList() - if (files.isNotEmpty()) break - Log.d(TAG, "Waiting for torrent metadata...") - delay(1_000L) - } - - if (files.isEmpty()) { - Log.w(TAG, "No files after metadata timeout, guessing index ${requestedIdx?.plus(1) ?: 1}") - return requestedIdx?.plus(1) ?: 1 - } - - if (!filename.isNullOrBlank()) { - val name = filename.trim() - val exact = files.firstOrNull { file -> - file.path.substringAfterLast('/').equals(name, ignoreCase = true) - } - if (exact != null) { - Log.d(TAG, "File resolved by exact filename match: ${exact.path} -> id=${exact.id}") - return exact.id - } - - val contains = files.firstOrNull { file -> - file.path.contains(name, ignoreCase = true) - } - if (contains != null) { - Log.d(TAG, "File resolved by filename contains match: ${contains.path} -> id=${contains.id}") - return contains.id - } - } - + val requestedName = filename?.trim()?.takeIf { it.isNotEmpty() } if (requestedIdx != null) { val torrServerIndex = requestedIdx + 1 - if (files.any { it.id == torrServerIndex }) { - Log.d(TAG, "File resolved by ID offset: id=$torrServerIndex") - return torrServerIndex + + if (requestedName != null) { + val files = waitForTorrentFiles( + hash = hash, + timeoutMs = FILE_INDEX_FAST_VALIDATION_TIMEOUT_MS, + ) + if (files.isNotEmpty()) { + resolveByFilename(files, requestedName)?.let { match -> + return match.id + } + + val requestedIdMatch = files.firstOrNull { it.id == torrServerIndex } + if (requestedIdMatch != null) { + return torrServerIndex + } + + val positionalFile = files.getOrNull(requestedIdx) + if (positionalFile != null) { + return positionalFile.id + } + } } + return torrServerIndex } - if (requestedIdx != null && requestedIdx in files.indices) { - val positionalFile = files[requestedIdx] - Log.d(TAG, "File resolved by positional index: [$requestedIdx] -> ${positionalFile.path} (id=${positionalFile.id})") - return positionalFile.id + val files = waitForTorrentFiles( + hash = hash, + timeoutMs = FILE_INDEX_METADATA_TIMEOUT_MS, + ) + + if (files.isEmpty()) { + return 1 + } + + if (requestedName != null) { + resolveByFilename(files, requestedName)?.let { match -> + return match.id + } } val videoFile = files @@ -239,13 +499,89 @@ actual object P2pStreamingEngine { .maxByOrNull { it.length } val result = videoFile?.id ?: files.maxByOrNull { it.length }?.id ?: 1 - Log.d(TAG, "File resolved by largest video fallback: id=$result") return result } - private fun startStatsPolling(hash: String, generation: Long) { - statsJob?.cancel() - statsJob = scope.launch { + private suspend fun waitForTorrentFiles( + hash: String, + timeoutMs: Long, + ): List { + val startedAt = System.currentTimeMillis() + var attempt = 0 + while (System.currentTimeMillis() - startedAt < timeoutMs) { + attempt += 1 + val stats = api.getTorrentStats(hash) + val files = stats?.files.orEmpty() + if (files.isNotEmpty()) { + return files + } + delay(FILE_INDEX_POLL_INTERVAL_MS) + } + return emptyList() + } + + private fun resolveByFilename(files: List, filename: String): TorrServerFile? { + val name = filename.trim() + val exactBasename = files.firstOrNull { file -> + file.path.substringAfterLast('/').equals(name, ignoreCase = true) + } + if (exactBasename != null) { + return exactBasename + } + + val exactPath = files.firstOrNull { file -> + file.path.equals(name, ignoreCase = true) + } + if (exactPath != null) { + return exactPath + } + + val contains = files.firstOrNull { file -> + file.path.contains(name, ignoreCase = true) + } + if (contains != null) { + return contains + } + + return null + } + + private fun startPreload( + hash: String, + generation: Long, + magnetLink: String, + selector: TorrServerStreamSelector, + ) { + val job = scope.launch { + try { + api.preloadTorrent( + magnetLink = magnetLink, + selector = selector, + ) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + Log.w(TAG, "Torrent preload failed", e) + } + } + val previousJob = synchronized(lifecycleLock) { + if (streamGeneration == generation && hashMatches(currentHash, hash)) { + val previous = preloadJob + preloadJob = job + previous + } else { + job.cancel() + null + } + } + previousJob?.cancel() + } + + private fun startStatsPolling( + hash: String, + generation: Long, + ) { + val job = scope.launch { while (isActive) { if (!isCurrentGeneration(generation)) return@launch try { @@ -272,22 +608,46 @@ actual object P2pStreamingEngine { delay(1_000L) } } + val previousJob = synchronized(lifecycleLock) { + if (streamGeneration == generation && hashMatches(currentHash, hash)) { + val previous = statsJob + statsJob = job + previous + } else { + job.cancel() + null + } + } + previousJob?.cancel() } - private fun requireContext(): Context = - appContext ?: throw P2pStreamingException("P2P streaming engine is not initialized") - private val DEFAULT_TRACKERS = listOf( "udp://tracker.opentrackr.org:1337/announce", + "udp://open.demonii.com:1337/announce", "udp://open.stealth.si:80/announce", - "udp://tracker.openbittorrent.com:6969/announce", - "udp://exodus.desync.com:6969/announce", + "https://torrent.tracker.durukanbal.com:443/announce", + "udp://wepzone.net:6969/announce", + "udp://tracker.wepzone.net:6969/announce", "udp://tracker.torrent.eu.org:451/announce", + "udp://tracker.theoks.net:6969/announce", + "udp://tracker.t-1.org:6969/announce", + "udp://tracker.darkness.services:6969/announce", + "udp://tracker-udp.gbitt.info:80/announce", + "udp://t.overflow.biz:6969/announce", + "udp://open.dstud.io:6969/announce", + "udp://explodie.org:6969/announce", + "udp://exodus.desync.com:6969/announce", + "udp://bittorrent-tracker.e-n-c-r-y-p-t.net:1337/announce", + "https://tracker.zhuqiy.com:443/announce", + "https://tracker.pmman.tech:443/announce", + "https://tracker.moeblog.cn:443/announce", + "https://tracker.bt4g.com:443/announce", ) private class TorrServerBinary { private var context: Context? = null private var process: Process? = null + private val startMutex = Mutex() private val healthClient = OkHttpClient.Builder() .connectTimeout(2, TimeUnit.SECONDS) .readTimeout(5, TimeUnit.SECONDS) @@ -299,67 +659,64 @@ actual object P2pStreamingEngine { this.context = context.applicationContext } - suspend fun start() = withContext(Dispatchers.IO) { - if (isRunning()) { - Log.d(TAG, "TorrServer already running") - return@withContext - } - - killOrphanedProcess() - - val ctx = requireContext() - val binaryFile = File(ctx.applicationInfo.nativeLibraryDir, "libtorrserver.so") - if (!binaryFile.exists()) { - throw P2pStreamingException("TorrServer binary not found at ${binaryFile.absolutePath}") - } - - if (!binaryFile.canExecute()) { - binaryFile.setExecutable(true) - } - - val configDir = File(ctx.filesDir, "torrserver").also { it.mkdirs() } - val processBuilder = ProcessBuilder( - binaryFile.absolutePath, - "--port", - PORT.toString(), - "--path", - configDir.absolutePath, - ) - processBuilder.directory(configDir) - processBuilder.redirectErrorStream(true) - - Log.d(TAG, "Starting TorrServer on port $PORT from ${binaryFile.absolutePath}") - process = processBuilder.start() - - val proc = process!! - Thread { - try { - proc.inputStream.bufferedReader().forEachLine { line -> - Log.d(TAG, "[server] $line") - } - } catch (_: Exception) { - } - }.apply { - isDaemon = true - start() - } - - val deadline = System.currentTimeMillis() + STARTUP_TIMEOUT_MS - while (System.currentTimeMillis() < deadline) { + suspend fun start() = startMutex.withLock { + withContext(Dispatchers.IO) { if (isRunning()) { - Log.d(TAG, "TorrServer started successfully") return@withContext } - if (!isProcessAlive(process)) { - val exitCode = process?.exitValue() ?: -1 - process = null - throw P2pStreamingException("TorrServer process died on startup (exit code $exitCode)") - } - delay(HEALTH_CHECK_INTERVAL_MS) - } - stop() - throw P2pStreamingException("TorrServer failed to start within ${STARTUP_TIMEOUT_MS / 1000}s") + killOrphanedProcess() + + val ctx = requireContext() + val binaryFile = File(ctx.applicationInfo.nativeLibraryDir, "libtorrserver.so") + if (!binaryFile.exists()) { + throw P2pStreamingException("TorrServer binary not found at ${binaryFile.absolutePath}") + } + + if (!binaryFile.canExecute()) { + binaryFile.setExecutable(true) + } + + val configDir = File(ctx.filesDir, "torrserver").also { it.mkdirs() } + val processBuilder = ProcessBuilder( + binaryFile.absolutePath, + "--port", + PORT.toString(), + "--path", + configDir.absolutePath, + ) + processBuilder.directory(configDir) + processBuilder.redirectErrorStream(true) + + process = processBuilder.start() + + val proc = process!! + Thread { + try { + proc.inputStream.bufferedReader().forEachLine { } + } catch (_: Exception) { + } + }.apply { + isDaemon = true + start() + } + + val deadline = System.currentTimeMillis() + STARTUP_TIMEOUT_MS + while (System.currentTimeMillis() < deadline) { + if (isRunning()) { + return@withContext + } + if (!isProcessAlive(process)) { + val exitCode = process?.exitValue() ?: -1 + process = null + throw P2pStreamingException("TorrServer process died on startup (exit code $exitCode)") + } + delay(HEALTH_CHECK_INTERVAL_MS) + } + + stop() + throw P2pStreamingException("TorrServer failed to start within ${STARTUP_TIMEOUT_MS / 1000}s") + } } fun isRunning(): Boolean { @@ -389,7 +746,6 @@ actual object P2pStreamingEngine { } } process = null - Log.d(TAG, "TorrServer stopped") } private fun killOrphanedProcess() { @@ -397,7 +753,6 @@ actual object P2pStreamingEngine { val request = Request.Builder().url("$baseUrl/shutdown").build() healthClient.newCall(request).execute().close() Thread.sleep(1_000L) - Log.d(TAG, "Shut down orphaned TorrServer instance") } catch (_: Exception) { } } @@ -426,31 +781,194 @@ actual object P2pStreamingEngine { private data class TorrServerFile( val id: Int, + val index: Int, val path: String, val length: Long, ) + private data class TorrServerStreamSelector( + val legacyIndex: Int?, + val fileIdx: Int?, + val filename: String?, + ) + private data class TorrServerStats( + val status: String, val downloadSpeed: Long, val uploadSpeed: Long, val peers: Int, val seeds: Int, + val totalPeers: Int, + val pendingPeers: Int, + val halfOpenPeers: Int, val preloadedBytes: Long, + val preloadSize: Long, val loadedSize: Long, val torrentSize: Long, + val bytesRead: Long, + val bytesReadUsefulData: Long, + val chunksRead: Long, + val chunksReadUseful: Long, + val chunksReadWasted: Long, + val piecesDirtiedGood: Long, + val piecesDirtiedBad: Long, val files: List, ) + private data class StreamingSettingsResult( + val success: Boolean, + val changed: Boolean, + ) + private class TorrServerApi( private val binary: TorrServerBinary, ) { + private val settingsMutex = Mutex() private val client = OkHttpClient.Builder() .connectTimeout(10, TimeUnit.SECONDS) .readTimeout(30, TimeUnit.SECONDS) .build() + private val preloadClient = client.newBuilder() + .readTimeout(0, TimeUnit.MILLISECONDS) + .build() private val baseUrl: String get() = binary.baseUrl + suspend fun ensureStreamingSettings(): StreamingSettingsResult = settingsMutex.withLock { + withContext(Dispatchers.IO) { + try { + val settings = getSettings() ?: return@withContext StreamingSettingsResult( + success = false, + changed = false, + ) + val changes = mutableListOf() + + putIfDifferent(settings, "CacheSize", STREAMING_CACHE_SIZE_BYTES, changes) + putIfDifferent(settings, "ConnectionsLimit", STREAMING_CONNECTION_LIMIT, changes) + putIfDifferent(settings, "HalfOpenConnectionsLimit", STREAMING_HALF_OPEN_CONNECTION_LIMIT, changes) + putIfDifferent(settings, "TotalHalfOpenConnectionsLimit", STREAMING_TOTAL_HALF_OPEN_CONNECTION_LIMIT, changes) + putIfDifferent(settings, "TorrentPeersHighWater", STREAMING_PEERS_HIGH_WATER, changes) + putIfDifferent(settings, "TorrentPeersLowWater", STREAMING_PEERS_LOW_WATER, changes) + putIfDifferent(settings, "NominalDialTimeoutMs", STREAMING_NOMINAL_DIAL_TIMEOUT_MS, changes) + putIfDifferent(settings, "MinDialTimeoutMs", STREAMING_MIN_DIAL_TIMEOUT_MS, changes) + putIfDifferent(settings, "HandshakeTimeoutMs", STREAMING_HANDSHAKE_TIMEOUT_MS, changes) + putIfDifferent(settings, "TorrentDisconnectTimeout", STREAMING_DISCONNECT_TIMEOUT_SECONDS, changes) + putIfDifferent(settings, "ReaderReadAHead", STREAMING_READ_AHEAD_PERCENT, changes) + putIfDifferent(settings, "PreloadCache", STREAMING_PRELOAD_CACHE_PERCENT, changes) + putIfDifferent(settings, "ResponsiveMode", true, changes) + putIfDifferent(settings, "DisableDHT", false, changes) + putIfDifferent(settings, "DisablePEX", false, changes) + putIfDifferent(settings, "DisableTCP", false, changes) + putIfDifferent(settings, "DisableUTP", false, changes) + putIfDifferent(settings, "DisableUpload", false, changes) + putIfDifferent(settings, "ForceEncrypt", false, changes) + putIfDifferent(settings, "DownloadRateLimit", 0, changes) + putIfDifferent(settings, "UploadRateLimit", 0, changes) + putIfDifferent(settings, "RetrackersMode", 1, changes) + putIfDifferent(settings, "EnableLPD", true, changes) + putIfDifferent(settings, "LPDIPv6", false, changes) + putIfDifferent(settings, "StoreSettingsInJson", true, changes) + + if (changes.isEmpty()) { + return@withContext StreamingSettingsResult( + success = true, + changed = false, + ) + } + + val body = JSONObject().apply { + put("action", "set") + put("sets", settings) + } + val request = Request.Builder() + .url("$baseUrl/settings") + .post(body.toString().toRequestBody(JSON_TYPE)) + .build() + + client.newCall(request).execute().use { response -> + if (!response.isSuccessful) { + Log.w(TAG, "streaming-settings: set failed code=${response.code}") + return@withContext StreamingSettingsResult( + success = false, + changed = false, + ) + } + } + StreamingSettingsResult( + success = true, + changed = true, + ) + } catch (e: Exception) { + Log.w(TAG, "streaming-settings: failed", e) + StreamingSettingsResult( + success = false, + changed = false, + ) + } + } + } + + private fun getSettings(): JSONObject? { + val body = JSONObject().apply { + put("action", "get") + } + val request = Request.Builder() + .url("$baseUrl/settings") + .post(body.toString().toRequestBody(JSON_TYPE)) + .build() + + return client.newCall(request).execute().use { response -> + if (!response.isSuccessful) { + Log.w(TAG, "streaming-settings: get failed code=${response.code}") + null + } else { + JSONObject(response.body?.string()?.takeIf { it.isNotBlank() } ?: "{}") + } + } + } + + private fun putIfDifferent( + settings: JSONObject, + key: String, + desiredValue: Int, + changes: MutableList, + ) { + val hasKey = settings.has(key) + val currentValue = settings.optInt(key, Int.MIN_VALUE) + if (!hasKey || currentValue != desiredValue) { + settings.put(key, desiredValue) + changes += key + } + } + + private fun putIfDifferent( + settings: JSONObject, + key: String, + desiredValue: Long, + changes: MutableList, + ) { + val hasKey = settings.has(key) + val currentValue = settings.optLong(key, Long.MIN_VALUE) + if (!hasKey || currentValue != desiredValue) { + settings.put(key, desiredValue) + changes += key + } + } + + private fun putIfDifferent( + settings: JSONObject, + key: String, + desiredValue: Boolean, + changes: MutableList, + ) { + val hasKey = settings.has(key) + val currentValue = settings.optBoolean(key, !desiredValue) + if (!hasKey || currentValue != desiredValue) { + settings.put(key, desiredValue) + changes += key + } + } + suspend fun addTorrent(magnetLink: String, title: String? = null): String? = withContext(Dispatchers.IO) { val body = JSONObject().apply { put("action", "add") @@ -472,7 +990,6 @@ actual object P2pStreamingEngine { } val json = JSONObject(response.body?.string() ?: "{}") val hash = json.optString("hash", "") - Log.d(TAG, "Torrent added: $hash") hash.ifEmpty { null } } } catch (e: Exception) { @@ -481,6 +998,38 @@ actual object P2pStreamingEngine { } } + suspend fun preloadTorrent( + magnetLink: String, + selector: TorrServerStreamSelector, + ): Boolean = withContext(Dispatchers.IO) { + val url = getStreamUrl( + magnetLink = magnetLink, + selector = selector, + mode = "preload", + ) + val request = Request.Builder() + .url(url) + .get() + .build() + val call = preloadClient.newCall(request) + val cancellationHandle = currentCoroutineContext()[Job]?.invokeOnCompletion { cause -> + if (cause is CancellationException) { + call.cancel() + } + } + try { + call.execute().use { response -> + if (!response.isSuccessful) { + Log.w(TAG, "preload: request failed code=${response.code}") + return@withContext false + } + true + } + } finally { + cancellationHandle?.dispose() + } + } + suspend fun getTorrentStats(hash: String): TorrServerStats? = withContext(Dispatchers.IO) { val body = JSONObject().apply { put("action", "get") @@ -504,6 +1053,7 @@ actual object P2pStreamingEngine { files.add( TorrServerFile( id = file.optInt("id", i + 1), + index = file.optInt("index", i), path = file.optString("path", ""), length = file.optLong("length", 0), ), @@ -511,13 +1061,25 @@ actual object P2pStreamingEngine { } TorrServerStats( + status = json.optString("stat_string", ""), downloadSpeed = json.optLong("download_speed", 0), uploadSpeed = json.optLong("upload_speed", 0), peers = json.optInt("active_peers", 0), seeds = json.optInt("connected_seeders", 0), + totalPeers = json.optInt("total_peers", 0), + pendingPeers = json.optInt("pending_peers", 0), + halfOpenPeers = json.optInt("half_open_peers", 0), preloadedBytes = json.optLong("preloaded_bytes", 0), + preloadSize = json.optLong("preload_size", 0), loadedSize = json.optLong("loaded_size", 0), torrentSize = json.optLong("torrent_size", 0), + bytesRead = json.optLong("bytes_read", 0), + bytesReadUsefulData = json.optLong("bytes_read_useful_data", 0), + chunksRead = json.optLong("chunks_read", 0), + chunksReadUseful = json.optLong("chunks_read_useful", 0), + chunksReadWasted = json.optLong("chunks_read_wasted", 0), + piecesDirtiedGood = json.optLong("pieces_dirtied_good", 0), + piecesDirtiedBad = json.optLong("pieces_dirtied_bad", 0), files = files, ) } @@ -540,15 +1102,27 @@ actual object P2pStreamingEngine { try { client.newCall(request).execute().close() - Log.d(TAG, "Torrent dropped: $hash") } catch (e: Exception) { Log.w(TAG, "dropTorrent error", e) } } - fun getStreamUrl(magnetLink: String, fileIdx: Int): String { - val encodedLink = URLEncoder.encode(magnetLink, "UTF-8") - return "$baseUrl/stream?link=$encodedLink&index=$fileIdx&play" + fun getStreamUrl( + magnetLink: String, + selector: TorrServerStreamSelector, + mode: String = "play", + ): String { + val params = mutableListOf( + "link=${URLEncoder.encode(magnetLink, "UTF-8")}", + mode, + ) + selector.legacyIndex?.let { params += "index=$it" } + selector.fileIdx?.let { params += "fileIdx=$it" } + selector.filename + ?.trim() + ?.takeIf { it.isNotEmpty() } + ?.let { params += "filename=${URLEncoder.encode(it, "UTF-8")}" } + return "$baseUrl/stream?${params.joinToString("&")}" } } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/App.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/App.kt index 444573b93..5c284a7bc 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/App.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/App.kt @@ -1697,7 +1697,7 @@ private fun MainAppContent( replaceStreamRoute: Boolean, ) { val infoHash = stream.p2pInfoHash ?: return - val sentinelUrl = p2pSentinelUrl(infoHash, stream.fileIdx) + val sentinelUrl = p2pSentinelUrl(infoHash, stream.p2pFileIdx) if (playerSettings.streamReuseLastLinkEnabled) { val cacheKey = StreamLinkCacheRepository.contentKey( type = launch.type, @@ -1714,11 +1714,12 @@ private fun MainAppContent( addonId = stream.addonId, requestHeaders = emptyMap(), responseHeaders = emptyMap(), - filename = stream.behaviorHints.filename, + filename = stream.p2pFilename, videoSize = stream.behaviorHints.videoSize, infoHash = infoHash, - fileIdx = stream.fileIdx, - sources = stream.sources, + fileIdx = stream.p2pFileIdx, + magnetUri = stream.torrentMagnetUri, + sources = stream.p2pSourceHints, bingeGroup = stream.behaviorHints.bingeGroup, ) } @@ -1745,8 +1746,9 @@ private fun MainAppContent( parentMetaId = launch.parentMetaId ?: effectiveVideoId, parentMetaType = launch.parentMetaType ?: launch.type, torrentInfoHash = infoHash, - torrentFileIdx = stream.fileIdx, - torrentFilename = stream.behaviorHints.filename, + torrentFileIdx = stream.p2pFileIdx, + torrentFilename = stream.p2pFilename, + torrentMagnetUri = stream.torrentMagnetUri, torrentTrackers = stream.p2pTrackers, initialPositionMs = resolvedResumePositionMs ?: 0L, initialProgressFraction = resolvedResumeProgressFraction, @@ -1814,10 +1816,10 @@ private fun MainAppContent( val maxAgeMs = playerSettings.streamReuseLastLinkCacheHours * 60L * 60L * 1000L val cached = StreamLinkCacheRepository.getValid(cacheKey, maxAgeMs) if (cached != null) { - if (cached.url.isBlank() && !cached.infoHash.isNullOrBlank()) { + if (cached.url.isBlank() && (!cached.infoHash.isNullOrBlank() || !cached.magnetUri.isNullOrBlank())) { val cachedStream = StreamItem( name = cached.streamName, - url = null, + url = cached.magnetUri, infoHash = cached.infoHash, fileIdx = cached.fileIdx, sources = cached.sources, @@ -2329,6 +2331,7 @@ private fun MainAppContent( torrentInfoHash = launch.torrentInfoHash, torrentFileIdx = launch.torrentFileIdx, torrentFilename = launch.torrentFilename, + torrentMagnetUri = launch.torrentMagnetUri, torrentTrackers = launch.torrentTrackers, initialPositionMs = launch.initialPositionMs, initialProgressFraction = launch.initialProgressFraction, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt index 092b855f7..9a2c5c516 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt @@ -45,6 +45,9 @@ object P2pSettingsRepository { if (p2pEnabled == enabled) return p2pEnabled = enabled P2pSettingsStorage.saveP2pEnabled(enabled) + if (!enabled) { + P2pStreamingEngine.shutdown() + } publish() } @@ -94,6 +97,7 @@ data class P2pStreamRequest( val infoHash: String, val fileIdx: Int?, val filename: String? = null, + val magnetUri: String? = null, val trackers: List = emptyList(), ) @@ -119,6 +123,8 @@ class P2pStreamingException(message: String) : Exception(message) expect object P2pStreamingEngine { val state: StateFlow + fun warmup() + fun cooldownWarmup() suspend fun startStream(request: P2pStreamRequest): String fun stopStream() fun shutdown() diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerModels.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerModels.kt index d4bcc0ecf..c2d1337fe 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerModels.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerModels.kt @@ -45,6 +45,7 @@ data class PlayerLaunch( val torrentInfoHash: String? = null, val torrentFileIdx: Int? = null, val torrentFilename: String? = null, + val torrentMagnetUri: String? = null, val torrentTrackers: List = emptyList(), val initialPositionMs: Long = 0L, val initialProgressFraction: Float? = null, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreen.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreen.kt index fc9487bdb..e00f53fc6 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreen.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreen.kt @@ -178,6 +178,7 @@ fun PlayerScreen( torrentInfoHash: String? = null, torrentFileIdx: Int? = null, torrentFilename: String? = null, + torrentMagnetUri: String? = null, torrentTrackers: List = emptyList(), initialPositionMs: Long = 0L, initialProgressFraction: Float? = null, @@ -248,6 +249,7 @@ fun PlayerScreen( var activeTorrentInfoHash by rememberSaveable { mutableStateOf(torrentInfoHash) } var activeTorrentFileIdx by rememberSaveable { mutableStateOf(torrentFileIdx) } var activeTorrentFilename by rememberSaveable { mutableStateOf(torrentFilename) } + var activeTorrentMagnetUri by rememberSaveable { mutableStateOf(torrentMagnetUri) } var activeTorrentTrackers by remember { mutableStateOf(torrentTrackers) } var p2pResolvedSourceUrl by remember { mutableStateOf(null) } val activePlaybackIdentity = activeTorrentInfoHash @@ -1203,6 +1205,7 @@ fun PlayerScreen( activeTorrentInfoHash = null activeTorrentFileIdx = null activeTorrentFilename = null + activeTorrentMagnetUri = null activeTorrentTrackers = emptyList() p2pResolvedSourceUrl = null } @@ -1230,11 +1233,12 @@ fun PlayerScreen( addonId = stream.addonId, requestHeaders = emptyMap(), responseHeaders = emptyMap(), - filename = stream.behaviorHints.filename, + filename = stream.p2pFilename, videoSize = stream.behaviorHints.videoSize, infoHash = infoHash, - fileIdx = stream.fileIdx, - sources = stream.sources, + fileIdx = stream.p2pFileIdx, + magnetUri = stream.torrentMagnetUri, + sources = stream.p2pSourceHints, bingeGroup = stream.behaviorHints.bingeGroup, ) } @@ -1255,13 +1259,14 @@ fun PlayerScreen( season = activeSeasonNumber, episode = activeEpisodeNumber, ) - activeSourceUrl = p2pSentinelUrl(infoHash, stream.fileIdx) + activeSourceUrl = p2pSentinelUrl(infoHash, stream.p2pFileIdx) activeSourceAudioUrl = null activeSourceHeaders = emptyMap() activeSourceResponseHeaders = emptyMap() activeTorrentInfoHash = infoHash - activeTorrentFileIdx = stream.fileIdx - activeTorrentFilename = stream.behaviorHints.filename + activeTorrentFileIdx = stream.p2pFileIdx + activeTorrentFilename = stream.p2pFilename + activeTorrentMagnetUri = stream.torrentMagnetUri activeTorrentTrackers = stream.p2pTrackers activeStreamTitle = stream.streamLabel activeStreamSubtitle = stream.streamSubtitle @@ -1313,13 +1318,14 @@ fun PlayerScreen( season = episode.season, episode = episode.episode, ) - activeSourceUrl = p2pSentinelUrl(infoHash, stream.fileIdx) + activeSourceUrl = p2pSentinelUrl(infoHash, stream.p2pFileIdx) activeSourceAudioUrl = null activeSourceHeaders = emptyMap() activeSourceResponseHeaders = emptyMap() activeTorrentInfoHash = infoHash - activeTorrentFileIdx = stream.fileIdx - activeTorrentFilename = stream.behaviorHints.filename + activeTorrentFileIdx = stream.p2pFileIdx + activeTorrentFilename = stream.p2pFilename + activeTorrentMagnetUri = stream.torrentMagnetUri activeTorrentTrackers = stream.p2pTrackers activeStreamTitle = stream.streamLabel activeStreamSubtitle = stream.streamSubtitle @@ -1938,6 +1944,7 @@ fun PlayerScreen( activeTorrentInfoHash, activeTorrentFileIdx, activeTorrentFilename, + activeTorrentMagnetUri, activeTorrentTrackers, p2pSettingsUiState.p2pEnabled, ) { @@ -1954,6 +1961,7 @@ fun PlayerScreen( p2pResolvedSourceUrl = null val requestedFileIdx = activeTorrentFileIdx val requestedFilename = activeTorrentFilename + val requestedMagnetUri = activeTorrentMagnetUri val requestedTrackers = activeTorrentTrackers errorMessage = null playerController = null @@ -1967,6 +1975,7 @@ fun PlayerScreen( infoHash = infoHash, fileIdx = requestedFileIdx, filename = requestedFilename, + magnetUri = requestedMagnetUri, trackers = requestedTrackers, ), ) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamLinkCacheRepository.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamLinkCacheRepository.kt index c9d1c34e1..ac2140f5c 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamLinkCacheRepository.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamLinkCacheRepository.kt @@ -16,6 +16,7 @@ data class CachedStreamLink( val videoSize: Long? = null, val infoHash: String? = null, val fileIdx: Int? = null, + val magnetUri: String? = null, val sources: List = emptyList(), val bingeGroup: String? = null, ) @@ -52,6 +53,7 @@ object StreamLinkCacheRepository { videoSize: Long? = null, infoHash: String? = null, fileIdx: Int? = null, + magnetUri: String? = null, sources: List = emptyList(), bingeGroup: String? = null, ) { @@ -67,6 +69,7 @@ object StreamLinkCacheRepository { videoSize = videoSize, infoHash = infoHash, fileIdx = fileIdx, + magnetUri = magnetUri, sources = sources, bingeGroup = bingeGroup, ) @@ -92,7 +95,7 @@ object StreamLinkCacheRepository { StreamLinkCacheStorage.removeEntry(hashedKey(contentKey)) return null } - if (entry.url.isBlank() && entry.infoHash.isNullOrBlank()) { + if (entry.url.isBlank() && entry.infoHash.isNullOrBlank() && entry.magnetUri.isNullOrBlank()) { StreamLinkCacheStorage.removeEntry(hashedKey(contentKey)) return null } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamModels.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamModels.kt index d7d5dec59..7a8406c33 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamModels.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamModels.kt @@ -36,7 +36,7 @@ data class StreamItem( .firstOrNull { !it.isMagnetLink() } val torrentMagnetUri: String? - get() = listOfNotNull(url, externalUrl) + get() = listOfNotNull(url, externalUrl, clientResolve?.magnetUri) .firstOrNull { it.isMagnetLink() } val isDirectDebridStream: Boolean @@ -61,10 +61,17 @@ data class StreamItem( val p2pInfoHash: String? get() = infoHash.normalizedInfoHash() ?: clientResolve?.infoHash.normalizedInfoHash() + ?: clientResolve?.magnetUri.extractBtihInfoHash() ?: torrentMagnetUri.extractBtihInfoHash() + val p2pFileIdx: Int? + get() = fileIdx ?: clientResolve?.fileIdx + + val p2pFilename: String? + get() = behaviorHints.filename ?: clientResolve?.filename + val p2pTrackers: List - get() = sources + get() = p2pSourceHints .asSequence() .filter { it.startsWith("tracker:") } .map { it.removePrefix("tracker:").trim() } @@ -72,6 +79,10 @@ data class StreamItem( .distinct() .toList() + val p2pSourceHints: List + get() = (sources + clientResolve?.sources.orEmpty()) + .distinct() + val isAddonDebridCandidate: Boolean get() = isInstalledAddonStream && (needsLocalDebridResolve || isDirectDebridStream) diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamsScreen.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamsScreen.kt index b107d575c..0c3385e24 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamsScreen.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/streams/StreamsScreen.kt @@ -53,6 +53,7 @@ import androidx.compose.material3.Icon import androidx.compose.material3.MaterialTheme import androidx.compose.material3.Text import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf @@ -91,6 +92,8 @@ import coil3.compose.AsyncImage import com.nuvio.app.core.ui.nuvioSafeBottomPadding import com.nuvio.app.features.debrid.DebridProviders import com.nuvio.app.features.debrid.DebridSettingsRepository +import com.nuvio.app.features.p2p.P2pSettingsRepository +import com.nuvio.app.features.p2p.P2pStreamingEngine import com.nuvio.app.features.player.PlayerSettingsRepository import com.nuvio.app.features.watchprogress.WatchProgressRepository import kotlinx.coroutines.launch @@ -140,6 +143,10 @@ fun StreamsScreen( DebridSettingsRepository.ensureLoaded() DebridSettingsRepository.uiState }.collectAsStateWithLifecycle() + val p2pSettings by remember { + P2pSettingsRepository.ensureLoaded() + P2pSettingsRepository.uiState + }.collectAsStateWithLifecycle() val watchProgressUiState by remember { WatchProgressRepository.ensureLoaded() WatchProgressRepository.uiState @@ -181,6 +188,15 @@ fun StreamsScreen( } } + DisposableEffect(P2pSettingsRepository.isVisible, p2pSettings.p2pEnabled) { + if (P2pSettingsRepository.isVisible && p2pSettings.p2pEnabled) { + P2pStreamingEngine.warmup() + } + onDispose { + P2pStreamingEngine.cooldownWarmup() + } + } + LaunchedEffect(type, videoId, seasonNumber, episodeNumber, manualSelection) { StreamsRepository.load( type = type, diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/streams/StreamItemP2pMetadataTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/streams/StreamItemP2pMetadataTest.kt new file mode 100644 index 000000000..ee8a08254 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/streams/StreamItemP2pMetadataTest.kt @@ -0,0 +1,57 @@ +package com.nuvio.app.features.streams + +import kotlin.test.Test +import kotlin.test.assertEquals + +class StreamItemP2pMetadataTest { + @Test + fun usesClientResolveP2pMetadataWhenTopLevelFieldsAreMissing() { + val magnet = "magnet:?xt=urn:btih:ABC123&dn=Movie&tr=udp%3A%2F%2Fmagnet.test%3A80%2Fannounce" + val stream = StreamItem( + name = "Resolved torrent", + addonName = "Addon", + addonId = "addon:test", + sources = listOf( + "tracker:udp://base.test:80/announce", + "dht:ABC123", + ), + clientResolve = StreamClientResolve( + infoHash = "ABC123", + fileIdx = 2, + magnetUri = magnet, + filename = "movie.mkv", + sources = listOf( + "tracker:udp://client.test:80/announce", + "tracker:udp://base.test:80/announce", + ), + ), + ) + + assertEquals("ABC123", stream.p2pInfoHash) + assertEquals(2, stream.p2pFileIdx) + assertEquals("movie.mkv", stream.p2pFilename) + assertEquals(magnet, stream.torrentMagnetUri) + assertEquals( + listOf( + "udp://base.test:80/announce", + "udp://client.test:80/announce", + ), + stream.p2pTrackers, + ) + } + + @Test + fun extractsP2pInfoHashFromClientResolveMagnet() { + val stream = StreamItem( + name = "Magnet-only torrent", + addonName = "Addon", + addonId = "addon:test", + clientResolve = StreamClientResolve( + magnetUri = "magnet:?xt=urn:btih:def456&dn=Movie", + ), + ) + + assertEquals("def456", stream.p2pInfoHash) + assertEquals("magnet:?xt=urn:btih:def456&dn=Movie", stream.torrentMagnetUri) + } +} diff --git a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt index df670f29c..ecca96a16 100644 --- a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt +++ b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt @@ -8,6 +8,14 @@ actual object P2pStreamingEngine { private val _state = MutableStateFlow(P2pStreamingState.Idle) actual val state: StateFlow = _state.asStateFlow() + actual fun warmup() { + _state.value = P2pStreamingState.Idle + } + + actual fun cooldownWarmup() { + _state.value = P2pStreamingState.Idle + } + actual suspend fun startStream(request: P2pStreamRequest): String { val message = "P2P streaming is not available on this platform" _state.value = P2pStreamingState.Error(message) diff --git a/scripts/build-torrserver-android.sh b/scripts/build-torrserver-android.sh new file mode 100755 index 000000000..1ca865170 --- /dev/null +++ b/scripts/build-torrserver-android.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +TORRSERVER_DIR="${TORRSERVER_DIR:-"${ROOT_DIR}/vendor/TorrServer/server"}" +OUT_DIR="${OUT_DIR:-"${ROOT_DIR}/composeApp/src/androidMain/jniLibs"}" +GO_BIN="${GO_BIN:-go}" +UPX_BIN="${UPX_BIN:-upx}" +SDK_ROOT="${ANDROID_HOME:-${ANDROID_SDK_ROOT:-"${HOME}/Library/Android/sdk"}}" +NDK_ROOT="${ANDROID_NDK_HOME:-}" +export GOCACHE="${GOCACHE:-"${ROOT_DIR}/build/go-cache"}" +export GOMODCACHE="${GOMODCACHE:-"${ROOT_DIR}/build/go-mod-cache"}" + +if ! command -v "${GO_BIN}" >/dev/null 2>&1; then + echo "Go toolchain not found. Install Go or set GO_BIN=/path/to/go." >&2 + exit 1 +fi + +if [[ ! -d "${TORRSERVER_DIR}" ]]; then + echo "TorrServer source not found at ${TORRSERVER_DIR}" >&2 + exit 1 +fi + +if [[ -z "${NDK_ROOT}" ]]; then + if [[ ! -d "${SDK_ROOT}/ndk" ]]; then + echo "Android NDK not found. Set ANDROID_HOME, ANDROID_SDK_ROOT, or ANDROID_NDK_HOME." >&2 + exit 1 + fi + NDK_VERSION="$(ls -1 "${SDK_ROOT}/ndk" | sort | tail -n 1)" + NDK_ROOT="${SDK_ROOT}/ndk/${NDK_VERSION}" +fi + +PREBUILT_ROOT="${NDK_ROOT}/toolchains/llvm/prebuilt" +HOST_TAG="" +for candidate in darwin-x86_64 linux-x86_64; do + if [[ -d "${PREBUILT_ROOT}/${candidate}" ]]; then + HOST_TAG="${candidate}" + break + fi +done + +if [[ -z "${HOST_TAG}" ]]; then + echo "Could not find an LLVM prebuilt toolchain under ${PREBUILT_ROOT}" >&2 + exit 1 +fi + +TOOLCHAIN="${PREBUILT_ROOT}/${HOST_TAG}" +LDFLAGS="${LDFLAGS:-"-s -w -checklinkname=0"}" +BUILD_FLAGS=(-tags=nosqlite -trimpath -ldflags="${LDFLAGS}") +mkdir -p "${GOCACHE}" "${GOMODCACHE}" + +build_abi() { + local abi="$1" + local goarch="$2" + local goarm="$3" + local triple="$4" + local api_level="$5" + local cc="${TOOLCHAIN}/bin/${triple}${api_level}-clang" + local cxx="${TOOLCHAIN}/bin/${triple}${api_level}-clang++" + local output="${OUT_DIR}/${abi}/libtorrserver.so" + + if [[ ! -x "${cc}" ]]; then + echo "Compiler not found: ${cc}" >&2 + exit 1 + fi + + mkdir -p "$(dirname "${output}")" + echo "Building ${abi} -> ${output}" + + local env_vars=( + GOOS=android + GOARCH="${goarch}" + CGO_ENABLED=1 + CC="${cc}" + CXX="${cxx}" + ) + if [[ -n "${goarm}" ]]; then + env_vars+=(GOARM="${goarm}") + fi + + ( + cd "${TORRSERVER_DIR}" + env "${env_vars[@]}" "${GO_BIN}" build "${BUILD_FLAGS[@]}" -o "${output}" ./cmd + ) + chmod 755 "${output}" + if [[ "${USE_UPX:-0}" == "1" ]] && command -v "${UPX_BIN}" >/dev/null 2>&1; then + "${UPX_BIN}" -q "${output}" || echo "UPX compression failed for ${output}; leaving uncompressed" >&2 + fi +} + +build_abi "arm64-v8a" "arm64" "" "aarch64-linux-android" "21" +build_abi "armeabi-v7a" "arm" "7" "armv7a-linux-androideabi" "21" +build_abi "x86" "386" "" "i686-linux-android" "21" +build_abi "x86_64" "amd64" "" "x86_64-linux-android" "21" + +echo "TorrServer Android binaries updated in ${OUT_DIR}" diff --git a/scripts/run-mobile.sh b/scripts/run-mobile.sh index b6b0e1476..ccd449987 100755 --- a/scripts/run-mobile.sh +++ b/scripts/run-mobile.sh @@ -140,46 +140,75 @@ android_flavor_task_part() { esac } -android_apk_path() { +android_apk_dir() { local flavor="$1" case "$flavor" in full) - echo "$ROOT_DIR/composeApp/build/outputs/apk/full/debug/composeApp-full-debug.apk" + echo "$ROOT_DIR/composeApp/build/outputs/apk/full/debug" ;; playstore) - echo "$ROOT_DIR/composeApp/build/outputs/apk/playstore/debug/composeApp-playstore-debug.apk" + echo "$ROOT_DIR/composeApp/build/outputs/apk/playstore/debug" ;; esac } -build_android_apk() { +android_device_primary_abi() { + local serial="$1" + adb -s "$serial" shell getprop ro.product.cpu.abi | tr -d '\r' +} + +android_split_apk_path() { local flavor="$1" - local flavor_task_part + local serial="$2" + local apk_dir + local abi local apk_path - flavor_task_part="$(android_flavor_task_part "$flavor")" - apk_path="$(android_apk_path "$flavor")" + apk_dir="$(android_apk_dir "$flavor")" + abi="$(android_device_primary_abi "$serial")" - echo "Building Android $flavor debug APK..." >&2 - "$GRADLEW" ":composeApp:assemble${flavor_task_part}Debug" >&2 + apk_path="$(find "$apk_dir" -maxdepth 1 -type f -name "*-${abi}-debug.apk" | sort | head -n 1)" + if [[ -z "$apk_path" ]]; then + apk_path="$(find "$apk_dir" -maxdepth 1 -type f -name "*${abi}*.apk" | sort | head -n 1)" + fi - if [[ ! -f "$apk_path" ]]; then - echo "Expected APK not found at: $apk_path" >&2 + if [[ -z "$apk_path" || ! -f "$apk_path" ]]; then + echo "Expected split APK for ABI '$abi' not found in: $apk_dir" >&2 + find "$apk_dir" -maxdepth 1 -type f -name "*.apk" -print >&2 || true exit 1 fi printf '%s\n' "$apk_path" } +build_android_apks() { + local flavor="$1" + local flavor_task_part + local apk_dir + + flavor_task_part="$(android_flavor_task_part "$flavor")" + apk_dir="$(android_apk_dir "$flavor")" + + echo "Building Android $flavor debug split APKs..." >&2 + "$GRADLEW" ":composeApp:assemble${flavor_task_part}Debug" >&2 + + if ! find "$apk_dir" -maxdepth 1 -type f -name "*.apk" | grep -q .; then + echo "Expected split APKs not found in: $apk_dir" >&2 + exit 1 + fi +} + install_and_launch_android() { local device_label="$1" - local apk_path="$2" + local flavor="$2" shift 2 + local apk_path local serial for serial in "$@"; do - echo "Installing on $device_label $serial..." + apk_path="$(android_split_apk_path "$flavor" "$serial")" + echo "Installing on $device_label $serial: $apk_path" adb -s "$serial" install -r "$apk_path" echo "Launching app on $serial..." @@ -248,10 +277,9 @@ run_android_emulator() { wait_for_android_emulator "$serial" done - local apk_path - apk_path="$(build_android_apk "$flavor")" + build_android_apks "$flavor" - install_and_launch_android "emulator" "$apk_path" "${booted_serials[@]}" + install_and_launch_android "emulator" "$flavor" "${booted_serials[@]}" } run_android_physical() { @@ -280,10 +308,9 @@ run_android_physical() { wait_for_android_device "$serial" done - local apk_path - apk_path="$(build_android_apk "$flavor")" + build_android_apks "$flavor" - install_and_launch_android "physical device" "$apk_path" "${serials[@]}" + install_and_launch_android "physical device" "$flavor" "${serials[@]}" } run_ios_simulator() {