teeStreamCapture function

Stream<List<int>> teeStreamCapture(Stream<List<int>> source, { required int maxBytes, Duration chunkDelay(int chunkBytes)?, required void onDone(Uint8List? captured, int totalBytes), required void onError(Object error, StackTrace stackTrace, Uint8List? captured, int totalBytes,), required void onCancel(Uint8List? captured, int totalBytes), })

Tees source into a new stream that:

• Immediately forwards every chunk to listeners,
• Accumulates up to maxBytes into a CappedBytesBuilder for traffic inspection,
• Delays chunk delivery if chunkDelay produces a duration (for network simulation),
• Calls exactly one termination callback (onDone, onError, or onCancel) with captured bytes.

Implementation

Stream<List<int>> teeStreamCapture(
  Stream<List<int>> source, {
  required int maxBytes,
  Duration Function(int chunkBytes)? chunkDelay,
  required void Function(Uint8List? captured, int totalBytes) onDone,
  required void Function(Object error, StackTrace stackTrace,
          Uint8List? captured, int totalBytes)
      onError,
  required void Function(Uint8List? captured, int totalBytes) onCancel,
}) {
  final buffer = CappedBytesBuilder(maxBytes);
  bool settled = false;
  bool isCanceled = false;

  late StreamSubscription<List<int>> sub;
  late final StreamController<List<int>> controller;

  void finishDone() {
    if (settled) return;
    settled = true;
    final captured = buffer.toBytesOrNull();
    guardCapture(() => onDone(captured, buffer.totalBytes));
  }

  void finishError(Object error, StackTrace st) {
    if (settled) return;
    settled = true;
    final captured = buffer.toBytesOrNull();
    guardCapture(() => onError(error, st, captured, buffer.totalBytes));
  }

  void finishCancel() {
    if (settled) return;
    settled = true;
    final captured = buffer.toBytesOrNull();
    guardCapture(() => onCancel(captured, buffer.totalBytes));
  }

  controller = StreamController<List<int>>(
    onListen: () {
      Future<void> deliveryChain = Future<void>.value();

      void handleChunk(List<int> chunk) {
        if (settled || controller.isClosed) return;
        controller.add(chunk);
        buffer.add(chunk);
      }

      sub = source.listen(
        (chunk) {
          final delay = chunkDelay?.call(chunk.length) ?? Duration.zero;
          if (delay > Duration.zero) {
            try {
              if (!isCanceled) sub.pause();
            } catch (_) {}
            deliveryChain = deliveryChain.then((_) async {
              await Future<void>.delayed(delay);
              if (!isCanceled && !controller.isClosed) {
                handleChunk(chunk);
                try {
                  if (sub.isPaused && !isCanceled) {
                    sub.resume();
                  }
                } catch (_) {}
              }
            }).catchError((_) {});
          } else {
            deliveryChain = deliveryChain.then((_) {
              if (!isCanceled && !controller.isClosed) {
                handleChunk(chunk);
              }
            }).catchError((_) {});
          }
        },
        onDone: () {
          deliveryChain = deliveryChain.then((_) {
            if (!isCanceled) {
              finishDone();
              controller.close();
            }
          }).catchError((_) {});
        },
        onError: (Object e, StackTrace st) {
          deliveryChain = deliveryChain.then((_) {
            if (!isCanceled) {
              finishError(e, st);
              controller.addError(e, st);
              controller.close();
            }
          }).catchError((_) {});
        },
        cancelOnError: false,
      );
    },
    onPause: () {
      if (!isCanceled) {
        try {
          sub.pause();
        } catch (_) {}
      }
    },
    onResume: () {
      if (!isCanceled) {
        try {
          if (sub.isPaused) {
            sub.resume();
          }
        } catch (_) {}
      }
    },
    onCancel: () {
      isCanceled = true;
      finishCancel();
      return sub.cancel();
    },
  );

  return controller.stream;
}