enqueue method

void enqueue(String type, Uint8List bytes, { Map<String, Object?>? payload, })

Enqueue by wire type. payload is the already-decoded frame when the producer has it — passed to createDecodedSendGuard so no frame is decoded more than once on its way out.

Implementation

void enqueue(String type, Uint8List bytes, {Map<String, Object?>? payload}) {
  if (_disposed) return;
  final lossy = _lossyTypes.contains(type);
  final substitutable = !lossy || _substitutableLossyTypes.contains(type);
  final byteCap = maxPayloadBytes;
  if (byteCap != null && bytes.length > byteCap) {
    final substitute = substitutable
        ? _substitute(type, bytes, 'payload_too_large')
        : null;
    if (substitute != null && substitute.length <= byteCap) {
      _logDrop(type, bytes.length, 'over_cap', substituted: true);
      _queue.add(_Queued(type: type, bytes: substitute, lossy: lossy));
      unawaited(_drain());
      return;
    }
    if (lossy) {
      droppedLossy += 1;
    } else {
      droppedGuaranteed += 1;
    }
    _logDrop(type, bytes.length, 'over_cap');
    _notifyDropped(type);
    return;
  }
  bool Function()? sendGuard;
  try {
    final decodedGuard = createDecodedSendGuard;
    sendGuard = payload != null && decodedGuard != null
        ? decodedGuard(type, payload)
        : createSendGuard?.call(bytes);
  } catch (_) {
    // A failed origin lookup must never silently remove the policy gate.
    sendGuard = () => false;
  }
  _queue.add(_Queued(
    type: type,
    bytes: bytes,
    lossy: lossy,
    substitutable: substitutable,
    terminal: _isTerminal(type, bytes, payload),
    sendGuard: sendGuard,
  ));

  if (lossy) {
    var lossyCount = _queue.where((q) => q.lossy).length;
    while (lossyCount > lossyCap) {
      final idx = _queue.indexWhere((q) => q.lossy);
      if (idx < 0) break;
      final removed = _queue.removeAt(idx);
      droppedLossy += 1;
      _logDrop(removed.type, removed.bytes.length, 'lossy_overflow');
      _notifyDropped(removed.type);
      lossyCount = _queue.where((q) => q.lossy).length;
    }
  } else {
    var gCount = _queue.where((q) => !q.lossy).length;
    while (gCount > guaranteedCap) {
      // Evict the oldest NON-terminal guaranteed frame. A terminal
      // mission_status is the one frame the backend is blocked on; losing
      // it strands the agent on a mission that already ended. A lane full
      // of terminals only (one per mission — tiny and bounded) is allowed
      // to exceed the cap rather than drop one.
      final idx = _queue.indexWhere((q) => !q.lossy && !q.terminal);
      if (idx < 0) break;
      final removed = _queue.removeAt(idx);
      droppedGuaranteed += 1;
      _logDrop(removed.type, removed.bytes.length, 'guaranteed_overflow');
      _notifyDropped(removed.type);
      gCount -= 1;
    }
  }
  unawaited(_drain());
}