pratikbaid3/flutter_client_sse

Dart package to help consume SSE API. Consumes server sent events by returning parsed model of the event, id, and the data

Hey there!
I am Pratik Baid.

GSoC @CCExtractor | SIH 2020 Winner | SDE - 2 @Niyo | SAMSUNG PRISM DEVELOPER | Founder @Hustle Creatives | GSSoC Extended-2020 |

Connect with me on -

Frameworks and Stack

Languages

Tools

Some of my GitHub Stats

flutter_client_sse's Issues

Stream has already been listened to after 2.0.1 -> 2.0.2 update

In the app, I have one screen calling SSEClient.subscribeToSSE, consecutive screen calls different server endpoint, but it also uses the same library method and I am getting the following exception:

[ERROR:flutter/runtime/dart_vm_initializer.cc(41)] Unhandled Exception: Bad state: Stream has already been listened to.
#0      _StreamController._subscribe (dart:async/stream_controller.dart:686:7)
#1      _ControllerStream._createSubscription (dart:async/stream_controller.dart:837:19)
#2      _StreamImpl.listen (dart:async/stream_impl.dart:497:9)
#3      new _ForwardingStreamSubscription (dart:async/stream_pipe.dart:114:10)
#4      _ForwardingStream._createSubscription (dart:async/stream_pipe.dart:86:16)
#5      _ForwardingStream.listen (dart:async/stream_pipe.dart:81:12)
#6      new _ForwardingStreamSubscription (dart:async/stream_pipe.dart:114:10)
#7      _ForwardingStream._createSubscription (dart:async/stream_pipe.dart:86:16)
#8      _ForwardingStream.listen (dart:async/stream_pipe.dart:81:12)
#9      new _ForwardingStreamSubscription (dart:async/stream_pipe.dart:114:10)
#10     _ForwardingStream._createSubscription (dart:async/stream_pipe.dart:86:16)
#11     _ForwardingStream.listen (dart:asyn<…>

Error on unsubscribe

Hello @pratikbaid3,

If I subscribe on SSE webservice (via subscribeToSSE static method), then I unsubscribe (via unsubscribeFromSSE static method) and finally I re-subscribe, an error occured :

flutter: ---ERROR---
flutter: Connection closed while receiving data

I think the problem comes from this line of code (line 35) because the subscription is not killed in unsubscribeFromSSE method :
response.asStream().listen()

Thanks in advance.
Anthony

Browser support not working

Should the browser version wrap dart sse client (https://pub.dev/documentation/sse/latest/client_sse_client/SseClient-class.html) instead of using http client?

Token header

Hi,
The package is really useful.
However, I find it unfortunate that the token is sent in the "Cookie" header. Depending on the usage needs, we may need to place the token in the "Authorization" header, for example.
This aspect could be made configurable.
Have a good day,

Add `Accept` header by default

It looks like the listener is only able to consume data formatted as text/event-stream, since that's the case, is there a good reason not to automatically set the Accept header to that mimetype, since nothing else is actually accepted?

On changing the project architecture and fixing bugs

First off, thank you for your package.

Long story short: I found your package useful, but the current implementation doesn't suit my need, and I was in urgent need of implementing something with SSE. So I copied the code locally, hacked into it, and finished my job first. Now I would like to contribute my changes back.

After forking and trying to update the code into your original repo, I feel hesistated, because I have made substantial changes to the original library, and there is no backward compatibility. So I would like to know your take on this before continuing, or submitting a pull request.

Here is my updated repo:
https://github.com/tamcy/dart_sse_client/tree/tamcy-dev

Example can be found here (README has not been updated Update: now it's updated):
https://github.com/tamcy/dart_sse_client/blob/tamcy-dev/example/main.dart

It's not finished for a pull request, but the code should work.

The changes are aimed to:

1. resolve architectural issues,
2. fix bugs,
3. add test, and
4. improve coding style, like making the package structure and namings more "Dartish".

Here is the longer summary:

Fixing architectural issues

Currently, calling subscribeToSSE() twice will cause the original http.Client being lost, and it's not possible to close the connection of previous http clients (not to mentione there's no way to close a stream). Which is why I made the SSE client an instance class.

I see SseClient something similar to but not the same as an HTTP client. An HTTP client sends a request and returns a response, while an SSE "client" should act more like a wrapper class that encapsulates the details of an SSE connection. With this in mind I have moved the connection details to the constructor. In my design, an SseClient instance will only be responsible of managing the connection of a single request. User should create a new SseClient instance for a different requests.

Also, while an SSE client needs a http.Request object for http.Client, I think it should not be opinionated on how the request object is constructed and the body format (the current library is json-encoding it). So instead of asking for the request method, URL, header and body, the updated version now asks for an http.Request object. This also elminates the need for an SSERequestType enum.

Fixing bugs

I found that the current implementation didn't quite follow the specification on message parsing, especially on how the white spaces and line breaks are handled. Imagine when the message is in plain text and you need to compare the string data. This has been fixed in my fork.

My fork also checks for the response's Content-Type according to the spec:

if res's status is not 200, or if res's Content-Type is not text/event-stream, then fail the connection.
This may not be a bug, but I think it's still essential to have such checking. I once connected to a wrong path (actually the path is correct, but there is a server problem on vhost configuration), the response is wrong, but since the response code is 200, the original code just accepted the connection (despite the content type being text/html). Of course, there is no SSE event received whatsoever, and I spent quite some time to figure out the cause.

Coding style fixes

I have also updated the code to make it more Dart-like, at least as per my understanding. For instance, I have separate the published event object from the one that is used for buffering. To me, the buffering object is an internal representation which, while very similar to the dispatch event in structure, should not be exposed to users. In particular, the fields of the emitted message should be immutable.

I have also updated the CHANGELOG.md file so that it shows the changes in descending order. I believe this is a more common practice.

This version should close #10, #18, and replaces pull requests #7, #11 and #15.

Others

One missing feature in the updated package is the ability to auto reconnect when fail, because I didn't need it. I am not sure if I can take the time to do it, but I will wait for your response first.

Update: Auto reconnection can be achieved using the new AutoReconnectSseClient class.

Besides, I'd also like to suggest changing the package name to something like "dart_sse_client" as the package doesn't depend on Flutter. Also, using a branch name like dev instead of uat should be less confusing.

Thanks for you time!

PlatformException(no_carrier_name, No carrier name, java.lang.SecurityException: Not allowed to access cell location, null)

android 14
flutter 3.23.0
I inquired about this related issue and added the ACCESS_FINE_LOCATION permission, but it still doesn't work. Has anyone encountered the same problem? If you know, please tell me. Thank you very much.

http post not working

why http post cant work? i cant receive request even i do print log on the server

Errors using ChatGPT

Trying to listen to a stream from GPT, I can see the data being printed out, but my listeners never fire.

flutter: FormatException: Unexpected character (at character 1)
flutter: data: {"id":"chatcmpl-72CZsAqYS71OoS3bA3uGmCjkkScEO","object":"chat.complet...
flutter: ^
flutter: 
flutter: data: {"id":"chatcmpl-72CZsAqYS71OoS3bA3uGmCjkkScEO","object":"chat.completion.chunk","created":1680759360,"model":"gpt-3.5-turbo-0301","choices":[{"delta":{"role":"assistant"},"index":0,"finish_reason":null}]}
flutter: 
flutter: data: {"id":"chatcmpl-72CZsAqYS71OoS3bA3uGmCjkkScEO","object":"chat.completion.

Usage example:

var url = Uri.https('api.openai.com', '/v1/chat/completions');
    final headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer $k'};
    final response = await http.post(
      url,
      headers: headers,
      body: json.encode({
        'stream': true,
        'model': 'gpt-3.5-turbo',
        'messages': messages.map((e) {
          var role = e.chatMessageType == ChatMessageType.user ? 'user' : 'assistant';
          if (e.chatMessageType == ChatMessageType.system) role = 'system';
          return {'role': role, 'content': e.text};
        }).toList(),
        'temperature': temperature ?? .7,
        'max_tokens': 2000,
        'top_p': 1,
        'frequency_penalty': 1.0,
        'presence_penalty': 1.0,
      }),
    );
    final _listener = SSEClient.subscribeToSSE(
      url: 'https://api.openai.com/v1/chat/completions',
      header: headers,
    ).listen((event) {
      print(event);
    });

Bug: Extraneous whitespace in event data and extra newline at the end

Bug description

There is extraneous whitespace at the beginning of each line of the event data retrieved from event.data property. I believe this is because the library treats the space between "data:" and the message body in SSE format as part of the message body.

In addition, there is an extra newline character at the end of the event data, which causes some parsing issues.

Steps to reproduce

Create an SSE server with Express.js that returns a text chunk as shown below:

  res.write(`data: part-1\n`);
  res.write(`data: part-2\n`);
  res.write(`data: part-3\n\n`);

Create an SSE client with flutter_client_sse :

SSEClient.subscribeToSSE(
    url: 'http://<My Server Url>/events', 
    header: {
        "Accept": "text/event-stream",
    }).listen((event) {
        event.data?.split("").forEach((element) {
            print("$element ${element.codeUnitAt(0)}");
    });
});

Then print each char and its ascii code of the event.data string after receiving data from the server:

I/flutter ( 1067): --SUBSCRIBING TO SSE---
I/flutter ( 1067):   32
I/flutter ( 1067): p 112
I/flutter ( 1067): a 97
I/flutter ( 1067): r 114
I/flutter ( 1067): t 116
I/flutter ( 1067): - 45
I/flutter ( 1067): 1 49
I/flutter ( 1067):  10
I/flutter ( 1067):   32
I/flutter ( 1067): p 112
I/flutter ( 1067): a 97
I/flutter ( 1067): r 114
I/flutter ( 1067): t 116
I/flutter ( 1067): - 45
I/flutter ( 1067): 2 50
I/flutter ( 1067):  10
I/flutter ( 1067):   32
I/flutter ( 1067): p 112
I/flutter ( 1067): a 97
I/flutter ( 1067): r 114
I/flutter ( 1067): t 116
I/flutter ( 1067): - 45
I/flutter ( 1067): 3 51
I/flutter ( 1067):  10

It can be observed from the output that there is an extraneous whitespace (ASCII code 32) at the beginning of each line, and an extra newline character (ASCII code 10) at the end of the entire message block.

onDone callback not working when connection closes

Hello there,

I have been using this library and it worked fine but I have one problem with it: when the server closes the connection (Which I have checked is done right and works on postman for example), I cannot figure out how to detect it.

So I have made a EventStreamManager class that handles the stream because I needed a broadcast stream (then I figured out I could maybe do it simpler but that's not the point).

I can get events and all but nothing ever passes in the onDone callback function. I have tried everything but I cannot figure out how to make it work.

Has someone ever made it work ?

EventStreamManager.dart

import 'dart:async';
import 'package:echo_app/api/api_client.dart';
import 'package:flutter/foundation.dart';
import 'package:flutter_client_sse/flutter_client_sse.dart';

class EventStreamManager {
  final ApiClient _apiClient;
  StreamController<SSEModel>? _streamController;

  EventStreamManager(this._apiClient);

  Stream<SSEModel> get stream => _streamController!.stream;

  Future<void> startListeningForEvents(String sessionId) async {
    _streamController = StreamController<SSEModel>.broadcast();

    Stream<SSEModel> eventStream =
        await _apiClient.startListeningForEvents(sessionId);
    eventStream.listen((event) {
      if (!_streamController!.isClosed) {
        _streamController!.sink.add(event);
      }
    }, onDone: () {
      if (kDebugMode) {
        print('SSE closed');
      }
      _streamController!.close();
    }, onError: (error) {
      _streamController!.addError(error);
    });
  }

  void stopListeningForEvents() {
    _apiClient.stopListeningForEvents();
    _streamController?.close();
    _streamController = null;
  }
}

ApiClient.startListeningForEvents function:

  Future<Stream<SSEModel>> startListeningForEvents(String sessionId) async {
    var token = await getAccessToken();
    return SSEClient.subscribeToSSE(
        method: SSERequestType.GET,
        url: '$baseUrl/sessions/$sessionId',
        header: {
          'Content-Type': 'application/json',
          'Authorization': 'Bearer $token',
        });
  }

Update http version

The package is running an earlier version of the HTTP package.
Because flutter_client_sse >=2.0.0-beta.1 depends on http ^0.13.5 and google_fonts 5.1.0 depends on http ^1.0.0, flutter_client_sse >=2.0.0-beta.1 is incompatible with google_fonts 5.1.0.
And because no versions of google_fonts match >5.1.0 <6.0.0, flutter_client_sse >=2.0.0-beta.1 is incompatible with google_fonts ^5.1.0.
So, because aku_staff_agent depends on both google_fonts ^5.1.0 and flutter_client_sse ^2.0.0, version solving failed.

Configuration for retry

Thank you for addressing #21

I have a use-case, where infinite retry might not be an ideal option.
On server 401 response I want to refresh JWT token, so it would be great if there is a chance to configure retry and maybe for my use-case an ability to put 0 retries and complete the stream with some error

Reconnect after server disconnect

Currently when server closes connection stream is ended, can we make it reconnect?

Subscription hangs when server responds with 4xx status code

Hello

Please take a look at the example project here: https://github.com/eye-dee/flutter_client_sse_validation
The client hangs on --SUBSCRIBING TO SSE---, when similar test with OK status code completes successfully