Skip to content

feat: add autoReconnect #44

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
yanmingLiu wants to merge 2 commits into
base: master
Choose a base branch
from
Open

feat: add autoReconnect #44

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
50743eb
fix: add max retry
yanmingLiu Mar 11, 2025
22eda4d
fix: (streamController.addError('Max retry limit reached'))
yanmingLiu Mar 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
242 changes: 116 additions & 126 deletions lib/flutter_client_sse.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,157 +2,133 @@ library flutter_client_sse;

import 'dart:async';
import 'dart:convert';

import 'package:flutter_client_sse/constants/sse_request_type_enum.dart';
import 'package:http/http.dart' as http;

part 'sse_event_model.dart';

/// A client for subscribing to Server-Sent Events (SSE).
class SSEClient {
static http.Client _client = new http.Client();

/// Retry the SSE connection after a delay.
///
/// [method] is the request method (GET or POST).
/// [url] is the URL of the SSE endpoint.
/// [header] is a map of request headers.
/// [body] is an optional request body for POST requests.
/// [streamController] is required to persist the stream from the old connection
static void _retryConnection(
{required SSERequestType method,
required String url,
required Map<String, String> header,
required StreamController<SSEModel> streamController,
Map<String, dynamic>? body}) {
print('---RETRY CONNECTION---');
static http.Client _client = http.Client();

static void _retryConnection({
required SSERequestType method,
required String url,
required Map<String, String> header,
required StreamController<SSEModel> streamController,
Map<String, dynamic>? body,
int? retriesLeft,
required int maxRetries,
}) {
if (retriesLeft == null || retriesLeft <= 0) {
print('---MAX RETRY LIMIT REACHED---');
streamController.addError('Erro: Max retry limit reached');
streamController.close();
_client.close();
return;
}

print('---RETRY CONNECTION ($retriesLeft retries left)---');
Future.delayed(Duration(seconds: 5), () {
subscribeToSSE(
method: method,
url: url,
header: header,
body: body,
oldStreamController: streamController,
maxRetries: maxRetries,
retriesLeft: retriesLeft - 1,
);
});
}

/// Subscribe to Server-Sent Events.
///
/// [method] is the request method (GET or POST).
/// [url] is the URL of the SSE endpoint.
/// [header] is a map of request headers.
/// [body] is an optional request body for POST requests.
///
/// Returns a [Stream] of [SSEModel] representing the SSE events.
static Stream<SSEModel> subscribeToSSE(
{required SSERequestType method,
required String url,
required Map<String, String> header,
StreamController<SSEModel>? oldStreamController,
Map<String, dynamic>? body}) {
StreamController<SSEModel> streamController = StreamController();
if (oldStreamController != null) {
streamController = oldStreamController;
}
static Stream<SSEModel> subscribeToSSE({
required SSERequestType method,
required String url,
required Map<String, String> header,
StreamController<SSEModel>? oldStreamController,
Map<String, dynamic>? body,
int maxRetries = 5,
int? retriesLeft,
}) {
StreamController<SSEModel> streamController = oldStreamController ?? StreamController();
retriesLeft ??= maxRetries;

var lineRegex = RegExp(r'^([^:]*)(?::)?(?: )?(.*)?$');
var currentSSEModel = SSEModel(data: '', id: '', event: '');
print("--SUBSCRIBING TO SSE---");
while (true) {
try {
_client = http.Client();
var request = new http.Request(
method == SSERequestType.GET ? "GET" : "POST",
Uri.parse(url),
);

/// Adding headers to the request
header.forEach((key, value) {
request.headers[key] = value;
});

/// Adding body to the request if exists
if (body != null) {
request.body = jsonEncode(body);
}

Future<http.StreamedResponse> response = _client.send(request);

/// Listening to the response as a stream
response.asStream().listen((data) {
/// Applying transforms and listening to it
data.stream
..transform(Utf8Decoder()).transform(LineSplitter()).listen(
(dataLine) {
if (dataLine.isEmpty) {
/// This means that the complete event set has been read.
/// We then add the event to the stream
streamController.add(currentSSEModel);
currentSSEModel = SSEModel(data: '', id: '', event: '');
return;
}

/// Get the match of each line through the regex
Match match = lineRegex.firstMatch(dataLine)!;
var field = match.group(1);
if (field!.isEmpty) {
return;
}
var value = '';
if (field == 'data') {
// If the field is data, we get the data through the substring
value = dataLine.substring(
5,
);
} else {
value = match.group(2) ?? '';
}
switch (field) {
case 'event':
currentSSEModel.event = value;
break;
case 'data':
currentSSEModel.data =
(currentSSEModel.data ?? '') + value + '\n';
break;
case 'id':
currentSSEModel.id = value;
break;
case 'retry':
break;
default:
print('---ERROR---');
print(dataLine);
_retryConnection(
method: method,
url: url,
header: header,
streamController: streamController,
);
}
},
onError: (e, s) {
try {
_client = http.Client();
var request = http.Request(
method == SSERequestType.GET ? "GET" : "POST",
Uri.parse(url),
);

header.forEach((key, value) {
request.headers[key] = value;
});

if (body != null) {
request.body = jsonEncode(body);
}

Future<http.StreamedResponse> response = _client.send(request);

response.asStream().listen((data) {
data.stream.transform(Utf8Decoder()).transform(LineSplitter()).listen(
(dataLine) {
if (dataLine.isEmpty) {
streamController.add(currentSSEModel);
currentSSEModel = SSEModel(data: '', id: '', event: '');
return;
}

Match? match = lineRegex.firstMatch(dataLine);
if (match == null) return;
var field = match.group(1);
if (field == null || field.isEmpty) return;
var value = match.group(2) ?? '';

switch (field) {
case 'event':
currentSSEModel.event = value;
break;
case 'data':
currentSSEModel.data = (currentSSEModel.data ?? '') + value + '\n';
break;
case 'id':
currentSSEModel.id = value;
break;
default:
print('---ERROR---');
print(e);
print(dataLine);
_retryConnection(
method: method,
url: url,
header: header,
body: body,
streamController: streamController,
body: body,
retriesLeft: retriesLeft,
maxRetries: maxRetries,
);
},
}
},
onError: (e, s) {
print('---ERROR---');
print(e);
_retryConnection(
method: method,
url: url,
header: header,
body: body,
streamController: streamController,
retriesLeft: retriesLeft,
maxRetries: maxRetries,
);
}, onError: (e, s) {
print('---ERROR---');
print(e);
_retryConnection(
method: method,
url: url,
header: header,
body: body,
streamController: streamController,
);
});
} catch (e) {
},
);
}, onError: (e, s) {
print('---ERROR---');
print(e);
_retryConnection(
Expand All @@ -161,13 +137,27 @@ class SSEClient {
header: header,
body: body,
streamController: streamController,
retriesLeft: retriesLeft,
maxRetries: maxRetries,
);
}
return streamController.stream;
});
} catch (e) {
print('---ERROR---');
print(e);
_retryConnection(
method: method,
url: url,
header: header,
body: body,
streamController: streamController,
retriesLeft: retriesLeft,
maxRetries: maxRetries,
);
}

return streamController.stream;
}

/// Unsubscribe from the SSE.
static void unsubscribeFromSSE() {
_client.close();
}
Expand Down