Browse Source

feat: resilient mobile websocket reconnection with exponential backoff and fast 350ms speech finalize

main
Ali Alavi 1 day ago
parent
commit
fd15d992c4
  1. 33
      android/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt
  2. 18
      server/relay_server.py

33
android/app/src/main/java/com/soniox/remotemic/StreamDictationClient.kt

@ -45,17 +45,19 @@ class StreamDictationClient(
private val mainHandler = Handler(Looper.getMainLooper()) private val mainHandler = Handler(Looper.getMainLooper())
private val okHttpClient = OkHttpClient.Builder() private val okHttpClient = OkHttpClient.Builder()
.connectTimeout(5, TimeUnit.SECONDS)
.connectTimeout(12, TimeUnit.SECONDS)
.readTimeout(0, TimeUnit.MILLISECONDS) // Persistent WebSocket .readTimeout(0, TimeUnit.MILLISECONDS) // Persistent WebSocket
.writeTimeout(5, TimeUnit.SECONDS)
.pingInterval(10, TimeUnit.SECONDS)
.writeTimeout(8, TimeUnit.SECONDS)
.pingInterval(8, TimeUnit.SECONDS)
.retryOnConnectionFailure(true) .retryOnConnectionFailure(true)
.build() .build()
private var activeWebSocket: WebSocket? = null private var activeWebSocket: WebSocket? = null
private val isConnected = AtomicBoolean(false) private val isConnected = AtomicBoolean(false)
private val isConnecting = AtomicBoolean(false)
private var currentSessionId: String = "" private var currentSessionId: String = ""
private var isSessionActive = AtomicBoolean(false) private var isSessionActive = AtomicBoolean(false)
private var retryAttempt = 0
init { init {
connectWebSocket() connectWebSocket()
@ -64,16 +66,23 @@ class StreamDictationClient(
@Synchronized @Synchronized
fun connectWebSocket() { fun connectWebSocket() {
if (isConnected.get() && activeWebSocket != null) return if (isConnected.get() && activeWebSocket != null) return
if (isConnecting.getAndSet(true)) return
val cleanHost = host.removePrefix("http://").removePrefix("https://").removePrefix("ws://").removePrefix("wss://") val cleanHost = host.removePrefix("http://").removePrefix("https://").removePrefix("ws://").removePrefix("wss://")
val wsUrl = "ws://$cleanHost/ws/stream" val wsUrl = "ws://$cleanHost/ws/stream"
AppLogger.log(tag, "اتصال به سوکت همگام‌سازی بلادرنگ: $wsUrl") AppLogger.log(tag, "اتصال به سوکت همگام‌سازی بلادرنگ: $wsUrl")
val req = Request.Builder().url(wsUrl).build()
val req = Request.Builder()
.url(wsUrl)
.header("User-Agent", "SonioxAndroidRemote/5.2")
.build()
activeWebSocket = okHttpClient.newWebSocket(req, object : WebSocketListener() { activeWebSocket = okHttpClient.newWebSocket(req, object : WebSocketListener() {
override fun onOpen(webSocket: WebSocket, response: Response) { override fun onOpen(webSocket: WebSocket, response: Response) {
AppLogger.log(tag, "🟢 سوکت همگام‌سازی بلادرنگ متصل شد") AppLogger.log(tag, "🟢 سوکت همگام‌سازی بلادرنگ متصل شد")
isConnected.set(true) isConnected.set(true)
isConnecting.set(false)
retryAttempt = 0
mainHandler.post { onConnectionStateChanged(true) } mainHandler.post { onConnectionStateChanged(true) }
} }
@ -123,18 +132,24 @@ class StreamDictationClient(
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) { override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
AppLogger.log(tag, "🔴 قطع سوکت: ${t.message}. تلاش مجدد...") AppLogger.log(tag, "🔴 قطع سوکت: ${t.message}. تلاش مجدد...")
isConnected.set(false)
activeWebSocket = null
mainHandler.post { onConnectionStateChanged(false) }
mainHandler.postDelayed({ connectWebSocket() }, 2000)
handleDisconnect()
} }
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) { override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
AppLogger.log(tag, "سوکت بسته شد ($code). تلاش مجدد...") AppLogger.log(tag, "سوکت بسته شد ($code). تلاش مجدد...")
handleDisconnect()
}
private fun handleDisconnect() {
isConnected.set(false) isConnected.set(false)
isConnecting.set(false)
activeWebSocket = null activeWebSocket = null
mainHandler.post { onConnectionStateChanged(false) } mainHandler.post { onConnectionStateChanged(false) }
mainHandler.postDelayed({ connectWebSocket() }, 2000)
// Exponential backoff: 800ms, 1600ms, 3200ms
retryAttempt++
val delayMs = minOf(800L * (1L shl minOf(retryAttempt, 3)), 5000L)
mainHandler.postDelayed({ connectWebSocket() }, delayMs)
} }
}) })
} }

18
server/relay_server.py

@ -1,9 +1,10 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
Soniox Collaborative Bi-Directional Input Synchronization Gateway (v5.0)
Soniox Collaborative Bi-Directional Input Synchronization Gateway (v5.2)
- Real-time Google Docs / Figma style collaborative mirroring between Mac and Android. - Real-time Google Docs / Figma style collaborative mirroring between Mac and Android.
- Ultra-low latency streaming speech recognition with pre-warmed Soniox pool.
- Monotonic revision counters, echo-loop suppression, and sub-15ms WebSocket routing. - Monotonic revision counters, echo-loop suppression, and sub-15ms WebSocket routing.
- Cursor-aware voice insertion and instant text editing.
- Process-targeted cursor-aware voice insertion.
""" """
import asyncio import asyncio
@ -59,7 +60,7 @@ def is_ws_open(ws) -> bool:
class SonioxPool: class SonioxPool:
"""Pre-warms upstream WebSockets to Soniox for 0ms speech start delay.""" """Pre-warms upstream WebSockets to Soniox for 0ms speech start delay."""
def __init__(self, size=3):
def __init__(self, size=4):
self._pool = asyncio.Queue(maxsize=size) self._pool = asyncio.Queue(maxsize=size)
self._refilling = False self._refilling = False
@ -83,7 +84,7 @@ class SonioxPool:
SONIOX_WS_URL, SONIOX_WS_URL,
additional_headers=SONIOX_HEADERS, additional_headers=SONIOX_HEADERS,
open_timeout=3.5, open_timeout=3.5,
ping_interval=20,
ping_interval=15,
) )
except Exception as e: except Exception as e:
logger.error("Failed to connect to upstream Soniox: %s", e) logger.error("Failed to connect to upstream Soniox: %s", e)
@ -156,7 +157,7 @@ async def broadcast_state(payload_dict: dict, exclude_ws=None):
async def handle_phone_stream_ws(request): async def handle_phone_stream_ws(request):
"""Duplex WebSocket for Android client (Real-time Live Sync + Audio Dictation).""" """Duplex WebSocket for Android client (Real-time Live Sync + Audio Dictation)."""
ws = web.WebSocketResponse(heartbeat=12.0)
ws = web.WebSocketResponse(heartbeat=10.0, autoping=True)
await ws.prepare(request) await ws.prepare(request)
client_ip = request.remote client_ip = request.remote
logger.info("📱 Android Client Connected: %s", client_ip) logger.info("📱 Android Client Connected: %s", client_ip)
@ -264,7 +265,8 @@ async def handle_phone_stream_ws(request):
if active_soniox_ws and is_ws_open(active_soniox_ws): if active_soniox_ws and is_ws_open(active_soniox_ws):
await active_soniox_ws.send(json.dumps({"type": "finalize"})) await active_soniox_ws.send(json.dumps({"type": "finalize"}))
try: try:
await asyncio.wait_for(stop_event.wait(), timeout=0.55)
# 350ms fast timeout for finalize
await asyncio.wait_for(stop_event.wait(), timeout=0.35)
except asyncio.TimeoutError: except asyncio.TimeoutError:
pass pass
@ -311,7 +313,7 @@ async def handle_phone_stream_ws(request):
async def handle_mac_ws(request): async def handle_mac_ws(request):
"""Persistent WebSocket for Mac Bridge (receives live Mac typing & pushes edits).""" """Persistent WebSocket for Mac Bridge (receives live Mac typing & pushes edits)."""
ws = web.WebSocketResponse(heartbeat=10.0)
ws = web.WebSocketResponse(heartbeat=10.0, autoping=True)
await ws.prepare(request) await ws.prepare(request)
client_ip = request.remote client_ip = request.remote
logger.info("🖥️ Mac client connected: %s", client_ip) logger.info("🖥️ Mac client connected: %s", client_ip)
@ -348,7 +350,7 @@ async def handle_health(request):
state_copy = dict(current_room_state) state_copy = dict(current_room_state)
return web.json_response({ return web.json_response({
"status": "ok", "status": "ok",
"service": "Soniox Collaborative Sync Gateway v5.0",
"service": "Soniox Collaborative Sync Gateway v5.2",
"connected_macs": len(connected_mac_websockets), "connected_macs": len(connected_mac_websockets),
"connected_phones": len(connected_phone_websockets), "connected_phones": len(connected_phone_websockets),
"current_app": state_copy.get("app", ""), "current_app": state_copy.get("app", ""),

Loading…
Cancel
Save