openAgentStream method

void openAgentStream(
  1. String key, {
  2. ISSEStream? agentStream,
  3. bool goCachePool = false,
  4. bool autoRemove = false,
  5. bool supportPeekDispatch = true,
  6. ValueChanged<List<ServerSentEvent>>? onDone,
})

Open agent stream, support set outside stream for dispatch sse. (for example sse from db) key is an identity marker that is associated with the agentStream. agentStream takes on the proxy responsibility and is responsible for the logic of external conversion of SSE streams goCachePool is put in cache pool then deliver sse autoRemove remove agentStream in _agentStreams when stream end supportPeekDispatch Directly distribute to the interceptors that support peek mode, it only take effect when goCachePool is true.

Implementation

void openAgentStream(String key,
    {ISSEStream? agentStream,
    bool goCachePool = false,
    bool autoRemove = false,
    bool supportPeekDispatch = true,
    ValueChanged<List<ServerSentEvent>>? onDone}) {
  slog.df("open agent stream key $key", tag: tag);
  if (agentStream != null) {
    _agentStreams.putIfAbsent(key, () => agentStream);
  }
  _sseCacheDeliverer.reset();
  ISSEStream? stream = _agentStreams[key];
  stream?.open(onReceive: (event) async {
    if (goCachePool) {
      _resolveSseEvent([event], "", toPeek: supportPeekDispatch);
    } else {
      _deliverSSEC(ServerSentEventCache(event, ""));
    }
  }, onDone: (list) async {
    _resolveSseEvent([SSEAutoRemoveInterceptor.genAutoRemoveSSE()], "");
    slog.df("stream transform done", tag: tag);
    onDone?.call(list);
    if (autoRemove) {
      stream.destroy();
      _agentStreams.remove(key);
    }
  }, onError: (e) {
    _resolveSseEvent([SSEAutoRemoveInterceptor.genAutoRemoveSSE()], "");
    slog.df("stream transform error $e", tag: tag);
    _handleConnectError();
  });
}