callStreamingServerEndpoint<T, G> method

@override dynamic callStreamingServerEndpoint<T, G>(String endpoint, String method, Map<String, dynamic> args, Map<String, Stream> streams,)

Calls a server endpoint method that supports streaming. The streams parameter is a map of stream names to stream objects. The method will listen to the streams and send the data to the server. Typically, this method is called by generated code.

T is the type of the return value of the endpoint stream. This is either a Stream or a Future.

G is the generic of T, such as T

If T is not a Stream or a Future, the method will throw an exception.

Implementation

@override
dynamic callStreamingServerEndpoint<T, G>(
  String endpoint,
  String method,
  Map<String, dynamic> args,
  Map<String, Stream> streams,
) {
  var connectionDetails = MethodStreamConnectionDetails(
    endpoint: endpoint,
    method: method,
    args: args,
    parameterStreams: streams,
    outputController: StreamController<G>(),
    authenticationProvider: () async =>
        authenticationKeyManager?.getHeaderValue(),
  );

  _methodStreamManager.openMethodStream(connectionDetails).catchError((e, _) {
    Object error;
    if (e is OpenMethodStreamException) {
      error = switch (e.responseType) {
        OpenMethodStreamResponseType.endpointNotFound =>
          ServerpodClientNotFound(),
        OpenMethodStreamResponseType.authenticationFailed =>
          ServerpodClientUnauthorized(),
        OpenMethodStreamResponseType.authorizationDeclined =>
          ServerpodClientForbidden(),
        OpenMethodStreamResponseType.invalidArguments =>
          ServerpodClientBadRequest(),
        OpenMethodStreamResponseType.success =>
          ServerpodClientException('Unknown error, data: $e', -1),
      };
    } else {
      error = e;
    }

    connectionDetails.outputController.addError(error);
    connectionDetails.outputController.close();
  });
  if (T == Stream<G>) {
    return connectionDetails.outputController.stream;
  } else if ((T == Future<G>) && G == getType<void>()) {
    var result = Completer<void>();
    // Listen to stream so that close can be called when method has returned.
    connectionDetails.outputController.stream.listen(
      (e) {},
      onError: ((e, _) => result.completeError(e)),
      onDone: () => result.complete(),
      cancelOnError: true,
    );
    return result.future;
  } else if (T == Future<G>) {
    var result = Completer<G>();
    connectionDetails.outputController.stream.first.then((e) {
      result.complete(e);
    }, onError: (e, _) {
      result.completeError(e);
    });
    return result.future;
  } else {
    throw UnsupportedError('Unsupported type $T');
  }
}