class SSEClient {
static http.Client _client = http.Client();
static StreamController<SSEModel>? _streamController;
static StreamSubscription? _subscription;
static StreamSubscription? _dataSubscription;
static Stream<SSEModel> subscribeToSSE()
static Future<void> unsubscribeFromSSE() async {}
}
dart
_streamController = StreamController();
_client = http.Client();
final request = http.Request(
method == 'GET'
Uri.parse(url),
);
var currentSSEModel = SSEModel(data: '', id: '', event: '');
dart
final headers = <String, String>{
'Authorization': 'Bearer ${token.accessToken}',
'Accept': 'text/event-stream',
'Cache-Control': 'no-cache',
};
header.forEach((key, value) {
request.headers[key] = value;
});
dart
Future<http.StreamedResponse> response = _client.send(request);
_subscription = response.asStream().listen((data) {}
dart
_dataSubscription = data.stream.transform(const Utf8Decoder()).transform(const LineSplitter()).listen((dataLine) {})
if (dataLine.isEmpty) {
_streamController!.add(currentSSEModel);
currentSSEModel = SSEModel(data: '', id: '', event: '');
return;
}
dart
final lineRegex = RegExp(r'^([^:]*)(?::)?(?: )?(.*)?$');
final match = lineRegex.firstMatch(dataLine)!;
final field = match.group(1);
if (field!.isEmpty) {
return;
}
var value = '';
if (field == 'data') {
value = dataLine.substring(
5,
);
} else {
value = match.group(2) ?? '';
}
switch (field) {
case 'event':
currentSSEModel.event = value;
case 'data':
currentSSEModel.data = '${currentSSEModel.data ?? ''}$value\n';
case 'id':
currentSSEModel.id = value;
case 'retry':
break;
}
return _streamController!.stream;