From 994acccc9d2017c817805ee9dda5800de5095085 Mon Sep 17 00:00:00 2001 From: Sikander Saleem Date: Wed, 21 Oct 2026 15:13:04 +0300 Subject: [PATCH] improvements --- .../copilot/controller/copilot_provider.dart | 29 +++- .../utils/ptt_signalr_http_client.dart | 164 ++++++++++++++++++ 2 files changed, 186 insertions(+), 7 deletions(-) create mode 100644 lib/modules/copilot/utils/ptt_signalr_http_client.dart diff --git a/lib/modules/copilot/controller/copilot_provider.dart b/lib/modules/copilot/controller/copilot_provider.dart index b74ed940..59a87860 100644 --- a/lib/modules/copilot/controller/copilot_provider.dart +++ b/lib/modules/copilot/controller/copilot_provider.dart @@ -13,7 +13,7 @@ import 'package:test_sa/controllers/api_routes/api_manager.dart'; import 'package:test_sa/controllers/api_routes/urls.dart'; import '../model/copilot_message_model.dart'; -import '../utils/resilient_signalr_http_client.dart'; +import '../utils/ptt_signalr_http_client.dart'; enum PttStatus { idle, connecting, recording, thinking, playingAudio, error } @@ -32,6 +32,7 @@ class CopilotProvider extends ChangeNotifier { StreamSubscription? _streamSub; HubConnection? _pttHub; + PttSignalRHttpClient? _pttHttpClient; // Serializes connect/disconnect so they never overlap when the user navigates // in and out of the Push-to-Talk screen quickly. Future _pttOpQueue = Future.value(); @@ -376,6 +377,8 @@ class CopilotProvider extends ChangeNotifier { final token = ApiManager.instance.user?.token ?? ''; final url = "${URLs.copilotHubUrl}?access_token=$token"; _pttLog('connectPushToTalkHub: building connection url=${URLs.copilotHubUrl}'); + final httpClient = PttSignalRHttpClient(); + _pttHttpClient = httpClient; final httpOp = HttpConnectionOptions( // Server only accepts LongPolling. transport: HttpTransportType.LongPolling, @@ -383,9 +386,12 @@ class CopilotProvider extends ChangeNotifier { // Never log message content: it contains base64 audio and is very heavy. logMessageContent: false, headers: MessageHeaders()..setHeaderValue('Authorization', 'Bearer $token'), - // Keeps the long-poll alive across poll timeouts / network blips instead - // of closing the whole connection (signalr_netcore bug workaround). - httpClient: ResilientSignalRHttpClient(), + // Ensures the Authorization header is also sent on poll + the DELETE + // that tells the server to drop the connection when we stop. + accessTokenFactory: () async => token, + // Closes sockets properly and truly aborts the pending long-poll on stop + // (signalr_netcore's default client leaks them). + httpClient: httpClient, ); final hub = HubConnectionBuilder().withUrl(url, options: httpOp).withAutomaticReconnect(retryDelays: [0, 2000, 5000, 10000, 20000]).build(); // More tolerant timeouts for mobile networks (defaults are 30s / 15s). @@ -475,6 +481,8 @@ class CopilotProvider extends ChangeNotifier { // Detach first so callbacks (onclose) from this hub are ignored. final hub = _pttHub; _pttHub = null; + final httpClient = _pttHttpClient; + _pttHttpClient = null; pttHubState = PttHubState.disconnected; try { await _abortMicStreaming(); @@ -485,6 +493,10 @@ class CopilotProvider extends ChangeNotifier { } } catch (e) { _pttLog('_stopHubAndCleanup: error=$e'); + } finally { + // Kill any request still open (e.g. a pending long-poll) so no orphan + // connections pile up between visits to the screen. + httpClient?.closeAll(); } } @@ -635,11 +647,12 @@ class CopilotProvider extends ChangeNotifier { _pttButtonHeld = false; liveSessionActive = false; _cancelResponseWatchdog(); + final hadActiveTurn = pttStatus == PttStatus.connecting || pttStatus == PttStatus.recording || pttStatus == PttStatus.thinking; try { await _abortMicStreaming(); - if (_pttHub != null && _pttHub!.state == HubConnectionState.Connected) { + if (hadActiveTurn && _pttHub != null && _pttHub!.state == HubConnectionState.Connected) { _pttLog('cancelPushToTalk: invoking CancelRecording'); - await _pttHub!.invoke("CancelRecording"); + await _pttHub!.invoke("CancelRecording").timeout(const Duration(seconds: 5)); } } catch (e) { _pttLog('cancelPushToTalk: error=$e'); @@ -877,7 +890,9 @@ class CopilotProvider extends ChangeNotifier { } final hub = _pttHub; _pttHub = null; - hub?.stop(); + final httpClient = _pttHttpClient; + _pttHttpClient = null; + hub?.stop().whenComplete(() => httpClient?.closeAll()); super.dispose(); } } diff --git a/lib/modules/copilot/utils/ptt_signalr_http_client.dart b/lib/modules/copilot/utils/ptt_signalr_http_client.dart new file mode 100644 index 00000000..5c1cb9d0 --- /dev/null +++ b/lib/modules/copilot/utils/ptt_signalr_http_client.dart @@ -0,0 +1,164 @@ +import 'dart:async'; +import 'dart:developer' as dev; +import 'dart:io'; + +import 'package:http/http.dart' as http; +import 'package:signalr_netcore/errors.dart'; +import 'package:signalr_netcore/ihub_protocol.dart'; +import 'package:signalr_netcore/signalr_client.dart'; + +/// SignalR HTTP client used for the push-to-talk LongPolling connection. +/// +/// Replaces signalr_netcore's [WebSupportingHttpClient], which has two bugs +/// that break LongPolling on mobile: +/// +/// 1. Socket leak / fake abort: it creates a new `http.Client` per request +/// and never closes it. When the hub stops, the pending long-poll is only +/// "aborted" by ignoring its result - the real HTTP socket stays open and +/// the server keeps treating it as an active poll. After a few visits to +/// the screen these orphan connections pile up and new connections hang on +/// "Connecting..." until the app is killed (which closes the sockets). +/// Here every request's client is closed when it completes, and aborting +/// closes the client, which really tears down the socket. [closeAll] kills +/// anything still open when the hub is disposed. +/// +/// 2. Poll timeouts close the connection: the transport only ignores its own +/// [TimeoutError], but the HTTP layer throws Dart's [TimeoutException] / +/// [SocketException] on a timeout or network blip, which closes the whole +/// connection. Those are converted to [TimeoutError] for poll (GET) +/// requests so the transport simply re-polls. If failures persist longer +/// than [maxContinuousFailure], the error is rethrown so automatic +/// reconnect takes over. +class PttSignalRHttpClient extends SignalRHttpClient { + PttSignalRHttpClient({ + this.maxContinuousFailure = const Duration(seconds: 30), + this.retryDelay = const Duration(seconds: 1), + }); + + final Duration maxContinuousFailure; + final Duration retryDelay; + + final Set _activeClients = {}; + DateTime? _firstFailureAt; + bool _closed = false; + + void _log(String msg) => dev.log('[PTT-HTTP] $msg'); + + /// Force-closes every in-flight request (including a pending long-poll). + void closeAll() { + _closed = true; + if (_activeClients.isNotEmpty) { + _log('closeAll: closing ${_activeClients.length} open request(s)'); + } + for (final c in _activeClients.toList()) { + try { + c.close(); + } catch (_) {} + } + _activeClients.clear(); + } + + @override + Future send(SignalRHttpRequest request) async { + final isPoll = (request.method ?? '').toUpperCase() == 'GET'; + try { + final response = await _send(request); + if (isPoll) _firstFailureAt = null; + return response; + } on TimeoutException { + if (!isPoll) rethrow; + // Normal long-poll timeout: just poll again. + _firstFailureAt = null; + throw TimeoutError('Long poll timed out'); + } catch (e) { + if (!isPoll || _closed) rethrow; + final transient = e is SocketException || e is http.ClientException || e is HttpException || e is HandshakeException; + if (!transient) rethrow; + + final now = DateTime.now(); + _firstFailureAt ??= now; + if (now.difference(_firstFailureAt!) > maxContinuousFailure) { + _firstFailureAt = null; + rethrow; // give up -> connection closes -> automatic reconnect + } + await Future.delayed(retryDelay); + throw TimeoutError('Transient poll failure: $e'); + } + } + + Future _send(SignalRHttpRequest request) async { + final abortSignal = request.abortSignal; + if (abortSignal != null && (abortSignal.aborted ?? false)) throw AbortError(); + if (_closed) throw AbortError('HTTP client closed'); + + final method = request.method; + final url = request.url; + if (method == null || method.isEmpty) throw ArgumentError('No method defined.'); + if (url == null || url.isEmpty) throw ArgumentError('No url defined.'); + + final uri = Uri.parse(url); + final client = http.Client(); + _activeClients.add(client); + + var aborted = false; + if (abortSignal != null) { + abortSignal.onabort = () { + aborted = true; + // Closing the client really tears down the underlying socket. + try { + client.close(); + } catch (_) {} + }; + } + + final content = request.content; + final isJson = content is String && content.startsWith('{'); + final headers = MessageHeaders() + ..setHeaderValue('X-Requested-With', 'FlutterHttpClient') + ..setHeaderValue('content-type', isJson ? 'application/json;charset=UTF-8' : 'text/plain;charset=UTF-8'); + if (request.headers != null) headers.addMessageHeaders(request.headers); + + try { + Future future; + switch (method.toLowerCase()) { + case 'post': + future = client.post(uri, body: content, headers: headers.asMap); + break; + case 'put': + future = client.put(uri, body: content, headers: headers.asMap); + break; + case 'delete': + future = client.delete(uri, body: content, headers: headers.asMap); + break; + case 'get': + default: + future = client.get(uri, headers: headers.asMap); + } + final timeoutMs = request.timeout; + if (timeoutMs != null && timeoutMs > 0) { + future = future.timeout(Duration(milliseconds: timeoutMs)); + } + + final http.Response resp; + try { + resp = await future; + } catch (_) { + if (aborted) throw AbortError(); + rethrow; + } + if (aborted) throw AbortError(); + + if (resp.statusCode >= 200 && resp.statusCode < 300) { + return SignalRHttpResponse(resp.statusCode, statusText: resp.reasonPhrase, content: resp.body); + } + throw HttpError(resp.reasonPhrase, resp.statusCode); + } finally { + if (abortSignal != null) abortSignal.onabort = null; + _activeClients.remove(client); + try { + client.close(); + } catch (_) {} + } + } +} +