From 8424b45b121d3b0132234109a953516a42f98586 Mon Sep 17 00:00:00 2001 From: Ali Alavi Date: Tue, 25 Aug 2026 18:29:23 +0000 Subject: [PATCH] fix(network): eliminate duplicate websocket connections and add dual-endpoint automatic failover --- .../soniox/remotemic/StreamDictationClient.kt | 101 ++++++++++++------ 1 file changed, 68 insertions(+), 33 deletions(-) diff --git a/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt b/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt index 44e48ac..0e52454 100644 --- a/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt +++ b/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt @@ -44,20 +44,28 @@ class StreamDictationClient( private val isRecording = AtomicBoolean(false) private val mainHandler = Handler(Looper.getMainLooper()) + private val endpoints = listOf( + host.removePrefix("http://").removePrefix("https://").removePrefix("ws://").removePrefix("wss://"), + "116.16.16.19:8999" + ).distinct() + private var currentEndpointIndex = 0 + private val okHttpClient = OkHttpClient.Builder() - .connectTimeout(10, TimeUnit.SECONDS) + .connectTimeout(4, TimeUnit.SECONDS) .readTimeout(0, TimeUnit.MILLISECONDS) // Persistent WebSocket - .writeTimeout(15, TimeUnit.SECONDS) - .pingInterval(25, TimeUnit.SECONDS) + .writeTimeout(10, TimeUnit.SECONDS) + .pingInterval(20, TimeUnit.SECONDS) .retryOnConnectionFailure(true) .build() private var webSocket: WebSocket? = null private val isConnected = AtomicBoolean(false) + private val isConnecting = AtomicBoolean(false) private var currentSessionId: String = "" private var isSessionActive = AtomicBoolean(false) private var lastPartialText: String = "" private var finalizeSafetyTimeoutRunnable: Runnable? = null + private var reconnectRunnable: Runnable? = null init { connectWebSocket() @@ -65,17 +73,25 @@ class StreamDictationClient( @Synchronized fun connectWebSocket() { - if (isConnected.get() && webSocket != null) return + if (isConnected.get() || isConnecting.get()) return + isConnecting.set(true) + + // Cancel previous socket before new attempt + try { + webSocket?.cancel() + } catch (e: Exception) {} + webSocket = null - val cleanHost = host.removePrefix("http://").removePrefix("https://").removePrefix("ws://").removePrefix("wss://") - val wsUrl = "ws://$cleanHost/ws/stream" + val currentHost = endpoints[currentEndpointIndex % endpoints.size] + val wsUrl = "ws://$currentHost/ws/stream" AppLogger.log(tag, "اتصال به سوکت همگام‌سازی بلادرنگ: $wsUrl") val req = Request.Builder().url(wsUrl).build() webSocket = okHttpClient.newWebSocket(req, object : WebSocketListener() { override fun onOpen(ws: WebSocket, response: Response) { - AppLogger.log(tag, "🟢 سوکت همگام‌سازی بلادرنگ متصل شد") + isConnecting.set(false) isConnected.set(true) + AppLogger.log(tag, "🟢 سوکت همگام‌سازی بلادرنگ متصل شد ($currentHost)") mainHandler.post { onConnectionStateChanged(true) } } @@ -92,7 +108,7 @@ class StreamDictationClient( val sel = json.optInt("selection", 0) val rev = json.optLong("revision", 0L) val ts = json.optDouble("timestamp", System.currentTimeMillis() / 1000.0) - + val payload = SyncStatePayload(source, app, txt, cursor, sel, rev, ts) mainHandler.post { onSyncStateReceived(payload) } return @@ -136,9 +152,13 @@ class StreamDictationClient( override fun onFailure(ws: WebSocket, t: Throwable, response: Response?) { AppLogger.log(tag, "🔴 قطع سوکت: ${t.message}. تلاش مجدد...") + isConnecting.set(false) isConnected.set(false) webSocket = null - + + // Cycle endpoint on failure + currentEndpointIndex = (currentEndpointIndex + 1) % endpoints.size + // If a speech session was in-flight, salvage the latest live transcript if (isSessionActive.get() && lastPartialText.isNotEmpty()) { isSessionActive.set(false) @@ -146,26 +166,35 @@ class StreamDictationClient( AppLogger.log(tag, "⚡ بازیابی خودکار متن زنده پس از قطعی: '$lastPartialText'") mainHandler.post { onSpeechCompleted(lastPartialText) } } - + mainHandler.post { onConnectionStateChanged(false) } - mainHandler.postDelayed({ connectWebSocket() }, 2000) + scheduleReconnect(1500) } override fun onClosed(ws: WebSocket, code: Int, reason: String) { AppLogger.log(tag, "سوکت بسته شد ($code). تلاش مجدد...") + isConnecting.set(false) isConnected.set(false) webSocket = null mainHandler.post { onConnectionStateChanged(false) } - mainHandler.postDelayed({ connectWebSocket() }, 2000) + scheduleReconnect(1500) } }) } + private fun scheduleReconnect(delayMs: Long) { + reconnectRunnable?.let { mainHandler.removeCallbacks(it) } + reconnectRunnable = Runnable { + connectWebSocket() + } + mainHandler.postDelayed(reconnectRunnable!!, delayMs) + } + /** * Broadcasts phone's updated text & cursor to Mac in sub-15ms */ fun sendLocalSyncState(text: String, cursor: Int, selection: Int = 0) { - if (!isConnected.get() || webSocket == null) { + if (!isConnected.get() && !isConnecting.get()) { connectWebSocket() } val payload = JSONObject().apply { @@ -182,42 +211,39 @@ class StreamDictationClient( @SuppressLint("MissingPermission") fun startRecording(cursorPos: Int): String { finalizeSafetyTimeoutRunnable?.let { mainHandler.removeCallbacks(it) } - + // Clean up previous recording thread if still finishing if (isRecording.get()) { isRecording.set(false) try { - audioRecord?.stop() - audioRecord?.release() - audioRecord = null - recordingThread?.join(150) - recordingThread = null + recordingThread?.join(200) } catch (e: Exception) {} } currentSessionId = "sess_${System.currentTimeMillis()}_${UUID.randomUUID().toString().take(6)}" - lastPartialText = "" - isRecording.set(true) isSessionActive.set(true) + lastPartialText = "" - if (!isConnected.get() || webSocket == null) { + if (!isConnected.get()) { connectWebSocket() } + // Send start control frame val startFrame = JSONObject().apply { put("type", "start") put("session_id", currentSessionId) put("cursor_pos", cursorPos) }.toString() webSocket?.send(startFrame) - AppLogger.log(tag, "🎙️ شروع استریم گفتار در موقعیت نشانگر $cursorPos...") - val minBufferSize = AudioRecord.getMinBufferSize(sampleRate, channelConfig, audioFormat) - val bufferSize = maxOf(minBufferSize, 2048) + AppLogger.log(tag, "🎙️ شروع استریم گفتار در موقعیت نشانگر $cursorPos...") try { + val minBufSize = AudioRecord.getMinBufferSize(sampleRate, channelConfig, audioFormat) + val bufferSize = maxOf(minBufSize * 2, 4096) + audioRecord = AudioRecord( - MediaRecorder.AudioSource.MIC, + MediaRecorder.AudioSource.VOICE_RECOGNITION, sampleRate, channelConfig, audioFormat, @@ -226,17 +252,17 @@ class StreamDictationClient( if (audioRecord?.state != AudioRecord.STATE_INITIALIZED) { AppLogger.log(tag, "❌ سخت‌افزار میکروفون راه‌اندازی نشد") - onError("خطا در راه‌اندازی سخت‌افزار میکروفون") - isRecording.set(false) + onError("خطای سخت‌افزار میکروفون") isSessionActive.set(false) return currentSessionId } audioRecord?.startRecording() + isRecording.set(true) recordingThread = Thread { - val chunk = ByteArray(2048) - val shortBuffer = ShortArray(1024) + val chunk = ByteArray(1024) + val shortBuffer = ShortArray(512) while (isRecording.get()) { val bytesRead = audioRecord?.read(chunk, 0, chunk.size) ?: -1 @@ -273,7 +299,7 @@ class StreamDictationClient( fun stopRecording(cursorPos: Int, tailExtensionMs: Long = 700L) { if (!isRecording.get()) return - + val sessionId = currentSessionId AppLogger.log(tag, "⏹️ دکمه رها شد. ضبط ${tailExtensionMs}ms دنباله صدا جهت ثبت کامل کلمات...") @@ -330,11 +356,20 @@ class StreamDictationClient( } fun release() { - isRecording.set(false) + reconnectRunnable?.let { mainHandler.removeCallbacks(it) } finalizeSafetyTimeoutRunnable?.let { mainHandler.removeCallbacks(it) } + isRecording.set(false) try { + audioRecord?.stop() audioRecord?.release() - webSocket?.close(1000, "Client Shutdown") + audioRecord = null + } catch (e: Exception) {} + try { + webSocket?.close(1000, "Client released") + webSocket?.cancel() } catch (e: Exception) {} + webSocket = null + isConnected.set(false) + isConnecting.set(false) } }