tryRecv method
Tries to receive a sample, without waiting.
Returns canon's own three-way outcome, undiluted:
- RecvData -- a sample was taken out of the buffer.
- RecvEmpty -- the channel is alive and its buffer is empty right now. Back off and call again; a later call may well succeed.
- RecvDisconnected -- the producing end is gone (its subscriber was undeclared, or its session closed). Terminal and sticky: every subsequent call reports it again. Stop polling.
This is the loop canon's own z_non_blocking_get.c writes, and the
discriminant is what makes its exit condition expressible:
loop:
while (true) {
switch (pull.tryRecv()) {
case RecvData(:final value):
handle(value);
case RecvEmpty():
await Future<void>.delayed(backoff);
case RecvDisconnected():
break loop;
}
}
Throws ZenohException if the call itself failed -- an allocation sized by the remote publisher could not be satisfied. That is a fault, not a channel state, so it is thrown rather than returned: no switch site should have to handle conditions canon calls call failures.
Throws StateError if this subscriber has been closed. That guard is ours, not canon's, and it is load-bearing rather than defensive: loaning a dropped handler is undefined behaviour in canon, never an error it reports.
Implementation
RecvResult<Sample> tryRecv() {
if (_closed) throw StateError('PullSubscriber is closed');
// THE STASH IS CONSULTED FIRST, ahead of the native channel.
//
// A sample that completed the [stream] drive loop's pull into a paused or
// cancelled subscription is held in a one-slot stash rather than dropped,
// and this is its retrieval exit. Taking it before the channel is what
// preserves the ordering [recv]'s own dartdoc already publishes: an
// interleaved `tryRecv` "is fine and *wins*". Measured without this
// branch, three samples published after a cancel came back as two.
//
// A caller who never touched [stream] has no gate at all, so this reads
// null and the path below is byte-identical to what shipped.
final stashed = _gate?.takeStash();
if (stashed != null) return RecvData(stashed);
final outKeyexpr = calloc<Pointer<Uint8>>();
final outKeyexprLen = calloc<Size>();
final outPayload = calloc<Pointer<Uint8>>();
final outPayloadLen = calloc<Size>();
final outKind = calloc<Int8>();
final outEncoding = calloc<Pointer<Char>>();
final outEncodingLen = calloc<Size>();
final outAttachment = calloc<Pointer<Uint8>>();
final outAttachmentLen = calloc<Size>();
final outTimestamp = calloc<Pointer<Uint8>>();
final outPriority = calloc<Int8>();
final outCongestion = calloc<Int8>();
final outExpress = calloc<Int8>();
// Seed [10a]: the retained-payload slot. ALLOCATE-LAST and only when this
// carrier opted in — a retention-off pull pays nothing. The size comes
// from zd_bytes_sizeof() at run time, never a literal: it is 40 on the
// unstable build and 32 on stable.
final retainSlot = _retainPayload
? calloc.allocate<Uint8>(bindings.zd_bytes_sizeof())
: nullptr;
final outHasRetained = calloc<Int32>();
// The five buffers the SHIM mallocs and hands over. Captured here so the
// finally can release them: they used to be freed by straight-line code in
// the try body, which any throw between the rc check and those lines
// skipped -- all five leaked together.
Pointer<Uint8> keyExprPtr = nullptr;
Pointer<Uint8> payloadPtr = nullptr;
Pointer<Char> encodingPtr = nullptr;
Pointer<Uint8> attachmentPtr = nullptr;
Pointer<Uint8> timestampPtr = nullptr;
// Whether a ZBytes took ownership of [retainSlot]. Until it does, the slot
// is ours and the finally releases it -- an EMPTY or DISCONNECTED result,
// or any throw, must not leak it.
var slotAdopted = false;
try {
final rc = bindings.zd_pull_subscriber_try_recv(
_handlerHandle,
_kind.value,
outKeyexpr.cast(),
outKeyexprLen,
outPayload.cast(),
outPayloadLen,
outKind,
outEncoding.cast(),
outEncodingLen,
outAttachment.cast(),
outAttachmentLen,
outTimestamp.cast(),
outPriority,
outCongestion,
outExpress,
retainSlot,
outHasRetained,
);
// Canon's own codes, passed through by the shim and preserved here.
// These two are STATES, not failures: they are the only positive
// result codes in the zenoh-c API, deliberately outside its negative
// error space, and a consumer switches on them.
if (rc == 1) return const RecvDisconnected<Sample>();
if (rc == 2) return const RecvEmpty<Sample>();
// Anything else is the call itself failing. Today that is only the
// shim's -1, raised when one of the four remote-length-driven mallocs
// returns NULL; the shim has already released whatever it held, so
// there is nothing here to reclaim beyond the finally below. It used
// to be swallowed into the same `null` as an empty buffer, which made
// an out-of-memory indistinguishable from "nothing published yet".
if (rc != 0) {
throw ZenohException('Failed to receive from pull subscriber', rc);
}
// Extract fields from malloc'd pointers
keyExprPtr = outKeyexpr.value;
payloadPtr = outPayload.value;
encodingPtr = outEncoding.value;
attachmentPtr = outAttachment.value;
timestampPtr = outTimestamp.value;
// Length-carried, never strlen-measured: the key expression grammar
// permits an interior NUL and canon carries one byte-exact, so reading
// this buffer as a C string would silently truncate a real value.
final keyExprStr = utf8.decode(
keyExprPtr.asTypedList(outKeyexprLen.value),
allowMalformed: true,
);
final payloadLen = outPayloadLen.value;
Uint8List payloadBytes;
String payloadStr;
if (payloadLen > 0 && payloadPtr != nullptr) {
payloadBytes = Uint8List.fromList(payloadPtr.asTypedList(payloadLen));
payloadStr = utf8.decode(payloadBytes, allowMalformed: true);
} else {
payloadBytes = Uint8List(0);
payloadStr = '';
}
final kind = outKind.value;
// Length-carried, exactly like the key expression above and for the
// same reason: a rendered MIME string is an arbitrary byte sequence and
// canon carries an interior NUL in one byte-exact. Reading to the first
// NUL truncated it.
//
// ⚠️ This replaces a helper whose doc comment said the opposite by
// implication -- "only the encoding still arrives this way; the key
// expression is length-carried, because its domain genuinely does admit
// an interior NUL". The contrast was false: BOTH domains admit one. The
// reason it carried (ownership sweep F-R1, which probe-refuted the remote
// non-UTF-8 trigger for this field) is true and is about a different
// property -- a strict-decode CRASH, not length carriage.
//
// The decode stays LENIENT, which is the part of that helper worth
// keeping: an invalid sequence becomes U+FFFD rather than throwing
// `FormatException`, as on every other receive surface here.
// (`Pointer<Utf8>.toDartString()` decodes strictly and has no lenient
// mode, which is why this is spelled out rather than delegated.)
//
// Empty is not absent -- the shim allocates unconditionally, so a
// present-but-empty encoding is a NON-NULL pointer at length 0 and reads
// as '' here, matching the push path.
String? encodingStr;
Uint8List? encodingBytes;
if (encodingPtr != nullptr) {
encodingBytes = Uint8List.fromList(
encodingPtr.cast<Uint8>().asTypedList(outEncodingLen.value),
);
encodingStr = utf8.decode(encodingBytes, allowMalformed: true);
}
final attachmentLen = outAttachmentLen.value;
String? attachmentStr;
Uint8List? attachmentBytes;
// Empty != absent: a present-but-empty attachment comes back as a
// non-null pointer with len 0 (the C shim mallocs >= 1 byte so the
// pointer is non-null). Key on the pointer, not the length, so an empty
// attachment surfaces as a non-null empty Uint8List rather than null.
if (attachmentPtr != nullptr) {
attachmentBytes = Uint8List.fromList(
attachmentPtr.asTypedList(attachmentLen),
);
attachmentStr = utf8.decode(attachmentBytes, allowMalformed: true);
}
// Timestamp (nullable): present when the C side malloc'd a 24-byte image.
Timestamp? timestamp;
if (timestampPtr != nullptr) {
timestamp = Timestamp.fromRaw(
Uint8List.fromList(timestampPtr.asTypedList(24)),
);
}
// QoS: wire priority is 1..7 -> Priority.fromWire; congestion is
// 0/1 -> CongestionControl.fromWire; express is a 0/1 flag.
final priority = Priority.fromWire(outPriority.value);
final congestionControl = CongestionControl.fromWire(outCongestion.value);
final express = outExpress.value != 0;
// Seed [10a]: adopt the retained slot. The slot is Dart-allocated, which
// is exactly what ZBytes.dispose() and the finalizer both require, so
// ownership passes to the ZBytes and the finally must NOT free it.
ZBytes? retained;
if (retainSlot != nullptr && outHasRetained.value != 0) {
retained = ZBytes.fromNative(retainSlot.cast());
slotAdopted = true;
}
return RecvData(
Sample(
keyExpr: keyExprStr,
payload: payloadStr,
payloadBytes: payloadBytes,
kind: kind == 0 ? SampleKind.put : SampleKind.delete,
attachment: attachmentStr,
attachmentBytes: attachmentBytes,
encoding: encodingStr,
encodingBytes: encodingBytes,
timestamp: timestamp,
priority: priority,
congestionControl: congestionControl,
express: express,
payloadZBytes: retained,
),
);
} finally {
// The shim-transferred buffers first (see the declarations above), then
// our own out-parameter cells.
if (keyExprPtr != nullptr) malloc.free(keyExprPtr.cast());
if (payloadPtr != nullptr) malloc.free(payloadPtr.cast());
if (encodingPtr != nullptr) malloc.free(encodingPtr.cast());
if (attachmentPtr != nullptr) malloc.free(attachmentPtr.cast());
if (timestampPtr != nullptr) malloc.free(timestampPtr.cast());
calloc
..free(outKeyexpr)
..free(outKeyexprLen)
..free(outPayload)
..free(outPayloadLen)
..free(outKind)
..free(outEncoding)
..free(outEncodingLen)
..free(outAttachment)
..free(outAttachmentLen)
..free(outTimestamp)
..free(outPriority)
..free(outCongestion)
..free(outExpress)
..free(outHasRetained);
if (retainSlot != nullptr && !slotAdopted) {
calloc.free(retainSlot);
}
}
}