120 lines
3.3 KiB
Dart
120 lines
3.3 KiB
Dart
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<Map<String, dynamic>> connect({
|
||
required String agentType,
|
||
required String message,
|
||
String? conversationId,
|
||
String? imageUrl,
|
||
String? pdfUrl,
|
||
required String token,
|
||
}) {
|
||
final params = <String, String>{'message': message};
|
||
if (conversationId != null) {
|
||
params['conversationId'] = conversationId;
|
||
}
|
||
if (imageUrl != null && imageUrl.isNotEmpty) {
|
||
params['imageUrl'] = imageUrl;
|
||
}
|
||
if (pdfUrl != null && pdfUrl.isNotEmpty) {
|
||
params['pdfUrl'] = pdfUrl;
|
||
}
|
||
final query = params.entries
|
||
.map((e) => '${e.key}=${Uri.encodeComponent(e.value)}')
|
||
.join('&');
|
||
final url = '$baseUrl/api/ai/$agentType/chat?$query';
|
||
|
||
final cancelToken = CancelToken();
|
||
late final StreamController<Map<String, dynamic>> controller;
|
||
controller = StreamController<Map<String, dynamic>>(
|
||
onListen: () {
|
||
_connect(controller, url, token, cancelToken);
|
||
},
|
||
onCancel: () {
|
||
if (!cancelToken.isCancelled) {
|
||
cancelToken.cancel('用户停止生成');
|
||
}
|
||
},
|
||
);
|
||
return controller.stream;
|
||
}
|
||
|
||
static Future<void> _connect(
|
||
StreamController<Map<String, dynamic>> controller,
|
||
String url,
|
||
String token,
|
||
CancelToken cancelToken,
|
||
) async {
|
||
try {
|
||
final dio = Dio(
|
||
BaseOptions(
|
||
connectTimeout: const Duration(seconds: 15),
|
||
receiveTimeout: const Duration(minutes: 5),
|
||
),
|
||
);
|
||
|
||
final response = await dio.get(
|
||
url,
|
||
cancelToken: cancelToken,
|
||
options: Options(
|
||
responseType: ResponseType.stream,
|
||
headers: {
|
||
'Authorization': 'Bearer $token',
|
||
'X-App-Policy': appPolicyProfile,
|
||
},
|
||
),
|
||
);
|
||
|
||
final stream = response.data.stream as Stream<List<int>>;
|
||
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<String, dynamic>);
|
||
} catch (_) {
|
||
// 跳过无法解析的行
|
||
}
|
||
}
|
||
}
|
||
controller.close();
|
||
} on DioException catch (e) {
|
||
if (CancelToken.isCancel(e)) {
|
||
if (!controller.isClosed) await controller.close();
|
||
return;
|
||
}
|
||
if (!controller.isClosed) {
|
||
controller.add({'action': 'error', 'message': e.toString()});
|
||
await controller.close();
|
||
}
|
||
} catch (e) {
|
||
if (!controller.isClosed) {
|
||
controller.add({'action': 'error', 'message': e.toString()});
|
||
await controller.close();
|
||
}
|
||
}
|
||
}
|
||
}
|