drainHttpQueue static method

void drainHttpQueue()

Drain commons' queued telemetry HTTP requests (signalled by the wake-up). Each request is an owned buffer framed as u32 LE endpointLenjson.

Implementation

static void drainHttpQueue() {
  final managerPtr = _managerPtr;
  if (managerPtr == null || !_acceptingHttpRequests) return;

  try {
    final lib = PlatformLoader.loadCommons();
    final pollFn = lib
        .lookupFunction<
          Int32 Function(Pointer<Void>, Pointer<RacProtoBuffer>),
          int Function(Pointer<Void>, Pointer<RacProtoBuffer>)
        >('rac_telemetry_manager_poll_http_request');
    final bindings = RacNative.bindings;

    // Bounded so a misbehaving producer can't spin forever; the loop also
    // exits as soon as poll reports an empty queue (rc != 0).
    for (var i = 0; i < 256; i++) {
      final out = calloc<RacProtoBuffer>();
      try {
        bindings.rac_proto_buffer_init(out);
        final code = pollFn(managerPtr, out);
        // RAC_SUCCESS == 0; anything else (e.g. RAC_ERROR_NOT_FOUND) = drained.
        if (code != 0 || out.ref.data == nullptr || out.ref.size < 5) {
          return;
        }
        final bytes = out.ref.data
            .asTypedList(out.ref.size)
            .toList(growable: false);
        final requiresAuth = bytes[0] != 0;
        final endpointLen =
            bytes[1] | (bytes[2] << 8) | (bytes[3] << 16) | (bytes[4] << 24);
        final endpoint = utf8.decode(bytes.sublist(5, 5 + endpointLen));
        final body = utf8.decode(bytes.sublist(5 + endpointLen));
        late final Future<void> request;
        request = _sendTelemetryHttp(
          endpoint,
          body,
          requiresAuth,
        ).whenComplete(() => _inFlightHttpRequests.remove(request));
        _inFlightHttpRequests.add(request);
      } finally {
        bindings.rac_proto_buffer_free(out);
        calloc.free(out);
      }
    }
  } catch (_) {
    _logger.debug('Telemetry HTTP queue drain failed');
  }
}