createSampleChannel static method

SampleChannel createSampleChannel({ bool retainPayload = false, })

Sets up the ReceivePort / StreamController pair that parses incoming NativePort sample messages into Sample objects.

Returns a SampleChannel. The caller passes channel.receivePort.sendPort.nativePort to the C shim, calls SampleChannel.abandon if the declaration itself fails, and SampleChannel.closeAndDrain on close.

retainPayload must match what was passed to the shim. It arms the delivered-tracking that releases retained handles nobody received; with retention off nothing is tracked and the stream is the bare controller.

Implementation

static SampleChannel createSampleChannel({bool retainPayload = false}) {
  final receivePort = ReceivePort();
  final controller = StreamController<Sample>();
  final channel = SampleChannel._(receivePort, controller, retainPayload);

  receivePort.listen((dynamic message) {
    if (message == null) {
      receivePort.close();
      unawaited(controller.close());
    } else if (message is List) {
      // The key expression arrives length-carried, not as a C string, so an
      // interior NUL survives -- the grammar permits one and canon carries
      // it byte-exact. Decoded leniently like every other display string
      // here: an invalid sequence becomes U+FFFD rather than throwing.
      final keyExprBytes = message[0] as Uint8List;
      final keyExpr = utf8.decode(keyExprBytes, allowMalformed: true);
      // ⛔ A VALUE THAT COULD NOT BE CONVERTED IS AN ERROR, NEVER AN
      // EMPTY SUCCESS. The shim used to post a zero-length buffer when
      // canon refused the conversion, which is indistinguishable from a
      // legitimately empty value — and empty is legitimate on every one of
      // these paths. Following the shipped call-failure rule: the error
      // goes to the error channel and the stream KEEPS RUNNING, because a
      // conversion failure is a failed call and not a dead channel.
      final payloadFailure = undecodableRc(message[1]);
      if (payloadFailure != null) {
        controller.addError(undecodableError('a payload', payloadFailure));
        return;
      }
      final attachmentFailure = undecodableRc(message[3]);
      if (attachmentFailure != null) {
        controller.addError(
          undecodableError('an attachment', attachmentFailure),
        );
        return;
      }
      final payloadBytes = message[1] as Uint8List;
      final kind = message[2] as int;
      final attachmentBytes = message[3] as Uint8List?;
      // The encoding arrives length-carried too, for the same reason the key
      // expression does: a rendered MIME string is an arbitrary byte
      // sequence and canon carries an interior NUL across the wire
      // byte-exact. The lenient String is a display view; encodingBytes is
      // the ground truth.
      final encodingBytes = message.length > 4
          ? message[4] as Uint8List?
          : null;
      // Slice 2: QoS/timestamp metadata (length-guarded, defensive).
      final timestampBytes = message.length > 5
          ? message[5] as Uint8List?
          : null;
      final priorityRaw = message.length > 6 ? message[6] as int : null;
      final congestionRaw = message.length > 7 ? message[7] as int : null;
      final expressRaw = message.length > 8 ? message[8] as int : null;
      // Seed [10a] element 9: the retained payload handle image, or null
      // when this carrier did not opt in. Length-guarded like every element
      // above it, so a retention-off carrier and an older message shape both
      // land on null rather than tripping a range error.
      final retainedImage = message.length > 9
          ? message[9] as Uint8List?
          : null;

      final sample = Sample(
        keyExpr: keyExpr,
        payload: utf8.decode(payloadBytes, allowMalformed: true),
        payloadBytes: payloadBytes,
        kind: kind == 0 ? SampleKind.put : SampleKind.delete,
        attachment: attachmentBytes != null
            ? utf8.decode(attachmentBytes, allowMalformed: true)
            : null,
        attachmentBytes: attachmentBytes,
        encoding: encodingBytes != null
            ? utf8.decode(encodingBytes, allowMalformed: true)
            : null,
        encodingBytes: encodingBytes,
        timestamp: timestampBytes != null
            ? Timestamp.fromRaw(timestampBytes)
            : null,
        // Wire priority is 1..7 -> Priority.fromWire (send symmetric).
        priority: priorityRaw != null
            ? Priority.fromWire(priorityRaw)
            : Priority.data,
        // Wire congestion is 0/1/2 -> CongestionControl.fromWire.
        congestionControl: congestionRaw != null
            ? CongestionControl.fromWire(congestionRaw)
            : CongestionControl.drop,
        express: expressRaw != null && (expressRaw != 0),
        payloadZBytes: ZBytes.fromPostedImage(retainedImage),
      );

      if (controller.isClosed) {
        // DRAIN BRANCH: closeAndDrain() already ran and this message was
        // still sitting in the port queue. Nobody can ever receive it, so
        // release its retained handle here instead of orphaning it.
        sample.payloadZBytes?.dispose();
        return;
      }

      final retained = sample.payloadZBytes;
      if (retained != null) {
        channel._undelivered.add(retained);
      }
      controller.add(sample);
    }
  });

  return channel;
}