proto-socket/dart/lib/src/inbound_gateway_io.dart

import 'dart:async';
import 'dart:isolate';
import 'dart:typed_data';

import 'inbound_gateway.dart';
import 'packets/message_common.pb.dart';

/// Long-running isolate worker gateway for Dart IO targets.
///
/// A single isolate is spawned once via [start] and kept alive to avoid
/// repeating spawn cost per frame. Raw frames are sent over a [SendPort] with
/// their [InboundFrame.seq]; the isolate decodes the [PacketBase] envelope
/// (pure work only) and sends the fields back. Results are reordered by [seq]
/// and emitted on [results] for the main-isolate coordinator to dispatch.
///
/// The gateway deliberately owns no stateful dispatch: listeners, request
/// handlers, the pending-request map, and the outbound write queue all remain
/// on the main isolate's receive coordinator.
class IsolateInboundGateway implements InboundGateway {
  /// Max frames handed to the worker without a result back before [submit]
  /// applies backpressure. Bounds the otherwise unbounded [SendPort] mailbox.
  static const int defaultCapacity = 1024;

  /// Threshold in bytes to use [TransferableTypedData] instead of copying.
  static const int transferableThreshold = 32768;

  /// Frames smaller than this decode in-process on the main isolate instead of
  /// being handed to the worker. A per-frame isolate round-trip costs far more
  /// than decoding a small envelope inline, so routing a high-rate burst of
  /// small frames through the worker collapses receive throughput (the WS burst
  /// regression). Large frames still offload to keep big-payload decode off the
  /// main isolate. Aligned with [transferableThreshold] so a single size
  /// boundary governs both isolate offload and zero-copy transfer. In-process
  /// results still flow through the shared seq-keyed [FrameReorderBuffer], so
  /// global dispatch order is preserved across both paths.
  static const int inProcessDecodeThreshold = transferableThreshold;

  final int capacity;

  final StreamController<DecodedFrame> _controller =
      StreamController<DecodedFrame>.broadcast();
  final FrameReorderBuffer _reorder = FrameReorderBuffer();

  Isolate? _isolate;
  SendPort? _toWorker;
  ReceivePort? _fromWorker;
  Completer<void>? _ready;
  bool _closed = false;

  int _inFlight = 0;
  Completer<void>? _capacityWaiter;

  IsolateInboundGateway({this.capacity = defaultCapacity})
      : assert(capacity > 0, 'capacity must be positive');

  @override
  Stream<DecodedFrame> get results => _controller.stream;

  @override
  int get inFlightCount => _inFlight;

  @override
  int get queuedCount => _reorder.pendingCount;

  @override
  Future<void> start() async {
    if (_isolate != null || _closed) return;
    final ready = Completer<void>();
    _ready = ready;
    final fromWorker = ReceivePort();
    _fromWorker = fromWorker;
    fromWorker.listen(_onWorkerMessage);
    _isolate = await Isolate.spawn(_gatewayEntry, fromWorker.sendPort);
    await ready.future;
  }

  void _onWorkerMessage(dynamic message) {
    if (message is SendPort) {
      _toWorker = message;
      if (_ready != null && !_ready!.isCompleted) _ready!.complete();
      _ready = null;
      return;
    }
    if (message is List) {
      _inFlight--;
      _releaseCapacity();
      final error = message[5];
      final dynamic rawData = message[2];
      final List<int> data;
      if (rawData is TransferableTypedData) {
        data = rawData.materialize().asUint8List();
      } else {
        data = (rawData as List).cast<int>();
      }
      final decoded = error != null
          ? DecodedFrame.error(seq: message[0] as int, error: error)
          : DecodedFrame(
              seq: message[0] as int,
              typeName: message[1] as String,
              data: data,
              incomingNonce: message[3] as int,
              responseNonce: message[4] as int,
            );
      _emitReleased(decoded);
    }
  }

  /// Pushes a decoded frame through the shared reorder buffer and emits whatever
  /// contiguous run is now releasable, in seq order. Used by both the worker
  /// reply path and the in-process small-frame path so ordering is identical.
  void _emitReleased(DecodedFrame decoded) {
    for (final ready in _reorder.release(decoded)) {
      if (!_controller.isClosed) _controller.add(ready);
    }
  }

  @override
  Future<void> submit(InboundFrame frame) async {
    if (_closed) return;
    // Small frames decode in-process: the isolate round-trip costs more than
    // decoding a small envelope inline, and a high-rate small-frame burst routed
    // through the worker collapses throughput. The result still flows through the
    // shared seq-keyed reorder buffer, so it stays ordered with isolate-decoded
    // large frames (a small seq buffers until any earlier large seq releases).
    if (frame.bytes.length < inProcessDecodeThreshold) {
      _emitReleased(_decodeInProcess(frame));
      return;
    }
    // Wait for the worker's SendPort so an early frame is never dropped.
    final ready = _ready;
    if (ready != null) await ready.future;
    // Backpressure: hold until an in-flight result frees a slot.
    while (!_closed && _inFlight >= capacity) {
      final waiter = _capacityWaiter ??= Completer<void>();
      await waiter.future;
    }
    if (_closed) return;
    final port = _toWorker;
    if (port == null) return;
    _inFlight++;
    final bytes = frame.bytes;
    final Object payload;
    if (bytes is Uint8List) {
      payload = bytes.length >= transferableThreshold
          ? TransferableTypedData.fromList([bytes])
          : bytes;
    } else {
      payload = bytes.length >= transferableThreshold
          ? TransferableTypedData.fromList([Uint8List.fromList(bytes)])
          : bytes;
    }
    port.send([frame.seq, payload]);
  }

  /// Decodes a small frame's [PacketBase] envelope on the main isolate. A
  /// malformed envelope becomes an ordered [DecodedFrame.error] occupying its
  /// seq, exactly as the worker path does, so the reorder buffer never stalls.
  DecodedFrame _decodeInProcess(InboundFrame frame) {
    try {
      final common = PacketBase.fromBuffer(frame.bytes);
      return DecodedFrame(
        seq: frame.seq,
        typeName: common.typeName,
        data: common.data,
        incomingNonce: common.nonce,
        responseNonce: common.responseNonce,
      );
    } catch (error) {
      return DecodedFrame.error(seq: frame.seq, error: error);
    }
  }

  void _releaseCapacity() {
    final waiter = _capacityWaiter;
    if (waiter != null && _inFlight < capacity) {
      _capacityWaiter = null;
      waiter.complete();
    }
  }

  @override
  Future<void> close() async {
    if (_closed) return;
    _closed = true;
    _toWorker = null;
    _isolate?.kill(priority: Isolate.immediate);
    _isolate = null;
    _fromWorker?.close();
    _fromWorker = null;
    _inFlight = 0;
    _reorder.clear();
    // Unblock any submit awaiting readiness or capacity so it observes _closed.
    if (_ready != null && !_ready!.isCompleted) _ready!.complete();
    _ready = null;
    final waiter = _capacityWaiter;
    _capacityWaiter = null;
    waiter?.complete();
    await _controller.close();
  }
}

/// Isolate entry point: decodes [PacketBase] envelopes and returns fields.
///
/// Runs only pure decode work. The first message back is the worker's
/// [SendPort]; subsequent messages are `[seq, bytes]` decode requests.
///
/// Each reply is a 6-element list `[seq, typeName, data, nonce, responseNonce,
/// error]`. A malformed frame is caught and reported as an ordered error result
/// (index 5 non-null) instead of crashing the isolate, so the reorder buffer
/// never stalls on the dropped seq.
void _gatewayEntry(SendPort toMain) {
  final fromMain = ReceivePort();
  toMain.send(fromMain.sendPort);
  fromMain.listen((dynamic message) {
    if (message == null) {
      fromMain.close();
      return;
    }
    final request = message as List;
    final seq = request[0] as int;
    final payload = request[1];
    final List<int> bytes;
    if (payload is TransferableTypedData) {
      bytes = payload.materialize().asUint8List();
    } else {
      bytes = (payload as List).cast<int>();
    }
    try {
      final common = PacketBase.fromBuffer(bytes);
      final List<int> commonData = common.data;
      final Object dataPayload;
      if (commonData is Uint8List) {
        dataPayload = commonData.length >= IsolateInboundGateway.transferableThreshold
            ? TransferableTypedData.fromList([commonData])
            : commonData;
      } else {
        dataPayload = commonData.length >= IsolateInboundGateway.transferableThreshold
            ? TransferableTypedData.fromList([Uint8List.fromList(commonData)])
            : commonData;
      }
      toMain.send([
        seq,
        common.typeName,
        dataPayload,
        common.nonce,
        common.responseNonce,
        null,
      ]);
    } catch (error) {
      toMain.send([seq, '', const <int>[], 0, 0, 'decode failed: $error']);
    }
  });
}