improvements

main_production_copilot
Sikander Saleem 1 day ago
parent 754bd0c939
commit 994acccc9d

@ -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<void> _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: <int>[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();
}
}

@ -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<http.Client> _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<SignalRHttpResponse> 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<SignalRHttpResponse> _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<http.Response> 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 (_) {}
}
}
}
Loading…
Cancel
Save