import 'dart:async'; import 'dart:convert'; import 'package:dio/dio.dart'; import '../core/api_client.dart'; /// 跨平台 SSE 流处理(基于 Dio 流式响应,支持 Android/iOS/Web) class SseHandler { /// 连接 SSE 端点,返回事件流 static Stream> connect({ required String agentType, required String message, String? conversationId, required String token, }) { final params = { 'message': message, 'token': token, }; if (conversationId != null) { params['conversationId'] = conversationId; } final query = params.entries .map((e) => '${e.key}=${Uri.encodeComponent(e.value)}') .join('&'); final url = '$baseUrl/api/ai/$agentType/chat?$query'; final controller = StreamController>(); _connect(controller, url); return controller.stream; } static Future _connect( StreamController> controller, String url, ) async { try { final dio = Dio(BaseOptions( connectTimeout: const Duration(seconds: 15), receiveTimeout: const Duration(minutes: 5), )); final response = await dio.get( url, options: Options(responseType: ResponseType.stream), ); final stream = response.data.stream as Stream>; var buffer = ''; await for (final chunk in stream) { if (controller.isClosed) break; final text = utf8.decode(chunk, allowMalformed: true); buffer += text; // 按行解析 SSE 数据 while (buffer.contains('\n')) { final newlineIdx = buffer.indexOf('\n'); var line = buffer.substring(0, newlineIdx).trim(); buffer = buffer.substring(newlineIdx + 1); if (line.isEmpty || !line.startsWith('data: ')) continue; final data = line.substring(6); if (data == '[DONE]') { controller.close(); return; } try { controller.add(jsonDecode(data) as Map); } catch (_) { // 跳过无法解析的行 } } } controller.close(); } catch (e) { if (!controller.isClosed) { controller.add({'action': 'error', 'message': e.toString()}); controller.close(); } } } }