sseProcessor top-level property

SSEProcessor get sseProcessor

Why:

In the theater module, AI-generated data is usually distributed in the form of a stream, and the client also needs to receive it in the form of a stream. Interacting in this mode will face several problems:

  1. Parsing: How to parse the streamed data after it is distributed? It is necessary to negotiate a protocol with the server and then convert it into the data structure object of the client.
  2. Rate control: Usually, the rate of the streamed data distributed by the server is irregular. After the client receives it, rate control is required for display.
  3. Distribution: How to display (logic processing) after obtaining the streamed data? Because the project is developed in modules, a data object transformed from a stream may be used for logic processing by multiple modules. At this time, a distribution mechanism is required.
  4. Priority level: When it comes to distribution, should priority be considered?
  5. Caching: For rate control, a loop cache pool can be looped should be designed for timed distribution.
  6. Automatic cleaning: In order for the user to not consider the cumbersome operations such as interceptor registration and deregistration, a set of automatic cleaning processes needs to be completed.
  7. Distribution status control: Including pausing/resuming distribution and monitoring the distribution status.
  8. Lifecycle: Since there is an automatic cleaning function, should lifecycle callbacks also be arranged?
  9. Native stream bridging: Because the network-related logic is placed in the native layer, the network stream data request should also be initiated by the native layer, so that a set of network components can be uniformly used.
  10. Connection status management: Although SSE is not a long connection, it is a little longer than the usual short connection, and then the connection status will be involved.

In order to solve the above problems, a set of SSE distribution components needs to be designed.

How:

A net request flow（Contain SSE deliver)）

                          [Retrofit]
                              |
                           Options
                              |
                            [Dio]
                              |
                           Request
                              |
                      Request Interceptor
                              |
         transformer.transformRequest (NativeTransformer)
                              |
            httpClientAdapter.fetch (NativeClientAdapter)
                              |
     ----------------------------------------------------

  (Flutter net component)		    （Native net component in Android for example）

        [HttpClient]			           [OkHttp]

         [Socket]				             [Socket]

     ----------------------------------------------------
                              |
            transformer.transformResponse (NativeTransformer)
                              |
                   If is return sse stream
                              ｜

------------(No)------------- ------(Yes)------ | | | -----------------------Dio(Interceptor SSEProcessor)Receive native stream data--------------------- Response Interceptor | | | SSECore(Parse stream data transform to sse) ----------------------- SSEConnectManager Error Interceptor ｜ | | SSEFilterService (The client processes,cutting and merging) -------------- Response ｜ | SSECacheDeliverer（Cache SSE and deliver it into interceptor）-------- ｜ SSEInterceptorManager(Deliver SSE to receiver) ｜ SSEInterceptor(SSE)

SSEProcessor is responsible parse sse stream data, cache sse and deliver sse to corresponding receiver. It also provide a connect manager by SSEConnectManager

SSEProcessor it's actually a dio interceptor, In order to associate the stream received after the dio component initiates a network request.

There is a SSEProcessor._agentStream parameter, it give a change to load data from outside data (Maybe is real stream or mock, Follow ISSEStream interface)

This class depend on 5 core class to operation. (Stream manager SSECore, Cache manager SSECacheDeliverer, Interceptor manager SSEInterceptorManager, Connection state manager SSEConnectManager, A bridge class transform from native net request stream SSEBridge)

Below will introduce them detail.

Stream management SSECore : Responsible open a stream, transform stream to ServerSentEvent, when completing a complete parsing ServerSentEvent, package as ServerSentEventCache put to SSECacheDeliverer.

Stream adapter [StreamAdapter[ : At first, the format of the streaming data issued by the server side was fixed. However, later, different business scenarios would have different formats of streaming data. Externally, it is desired to use the same ServerSentEvent format. Therefore, a custom protocol conversion mechanism is provided here for the upper-layer business to use.

Cache management SSECacheDeliverer : This will loop deliver ServerSentEventCache to SSEInterceptorManager by timer. It provide some feature. Such as set interval, set active and pause deliver state.

Now have 2 chance to flush cache data, send to the interceptors that meets the requirements. Chance 1, When after add sse interceptor, this hope to guarantee added interceptor will receive cache sse data instantly. Change 2, Switch deliver state DelivererState.active by SSECacheDeliverer.setState

About cache auto remove : Now each ServerSentEventCache will generate a time stamp, when remove a ServerSentEventCache from cache by SSEInterceptor return SSEResponse, Before this time stamp cache sse will auto remove.

This action is for guarantee cache pool doesn't exist redundancy cache event, but there are some special situation. You can use SSEResponse to prevent auto remove. More detail please link to SSEResponse

About cache deliver state switch : cache deliver provide pause and resume ability. Switch by SSEProcessor.setDeliverState DelivererState.active Active cache event deliver to sse interceptor. DelivererState.pause Pause cache event deliver to sse interceptor.

SSE Interceptor management SSEInterceptorManager : SSECacheDeliverer delivered sse will to match interceptor, first according to SSEInterceptor.watchEvents match sse. Then create a responsible chain SSEChain, delivered sse according order in SSEChain, SSEInterceptor that in SSEChain will invoke SSEInterceptor.intercept process it. SSEChain.proceed invoke in SSEInterceptor will continue deliver to next interceptor. If return SSEResponse straightway will break off deliver in SSEChain.

SSEResponse.removeCache As whether remove from cache pool. (Generally need to remove) SSEResponse.autoRemove Whether auto remove from cache pool. This flag only available when removeCache is false, set ture is auto remove event according to last time stamp matched. false will wait interceptor to remove it.

Connection state management SSEConnectManager : Responsible switch connection state by sse stream state, add and remove connection state observer. Add connection observer SSEConnectManager.addConnectStateObserver call back as ConnectState to receive new connection state. SSEConnectManager.removeConnectStateObserver remove when appropriate time.

  • About interceptor auto remove mechanism * In order to receive hopeless sse, For example, add an interceptor receive A event, then an interceptor is created to receive A event, Now deliver A event will notified previous two interceptors, maybe you expect not like as. For meet this situation, provide auto remove interceptors mechanism. SSEInterceptor.autoClean represent whether auto remove when appropriate time. (This appropriate time will change according requirement in the future)

Now auto remove mechanism refer to SSEChain.deliver

Native sse stream bridge SSEBridge : Cache byte stream from native, create new stream to SSECore, Thought dio to adapt it.

Sio It it mainly used to replace Dio to perform network requests that can receive streaming data, avoiding requests for streaming data in A module and receiving streaming data in B module (another module). It also supports receiving json format data return from server side, completely according wishes of server side, usually api request error will return json format response data.

Way of use:

Step 1：Init config

SSEProcessor.init(SSEProcessorConfig config, Dio dio);

Step 2：Add connection state observer

sseProcessor.addConnectStateObserver();

Step 3：Add sse interceptors

sseProcessor.addSSEInterceptor();

Step 4: Request stream, Now send request that support response stream by retrofit. Associating dio must use the dio object passed in at initialization.

@GET("/v2/interaction/keep-on/{sessionId}") @Headers(<String, dynamic>{ 'Accept': 'text/event-stream', 'Client-Api-Timeout': 1000 * 60 * 2, }) Future

Step 5：destroy resource

sseProcessor.destroy()

  • This mode compared with Stream *

In Flutter, Stream itself also provides transformation and distribution functions, but there are some limitations. There are two types of Streams here. One is the Single subscription stream, and the other is the Broadcast Stream. The Single subscription stream is abandoned first because it can only have one listener registered. The Broadcast Stream also cannot meet the requirements. Although it can register multiple Listeners, it cannot guarantee to receive all events in the stream and has no concept of priority. So simply using Stream for distribution does not meet the requirements. (In addition to the deficiencies mentioned above, there are many other places that cannot be satisfied)

Implementation

SSEProcessor get sseProcessor {
  assert(SSEProcessor.sseProcessorFactory != null,
      "SSEProcessor not init, please call SSEProcessor.init first");
  SSEProcessor? sseProcessor = SSEProcessor.sseProcessorFactory?.call();
  assert(sseProcessor != null,
      "delegate error, Please check whether the delegate parameter is operating correctly");
  return sseProcessor!;
}