diff --git a/app/src/androidMain/kotlin/com/aerotoss/transfer/AndroidTransferManager.kt b/app/src/androidMain/kotlin/com/aerotoss/transfer/AndroidTransferManager.kt index 3584e54..a06a0bc 100644 --- a/app/src/androidMain/kotlin/com/aerotoss/transfer/AndroidTransferManager.kt +++ b/app/src/androidMain/kotlin/com/aerotoss/transfer/AndroidTransferManager.kt @@ -21,17 +21,16 @@ import java.util.concurrent.Semaphore import java.util.concurrent.atomic.AtomicBoolean @OptIn(ExperimentalCoroutinesApi::class) - class AndroidFileTransferManager(private val context: Context) : TransferManager { private val _incomingTransfers = MutableStateFlow>(emptyList()) - override val incomingTransfers: Flow = _incomingTransfers.mapLatest { list -> - list.lastOrNull() - }.filterNotNull() + override val incomingTransfers: Flow = _incomingTransfers.flatMapLatest { list -> + flow { list.forEach { emit(it) } } + } private val _outgoingTransfers = MutableStateFlow>(emptyList()) - override val outgoingTransfers: Flow = _outgoingTransfers.mapLatest { list -> - list.lastOrNull() - }.filterNotNull() + override val outgoingTransfers: Flow = _outgoingTransfers.flatMapLatest { list -> + flow { list.forEach { emit(it) } } + } private var serverSocket: ServerSocket? = null private var serverThread: Thread? = null @@ -40,6 +39,7 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager private val activeSockets = ConcurrentHashMap() private val running = AtomicBoolean(false) private val connectionLimiter = Semaphore(10) + private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true } private val downloadsDir: File = run { val dir = context.getExternalFilesDir(Environment.DIRECTORY_DOWNLOADS) @@ -51,11 +51,9 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager fun startServer(port: Int = 0): Int { if (!running.compareAndSet(false, true)) return serverSocket?.localPort ?: 0 - try { val socket = ServerSocket(port, 50, InetAddress.getByName("0.0.0.0")) serverSocket = socket - serverThread = Thread { while (running.get() && !socket.isClosed) { try { @@ -65,12 +63,7 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager if (running.get()) break } } - }.apply { - isDaemon = true - name = "aerotoss-server" - start() - } - + }.apply { isDaemon = true; name = "aerotoss-server"; start() } return socket.localPort } catch (e: Exception) { e.printStackTrace() @@ -92,35 +85,31 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager var requestFileName: String = "" try { socket.use { sock -> - sock.soTimeout = 60_000 - val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) - val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) + sock.soTimeout = 120_000 + val rawIn = sock.getInputStream() + val rawOut = sock.getOutputStream() + val din = DataInputStream(BufferedInputStream(rawIn, 4096)) + val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096)) - val requestJson = input.readUTF() - val request = try { - Json.decodeFromString(requestJson) - } catch (_: Exception) { - return@withContext - } + val requestJson = din.readUTF() + val request = try { Json.decodeFromString(requestJson) } catch (_: Exception) { return@withContext } requestFileName = request.fileName - if (request.fileSize <= 0 || request.fileSize > FileUtils.MAX_FILE_SIZE) { - output.writeBoolean(false) + if (request.fileSize < 0 || request.fileSize > FileUtils.MAX_FILE_SIZE) { + dout.writeBoolean(false) return@withContext } val progress = TransferProgress( - id = progressId, - request = request, - bytesTransferred = 0, - totalBytes = request.fileSize, + id = progressId, request = request, + bytesTransferred = 0, totalBytes = request.fileSize, state = TransferState.PENDING ) updateIncoming(progress) - output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) + dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) - val accepted = input.readBoolean() + val accepted = din.readBoolean() if (!accepted) { updateIncoming(progress.copy(state = TransferState.CANCELLED)) return@withContext @@ -128,57 +117,53 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager val file = FileUtils.resolveUniqueFile(downloadsDir, request.fileName) if (!FileUtils.validateFilePath(file, downloadsDir)) { - output.writeBoolean(false) + dout.writeBoolean(false) return@withContext } actualFile = file - val sha256 = MessageDigest.getInstance("SHA-256") - var bytesWritten = 0L updateIncoming(progress.copy(state = TransferState.TRANSFERRING)) + var bytesWritten = 0L + val buffer = ByteArray(1048576) file.outputStream().use { fos -> - val buffer = ByteArray(262144) var remaining = request.fileSize while (remaining > 0) { val toRead = minOf(buffer.size.toLong(), remaining).toInt() - val read = input.read(buffer, 0, toRead) + val read = rawIn.read(buffer, 0, toRead) if (read == -1) break fos.write(buffer, 0, read) - sha256.update(buffer, 0, read) bytesWritten += read remaining -= read - updateIncomingById(progressId, TransferProgress( - id = progressId, - request = request, - bytesTransferred = bytesWritten, - totalBytes = request.fileSize, - state = TransferState.TRANSFERRING - )) + } + } + + val sha256 = MessageDigest.getInstance("SHA-256") + file.inputStream().use { fis -> + val buf = ByteArray(1048576) + while (true) { + val read = fis.read(buf) + if (read == -1) break + sha256.update(buf, 0, read) } } val checksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" - output.writeUTF(checksum) - output.writeLong(bytesWritten) + dout.writeUTF(checksum) + dout.writeLong(bytesWritten) if (bytesWritten == request.fileSize) { updateIncomingById(progressId, TransferProgress( - id = progressId, - request = request, - bytesTransferred = bytesWritten, - totalBytes = request.fileSize, + id = progressId, request = request, + bytesTransferred = bytesWritten, totalBytes = request.fileSize, state = TransferState.COMPLETED )) } else { FileUtils.deleteIfExists(file) updateIncomingById(progressId, TransferProgress( - id = progressId, - request = request, - bytesTransferred = bytesWritten, - totalBytes = request.fileSize, - state = TransferState.FAILED, - error = "Incomplete transfer: expected ${request.fileSize}, got $bytesWritten" + id = progressId, request = request, + bytesTransferred = bytesWritten, totalBytes = request.fileSize, + state = TransferState.FAILED, error = "Incomplete transfer" )) } } @@ -186,12 +171,8 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager actualFile?.let { FileUtils.deleteIfExists(it) } val current = _incomingTransfers.value.find { it.id == progressId } if (current != null && current.state != TransferState.COMPLETED && - current.state != TransferState.FAILED && current.state != TransferState.CANCELLED - ) { - updateIncomingById(progressId, current.copy( - state = TransferState.FAILED, - error = "Transfer failed" - )) + current.state != TransferState.FAILED && current.state != TransferState.CANCELLED) { + updateIncomingById(progressId, current.copy(state = TransferState.FAILED, error = "Transfer failed")) } } finally { connectionLimiter.release() @@ -200,24 +181,16 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager } override suspend fun sendFile(file: File, targetHost: String, targetPort: Int): Flow { - require(file.exists() && file.isFile) { "File does not exist or is not a regular file" } - require(file.length() <= FileUtils.MAX_FILE_SIZE) { "File exceeds maximum size" } - + require(file.exists() && file.isFile) + require(file.length() <= FileUtils.MAX_FILE_SIZE) val requestId = UUID.randomUUID().toString() val request = TransferRequest( - fileName = file.name, - fileSize = file.length(), - mimeType = "application/octet-stream", - senderId = "", - senderName = getDeviceName() + fileName = file.name, fileSize = file.length(), + mimeType = "application/octet-stream", senderId = "", senderName = getDeviceName() ) - val initialProgress = TransferProgress( - id = requestId, - request = request, - bytesTransferred = 0, - totalBytes = file.length(), - state = TransferState.PENDING + id = requestId, request = request, + bytesTransferred = 0, totalBytes = file.length(), state = TransferState.PENDING ) updateOutgoing(initialProgress) @@ -230,44 +203,39 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000) activeSockets[requestId] = socket socket.use { sock -> - sock.soTimeout = 60_000 - val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) - val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) + sock.soTimeout = 120_000 + val rawOut = sock.getOutputStream() + val rawIn = sock.getInputStream() + val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096)) + val din = DataInputStream(BufferedInputStream(rawIn, 4096)) - output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) + dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) + val serverResponseJson = din.readUTF() + try { Json.decodeFromString(serverResponseJson) } catch (_: Exception) {} + dout.writeBoolean(true) + dout.flush() - val serverResponseJson = input.readUTF() - val serverRequest = try { - Json.decodeFromString(serverResponseJson) - } catch (_: Exception) { - request - } + updateOutgoingById(requestId, initialProgress.copy( + state = TransferState.TRANSFERRING, + lastUpdateTimeMs = System.currentTimeMillis(), lastBytesTransferred = 0 + )) - output.writeBoolean(true) - - updateOutgoingById(requestId, initialProgress.copy(state = TransferState.TRANSFERRING)) - - val sha256 = MessageDigest.getInstance("SHA-256") var bytesSent = 0L var lastProgressTime = System.currentTimeMillis() var lastProgressBytes = 0L - val buffer = ByteArray(262144) + val buffer = ByteArray(1048576) file.inputStream().use { fis -> while (true) { val read = fis.read(buffer) if (read == -1) break - output.write(buffer, 0, read) - sha256.update(buffer, 0, read) + rawOut.write(buffer, 0, read) bytesSent += read val now = System.currentTimeMillis() - if (now - lastProgressTime >= 500) { + if (now - lastProgressTime >= 1000) { updateOutgoingById(requestId, TransferProgress( - id = requestId, - request = request, - bytesTransferred = bytesSent, - totalBytes = file.length(), - lastUpdateTimeMs = now, - lastBytesTransferred = lastProgressBytes, + id = requestId, request = request, + bytesTransferred = bytesSent, totalBytes = file.length(), + lastUpdateTimeMs = now, lastBytesTransferred = lastProgressBytes, state = TransferState.TRANSFERRING )) lastProgressTime = now @@ -275,42 +243,37 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager } } } - output.flush() - - val serverChecksum = input.readUTF() - val bytesReceived = input.readLong() + rawOut.flush() + val sha256 = MessageDigest.getInstance("SHA-256") + file.inputStream().use { fis -> + val buf = ByteArray(1048576) + while (true) { val read = fis.read(buf); if (read == -1) break; sha256.update(buf, 0, read) } + } val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" + val serverChecksum = din.readUTF() + val bytesReceived = din.readLong() if (bytesReceived == file.length() && serverChecksum == localChecksum) { updateOutgoingById(requestId, TransferProgress( - id = requestId, - request = request, - bytesTransferred = file.length(), - totalBytes = file.length(), + id = requestId, request = request, + bytesTransferred = file.length(), totalBytes = file.length(), state = TransferState.COMPLETED )) } else { updateOutgoingById(requestId, TransferProgress( - id = requestId, - request = request, - bytesTransferred = bytesSent, - totalBytes = file.length(), - state = TransferState.FAILED, - error = "Checksum mismatch or incomplete transfer" + id = requestId, request = request, + bytesTransferred = bytesSent, totalBytes = file.length(), + state = TransferState.FAILED, error = "Checksum mismatch" )) } } } catch (_: Exception) { - updateOutgoingById(requestId, initialProgress.copy( - state = TransferState.FAILED, - error = "Transfer failed" - )) + updateOutgoingById(requestId, initialProgress.copy(state = TransferState.FAILED, error = "Transfer failed")) } finally { activeJobs.remove(requestId) } } - activeJobs[requestId] = job return outgoingTransfers.filter { it.id == requestId } } @@ -329,32 +292,25 @@ class AndroidFileTransferManager(private val context: Context) : TransferManager running.set(false) activeJobs.values.forEach { it.cancel() } activeJobs.clear() + activeSockets.clear() scope.cancel() try { serverSocket?.close() } catch (_: Exception) {} serverThread?.interrupt() } private fun updateIncoming(progress: TransferProgress) { - _incomingTransfers.update { list -> - list.filter { it.id != progress.id }.plus(progress).takeLast(50) - } + _incomingTransfers.update { list -> list.filter { it.id != progress.id }.plus(progress).takeLast(50) } } private fun updateIncomingById(id: String, progress: TransferProgress) { - _incomingTransfers.update { list -> - list.filter { it.id != id }.plus(progress).takeLast(50) - } + _incomingTransfers.update { list -> list.filter { it.id != id }.plus(progress).takeLast(50) } } private fun updateOutgoing(progress: TransferProgress) { - _outgoingTransfers.update { list -> - list.filter { it.id != progress.id }.plus(progress).takeLast(50) - } + _outgoingTransfers.update { list -> list.filter { it.id != progress.id }.plus(progress).takeLast(50) } } private fun updateOutgoingById(id: String, progress: TransferProgress) { - _outgoingTransfers.update { list -> - list.filter { it.id != id }.plus(progress) - } + _outgoingTransfers.update { list -> list.filter { it.id != id }.plus(progress).takeLast(50) } } } diff --git a/app/src/desktopMain/kotlin/com/aerotoss/transfer/DesktopTransferManager.kt b/app/src/desktopMain/kotlin/com/aerotoss/transfer/DesktopTransferManager.kt index c9b4262..0dcd959 100644 --- a/app/src/desktopMain/kotlin/com/aerotoss/transfer/DesktopTransferManager.kt +++ b/app/src/desktopMain/kotlin/com/aerotoss/transfer/DesktopTransferManager.kt @@ -228,19 +228,19 @@ class DesktopFileTransferManager : TransferManager { socket.connect(java.net.InetSocketAddress(targetHost, targetPort), 5000) activeSockets[requestId] = socket socket.use { sock -> - sock.soTimeout = 60_000 - val input = DataInputStream(BufferedInputStream(sock.getInputStream(), 262144)) - val output = DataOutputStream(BufferedOutputStream(sock.getOutputStream(), 262144)) + sock.soTimeout = 120_000 + val rawOut = sock.getOutputStream() + val rawIn = sock.getInputStream() - output.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) + // Protocol handshake with DataOutputStream + val dout = DataOutputStream(BufferedOutputStream(rawOut, 4096)) + val din = DataInputStream(BufferedInputStream(rawIn, 4096)) - val serverResponseJson = input.readUTF() - try { - Json.decodeFromString(serverResponseJson) - } catch (_: Exception) { - } - - output.writeBoolean(true) + dout.writeUTF(Json.encodeToString(TransferRequest.serializer(), request)) + val serverResponseJson = din.readUTF() + try { Json.decodeFromString(serverResponseJson) } catch (_: Exception) {} + dout.writeBoolean(true) + dout.flush() stateFlow.value = stateFlow.value.copy( state = TransferState.TRANSFERRING, @@ -249,20 +249,19 @@ class DesktopFileTransferManager : TransferManager { ) updateOutgoing(stateFlow.value) - val sha256 = MessageDigest.getInstance("SHA-256") + // File data transfer with raw streams (maximum speed) var bytesSent = 0L var lastProgressTime = System.currentTimeMillis() var lastProgressBytes = 0L - val buffer = ByteArray(262144) + val buffer = ByteArray(1048576) file.inputStream().use { fis -> while (true) { val read = fis.read(buffer) if (read == -1) break - output.write(buffer, 0, read) - sha256.update(buffer, 0, read) + rawOut.write(buffer, 0, read) bytesSent += read val now = System.currentTimeMillis() - if (now - lastProgressTime >= 500) { + if (now - lastProgressTime >= 1000) { stateFlow.value = stateFlow.value.copy( bytesTransferred = bytesSent, lastUpdateTimeMs = now, @@ -275,10 +274,21 @@ class DesktopFileTransferManager : TransferManager { } } } - output.flush() + rawOut.flush() - val serverChecksum = input.readUTF() - val bytesReceived = input.readLong() + // Compute checksum after transfer + val sha256 = MessageDigest.getInstance("SHA-256") + file.inputStream().use { fis -> + val buf = ByteArray(1048576) + while (true) { + val read = fis.read(buf) + if (read == -1) break + sha256.update(buf, 0, read) + } + } + val checksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}" + val serverChecksum = din.readUTF() + val bytesReceived = din.readLong() val localChecksum = "sha256:${sha256.digest().joinToString("") { "%02x".format(it) }}"