SSE Processor

A pure Dart package for managing Server-Sent Events (SSE) delivery.

Project Introduction

SSE Processor is an SSE handling library designed specifically for Flutter application. It addresses a series of challenges encountered when processing server-sent event streams, including data parsing, rete control, dispatch mechanisms, priority management, caching strategies, and more.

Why This Library is Needed

When handling streaming data generated by AI, clients often face the following challenges:

  • Parsing Issues: How to parse streaming data and convert it into a data structure usable by the client.
  • Rate Control: How to control the display rate of stream data sent irregularly by the server.
  • Dispatch Mechanism: How to distribute the converted stream data to multiple modules for logical processing.
  • Priority Management: How to consider priority during the dispatch process.
  • Caching Strategy: How to design a circular buffer pool for timed dispatch.
  • Automatic Cleanup: How to simplify the registration and unregistration of interceptors.
  • Dispatch State Control: How to pause/resume dispatch and monitor dispatch status.
  • Lifecycle Management: How to handle lifecycle callbacks related to automatic cleanup features.
  • Native Stream Bridging: How to integrate with native layer network components. (Maybe your streaming data is provided from the native layer; if not, you can ignore it.)
  • Connection State Management: How to monitor and manage SSE connection status.

What features does this library have?

  • Complete SSE Processing Flow: Full lifecycle management from stream data parsing to caching and dispatch.
  • Flexible Interceptor Mechanism: Supports adding multiple interceptors to receive different types of SSE events.
  • Intelligent Cache Management: Provides a circular buffer pool and timed dispatch functionality.
  • Connection State Monitoring: Real-time monitoring of connection status and notification to observers.
  • Interceptor Auto-Cleanup: Avoids receiving unnecessary SSE events.
  • Custom Protocol Conversion: Supports conversion of stream data formats for different business scenarios.
  • Stream Filter: Supports converting a single SSE object into multiple SSE objects.

Installation

Add the following dependency to your pubspec.yaml file:

dependencies:
  sse_processor: ^1.0.0

Then, execute the following command to get the dependencies:

flutter pub get

Usage Guide

Initialization

import 'package:dio/dio.dart';
import 'package:sse_processor/sse.dart';

// Create a Dio instance
final dio = Dio();

// initialize SSEProcessor
SSEProcessor.init(
  SSEProcessorConfig(
    version: '1.0.0',
    debug: true,
    idleTimeout: 30.0, // seconds
    logFileName: 'sse_processor_log',
    exceptionTimeout: 60, // seconds
    sseBufferExtractInterval: 100, // millisecond
    // Optional configurations
    debugTag: 'MY_APP_SSE',
    eleTypesInInterval: {'text', 'message'},
    unCheckConnectStatePaths: {'/api/stream/health'},
    // Custom Filter (Optional)
    sseFilter: MyCustomSSEFilter(),
    // Custom Stream Adapter (Optional)
    streamAdapter: MyCustomStreamAdapter(),
  ),
  dio,
);

Add connection status observer

// Add connection status observer
sseProcessor.addConnectStateObserver(
  SSEConnectObserver(
    'main_observer',
    NotifyPriority.high, // High priority
    (ConnectState state) {
      switch (state) {
        case ConnectState.connectActive:
          print('The connection is active and data is being transmitted.');
          break;
        case ConnectState.connectIdle:
          print('The connection is idle with no data transmission.');
          break;
        case ConnectState.connectException:
          print('Connection exception.');
          break;
        case ConnectState.disconnectError:
          print('Connection error.');
          break;
        // Add more cases as needed
        default:
          break;
      }
      return false; // Returning true means not continuing to pass to subsequent observers
    },
  ),
);

// Add SSE interceptor
sseProcessor.addSSEInterceptor(
  SSEInterceptor(
    // Type of event to listen to
    watchEvents: {'message', 'text', 'status'},
    // Whether to automatically clean up when appropriate time
    autoClean: true,
    // Interceptor callback
    intercept: (ServerSentEvent sse, SSEChain chain) {
      // Process the SSE event here
      print('Received SSE event: ${sse.elementType}');
      print('Event content: ${sse.result}');
      
      // Continue to pass it to the next interceptor in the chain
      return chain.proceed(sse);
      
      // Or interrupt the transmission and return a response
      // return SSEResponse(removeCache: true, autoRemove: false);
    },
  ),
);

Control the distribution status

// pause distribution
sseProcessor.setDeliverState(DelivererState.pause);

// resume distribution
sseProcessor.setDeliverState(DelivererState.active);

Resource release

Remember to release resource when SSEProcessor is no longer needed:

// Release resources
sseProcessor.destroy();

Core concept

ServerSentEvent

The basic data structure of an SSE event includes the following fields:

  • sessionLogId: Session Log ID (Usually represents a unique identifier)
  • elementType: Element type
  • result: Event result content
  • extra: Extra information
  • isHistory: Is it a historical event?

SSEInterceptor

An interceptor for SSE events, used to handle specific types of SSE events:

  • watchEvents: Collection of monitored event types
  • autoClean: Whether to automatically clean up
  • intercept: Interception handling callback

ConnectState Connection status enumeration, including the following states:

  • connectActive: The connection is normal, and data is being transmitted.
  • connectIdle: The connection is normal, with no data transmission.
  • connectException: The connection is exception
  • connectSuspend: The connection was suspended by the client
  • disconnectRepairing: The connection is disconnected, but is being repaired
  • disconnectError: Connection disconnected, exceeding the number of retries
  • disconnectNormal: The connection is disconnected normally.

Advanced usage

Custom SSE Filter

class MyCustomSSEFilter implements SSEFilter {
  @override
  Future<List<ServerSentEvent>> resolve(ServerSentEvent sse) async {
    // For example, split a complete message into multiple character events to simulate a typewriter effect
    final events = <ServerSentEvent>[];
    if (sse.elementType == 'message' && sse.result != null) {
      for (int i = 0; i < sse.result!.length; i++) {
        final charEvent = ServerSentEvent()
          ..sessionLogId = sse.sessionLogId
          ..elementType = 'message_char'
          ..result = sse.result![i]
          ..extra = sse.extra;
        events.add(charEvent);
      }
    } else {
      events.add(sse);
    }
    return events;
  }
}

Custom stream adapter

class MyCustomStreamAdapter implements StreamAdapter {
  @override
  Future<ServerSentEvent?> adapt(dynamic data) async {
    // Process stream data in a custom format and convert it to ServerSentEvent
    if (data is Map<String, dynamic>) {
      return ServerSentEvent()
        ..sessionLogId = data['session_id']?.toString() ?? ''
        ..elementType = data['type']?.toString() ?? ''
        ..result = data['content']?.toString() ?? ''
        ..extra = jsonEncode(data['metadata'] ?? {});
    }
    return null;
  }
}

Notes

  • Ensure the use of the correct Dio instance: when initiating a streaming request, you must use the Dis instance passed in during initialization.
  • Release resource when SSEProcessor is no longer needed: call the destroy() method to release resources when the SSEProcessor is no longer needed.
  • Handling connection state changes: Monitor connection state changes by adding a connection state observer.
  • Use automatic cleaning properly: For temporarily used interceptors, it is recommended to enable autoCLean to void memory leaks.
  • Log debugging: In the development environment, you can set debug to true to enable detailed logs.

Libraries

sse