More casting fixes.

app/src/main/java/com/futo/platformplayer/casting/FCastCastingDevice.kt

@@ -15,6 +15,7 @@ import com.futo.platformplayer.casting.models.FCastSetSpeedMessage
 import com.futo.platformplayer.casting.models.FCastSetVolumeMessage
 import com.futo.platformplayer.casting.models.FCastVersionMessage
 import com.futo.platformplayer.casting.models.FCastVolumeUpdateMessage
+import com.futo.platformplayer.ensureNotMainThread
 import com.futo.platformplayer.getConnectedSocket
 import com.futo.platformplayer.logging.Logger
 import com.futo.platformplayer.models.CastingDeviceInfo
@@ -90,6 +91,7 @@ class FCastCastingDevice : CastingDevice {
     private var _thread: Thread? = null
     private var _pingThread: Thread? = null
     private var _lastPongTime = -1L
+    private var _outputStreamLock = Object()
 
     constructor(name: String, addresses: Array<InetAddress>, port: Int) : super() {
         this.name = name;
@@ -297,6 +299,8 @@ class FCastCastingDevice : CastingDevice {
 
                     try {
                         _socket?.close()
+                        _inputStream?.close()
+                        _outputStream?.close()
                         if (connectedSocket != null) {
                             Logger.i(TAG, "Using connected socket.");
                             _socket = connectedSocket
@@ -310,7 +314,9 @@ class FCastCastingDevice : CastingDevice {
                         _outputStream = _socket?.outputStream;
                         _inputStream = _socket?.inputStream;
                     } catch (e: IOException) {
-                        _socket?.close();
+                        _socket?.close()
+                        _inputStream?.close()
+                        _outputStream?.close()
                         Logger.i(TAG, "Failed to connect to FastCast.", e);
 
                         connectionState = CastConnectionState.CONNECTING;
@@ -332,7 +338,10 @@ class FCastCastingDevice : CastingDevice {
 
                             var headerBytesRead = 0
                             while (headerBytesRead < 4) {
-                                headerBytesRead += inputStream.read(buffer, headerBytesRead, 4 - headerBytesRead)
+                                val read = inputStream.read(buffer, headerBytesRead, 4 - headerBytesRead)
+                                if (read == -1)
+                                    throw Exception("Stream closed")
+                                headerBytesRead += read
                             }
 
                             val size = ((buffer[3].toLong() shl 24) or (buffer[2].toLong() shl 16) or (buffer[1].toLong() shl 8) or buffer[0].toLong()).toInt();
@@ -344,7 +353,10 @@ class FCastCastingDevice : CastingDevice {
                             Log.d(TAG, "Received header indicating $size bytes. Waiting for message.");
                             var bytesRead = 0
                             while (bytesRead < size) {
-                                bytesRead += inputStream.read(buffer, bytesRead, size - bytesRead)
+                                val read = inputStream.read(buffer, bytesRead, size - bytesRead)
+                                if (read == -1)
+                                    throw Exception("Stream closed")
+                                bytesRead += read
                             }
 
                             val messageBytes = buffer.sliceArray(IntRange(0, size));
@@ -373,7 +385,8 @@ class FCastCastingDevice : CastingDevice {
 
                     try {
                         _socket?.close()
-                        _socket = null
+                        _inputStream?.close()
+                        _outputStream?.close()
                         Logger.i(TAG, "Socket disconnected.");
                     } catch (e: Throwable) {
                         Logger.e(TAG, "Failed to close socket.", e)
@@ -398,12 +411,14 @@ class FCastCastingDevice : CastingDevice {
 
                         try {
                             _socket?.close()
+                            _inputStream?.close()
+                            _outputStream?.close()
                         } catch (e: Throwable) {
                             Log.w(TAG, "Failed to close socket.", e)
                         }
                     }
 
-                    if (_lastPongTime != -1L && System.currentTimeMillis() - _lastPongTime > 6000) {
+                    /*if (_lastPongTime != -1L && System.currentTimeMillis() - _lastPongTime > 6000) {
                         Logger.w(TAG, "Closing socket due to last pong time being larger than 6 seconds.")
 
                         try {
@@ -411,7 +426,7 @@ class FCastCastingDevice : CastingDevice {
                         } catch (e: Throwable) {
                             Log.w(TAG, "Failed to close socket.", e)
                         }
-                    }
+                    }*/
 
                     Thread.sleep(2000)
                 }
@@ -476,34 +491,38 @@ class FCastCastingDevice : CastingDevice {
     }
 
     private fun send(opcode: Opcode, message: String? = null) {
-        try {
-            val data: ByteArray = message?.encodeToByteArray() ?: ByteArray(0)
-            val size = 1 + data.size
-            val outputStream = _outputStream
-            if (outputStream == null) {
-                Log.w(TAG, "Failed to send $size bytes, output stream is null.")
-                return
+        ensureNotMainThread()
+
+        synchronized (_outputStreamLock) {
+            try {
+                val data: ByteArray = message?.encodeToByteArray() ?: ByteArray(0)
+                val size = 1 + data.size
+                val outputStream = _outputStream
+                if (outputStream == null) {
+                    Log.w(TAG, "Failed to send $size bytes, output stream is null.")
+                    return
+                }
+
+                val serializedSizeLE = ByteArray(4)
+                serializedSizeLE[0] = (size and 0xff).toByte()
+                serializedSizeLE[1] = (size shr 8 and 0xff).toByte()
+                serializedSizeLE[2] = (size shr 16 and 0xff).toByte()
+                serializedSizeLE[3] = (size shr 24 and 0xff).toByte()
+                outputStream.write(serializedSizeLE)
+
+                val opcodeBytes = ByteArray(1)
+                opcodeBytes[0] = opcode.value
+                outputStream.write(opcodeBytes)
+
+                if (data.isNotEmpty()) {
+                    outputStream.write(data)
+                }
+
+                Log.d(TAG, "Sent $size bytes: (opcode: $opcode, body: $message).")
+            } catch (e: Throwable) {
+                Log.i(TAG, "Failed to send message.", e)
+                throw e
             }
-
-            val serializedSizeLE = ByteArray(4)
-            serializedSizeLE[0] = (size and 0xff).toByte()
-            serializedSizeLE[1] = (size shr 8 and 0xff).toByte()
-            serializedSizeLE[2] = (size shr 16 and 0xff).toByte()
-            serializedSizeLE[3] = (size shr 24 and 0xff).toByte()
-            outputStream.write(serializedSizeLE)
-
-            val opcodeBytes = ByteArray(1)
-            opcodeBytes[0] = opcode.value
-            outputStream.write(opcodeBytes)
-
-            if (data.isNotEmpty()) {
-                outputStream.write(data)
-            }
-
-            Log.d(TAG, "Sent $size bytes: (opcode: $opcode, body: $message).")
-        } catch (e: Throwable) {
-            Log.i(TAG, "Failed to send message.", e)
-            throw e
         }
     }
 
@@ -533,6 +552,8 @@ class FCastCastingDevice : CastingDevice {
 
             scopeIO.launch {
                 socket.close();
+                _inputStream?.close()
+                _outputStream?.close()
                 connectionState = CastConnectionState.DISCONNECTED;
                 scopeIO.cancel();
                 Logger.i(TAG, "Cancelled scopeIO with open socket.")