callTidbDataEndpoint function

Future<Map<String, dynamic>> callTidbDataEndpoint({ required String region, required String appId, required String path, required String publicKey, required String privateKey, String method = "GET", Map<String, String>? query, Object? body, })

Calls a deployed Data Service endpoint with Digest authentication.

Implementation

Future<Map<String, dynamic>> callTidbDataEndpoint({
  required String region,
  required String appId,
  required String path,
  required String publicKey,
  required String privateKey,
  String method = "GET",
  Map<String, String>? query,
  Object? body,
}) async {
  final client = HttpClient();
  client.authenticate = (uri, scheme, realm) {
    client.addCredentials(
      uri,
      realm ?? "",
      HttpClientDigestCredentials(publicKey, privateKey),
    );
    return Future.value(true);
  };
  final normalizedPath = path
      .split("/")
      .where((segment) => segment.isNotEmpty)
      .map(Uri.encodeComponent)
      .join("/");
  final uri = Uri.https(
    "$region.data.tidbcloud.com",
    "/api/v1beta/app/${Uri.encodeComponent(appId)}/endpoint/$normalizedPath",
    query,
  );
  late HttpClientResponse response;
  var text = "";
  for (var attempt = 0; attempt < 2; attempt++) {
    try {
      final request = await client.openUrl(method, uri);
      request.headers
        ..set(HttpHeaders.acceptHeader, "application/json")
        ..set(HttpHeaders.contentTypeHeader, "application/json");
      if (body != null) {
        final encodedBody = utf8.encode(jsonEncode(body));
        request.contentLength = encodedBody.length;
        request.add(encodedBody);
      }
      response = await request.close();
      text = await utf8.decoder.bind(response).join();
    } on HttpException catch (exception) {
      if (attempt == 0 &&
          exception.message.contains(
            "Connection closed before full header was received",
          )) {
        continue;
      }
      if (exception.message.contains(
        "Connection closed before full header was received",
      )) {
        client.close();
        for (var curlAttempt = 0; curlAttempt < 10; curlAttempt++) {
          try {
            return await _callTidbDataEndpointWithCurl(
              uri: uri,
              publicKey: publicKey,
              privateKey: privateKey,
              method: method,
              body: body,
            );
          } on HttpException catch (curlException) {
            if (curlAttempt == 9 ||
                !curlException.message.contains(
                  "deployed endpoint not found",
                )) {
              rethrow;
            }
            await Future<void>.delayed(const Duration(seconds: 2));
          }
        }
      }
      rethrow;
    }
    // HttpClient's first Digest challenge can consume a POST body without
    // replaying it. The authenticated connection is cached after that
    // challenge, so retry the same body once when the endpoint reports a
    // missing parameter.
    if (response.statusCode != HttpStatus.badRequest ||
        body == null ||
        attempt == 1 ||
        !text.contains("parameter is required")) {
      break;
    }
  }
  client.close();
  final decoded = text.isEmpty ? <String, dynamic>{} : jsonDecode(text);
  if (response.statusCode < 200 || response.statusCode >= 300) {
    throw HttpException(
      "TiDB Data Service smoke test failed "
      "(${response.statusCode}): $text",
      uri: uri,
    );
  }
  final normalized = decoded is Map
      ? decoded.map((key, value) => MapEntry(key.toString(), value))
      : <String, dynamic>{"data": decoded};
  _validateTidbDataResult(normalized, uri);
  return normalized;
}