fetchLive method

Implementation

Stream<SanityQueryResponse> fetchLive(
  String query, {
  Map<String, String>? params,
  includeDrafts = false,
}) {
  assert(
    includeDrafts
        ? config.perspective == Perspective.drafts && !config.useCdn
        : true,
    'When includeDrafts is true, the config must have perspective set to previewDrafts and useCdn set to false',
  );

  final liveConfig = includeDrafts ? LiveConfig.withDrafts() : LiveConfig();
  final sanityRequest = SanityRequest(
    urlBuilder: urlBuilder,
    query: '',
    live: liveConfig,
  );

  final uri = sanityRequest.getUri;

  final headers = Map<String, String>.from(_requestHeaders);
  headers['Accept'] = 'text/event-stream';

  final controller = StreamController<SanityQueryResponse>();

  final flux = SanityConfig.createEventFlux();

  flux.connect(
    EventFluxConnectionType.get,
    uri.toString(),
    autoReconnect: true,
    reconnectConfig: ReconnectConfig(
      mode: ReconnectMode.linear,
      maxAttempts: 5,
      onReconnect: () => debugPrint('Reconnected to Live API'),
    ),
    header: headers,
    httpClient: _EventFluxHttpClientAdapter(httpClient: httpClient),
    tag: query,
    onSuccessCallback: (response) {
      if (response == null || response.stream == null) {
        throw LiveConnectException('With query: $query, params: $params');
      }

      _onLiveConnectCallback(query, params, config: (
        liveConfig: liveConfig,
        controller: controller,
        stream: response.stream!,
        flux: flux,
      ));
    },
    onError: (error) {
      controller.close();
      throw LiveConnectException(
          'With query: $query, params: $params, error: $error');
    },
  );

  return controller.stream;
}