tryRecv method
Tries to take one reply, without waiting.
Returns canon's own three-way outcome, undiluted:
- RecvData — a reply was taken out of the buffer. It carries the same Reply the stream path delivers, ok/error discriminant included.
- RecvEmpty — the query is still in flight and nothing is buffered right now. Back off and call again.
- RecvDisconnected — the query has completed. Terminal and sticky: every subsequent call reports it again. Stop polling.
This is the loop canon's own z_non_blocking_get.c writes:
loop:
while (true) {
switch (replies.tryRecv()) {
case RecvData(:final value):
handle(value);
case RecvEmpty():
await Future<void>.delayed(backoff);
case RecvDisconnected():
break loop;
}
}
⚠️ A ring channel recovers nothing once the query has completed — see ChannelKind.ring's discard-at-disconnect behaviour. A get's completion normally follows its replies immediately, so a ring reply channel has to be polled while the query is in flight. A fifo hands its buffer over first and only then reports RecvDisconnected.
⚠️ Timeout expiry is not silent. When the get's clock runs out, canon
delivers an ERROR reply whose payload is 'Timeout' through this channel,
and the channel disconnects after it. A consumer that treats every
RecvDisconnected as a clean end will report a failed query as a
successful empty one.
Throws ZenohException if the call itself failed — an allocation sized by the replier could not be satisfied. That is a fault, not a channel state, so it is thrown rather than returned.
⚠️ Not usable on a fifo channel of capacity 0
Measured through this stack, per kind: a capacity-0 fifo is a
rendezvous, not a one-slot buffer — it is full when it is empty. A
delivery therefore blocks waiting for a concurrent consumer, and the
readiness signal this future waits on is only raised after that delivery
returns. The parked recv() is the only consumer that could release it,
so neither side moves and the future does not resolve.
Use tryRecv at capacity 0 — a synchronous poll is the concurrent consumer, so it releases the delivery and works normally. A capacity-0 ring is unaffected (it never blocks its producer), and so is every capacity of 1 or more on either kind.
Throws StateError if this handle has been disposed. 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<Reply> tryRecv() {
if (_closed) throw StateError('PullReplies has been disposed');
final outIsOk = calloc<Int8>();
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>();
final outReplierZid = calloc<Pointer<Uint8>>();
final outReplierEid = calloc<Int64>();
// Seed [10a]: the retained-payload slot, only when this carrier opted in.
// Sized from zd_bytes_sizeof() at run time, never a literal: 40 unstable,
// 32 stable.
final retainSlot = _retainPayload
? calloc.allocate<Uint8>(bindings.zd_bytes_sizeof())
: nullptr;
final outHasRetained = calloc<Int32>();
// Whether a ZBytes took ownership of [retainSlot]. Until it does the slot
// is ours, and the finally releases it on every other path.
var slotAdopted = false;
// The buffers the SHIM mallocs and hands over. Captured out here so the
// finally releases them on every path, including a throw between the rc
// check and the reads.
Pointer<Uint8> keyExprPtr = nullptr;
Pointer<Uint8> payloadPtr = nullptr;
Pointer<Char> encodingPtr = nullptr;
Pointer<Uint8> attachmentPtr = nullptr;
Pointer<Uint8> timestampPtr = nullptr;
Pointer<Uint8> replierZidPtr = nullptr;
try {
final rc = bindings.zd_reply_channel_try_recv(
_handlerHandle,
_kind.value,
outIsOk,
outKeyExpr.cast(),
outKeyExprLen,
outPayload.cast(),
outPayloadLen,
outKind,
outEncoding.cast(),
outEncodingLen,
outAttachment.cast(),
outAttachmentLen,
outTimestamp.cast(),
outPriority,
outCongestion,
outExpress,
outReplierZid.cast(),
outReplierEid,
retainSlot,
outHasRetained,
);
// Canon's own codes, passed through by the shim and preserved here.
// These two are STATES, not failures: the only positive result codes in
// the zenoh-c API, deliberately outside its negative error space.
if (rc == 1) return const RecvDisconnected<Reply>();
if (rc == 2) return const RecvEmpty<Reply>();
if (rc != 0) {
throw ZenohException('Failed to receive from reply channel', rc);
}
keyExprPtr = outKeyExpr.value;
payloadPtr = outPayload.value;
encodingPtr = outEncoding.value;
attachmentPtr = outAttachment.value;
timestampPtr = outTimestamp.value;
replierZidPtr = outReplierZid.value;
final payloadLen = outPayloadLen.value;
final payloadBytes = (payloadLen > 0 && payloadPtr != nullptr)
? Uint8List.fromList(payloadPtr.asTypedList(payloadLen))
: Uint8List(0);
// Present on BOTH branches where the unstable API is compiled in; a null
// zid pointer is the absence discriminator, so the parse is identical on
// the stable variant rather than shape-dependent.
final replierId = replierZidPtr == nullptr
? null
: EntityGlobalId(
ZenohId(Uint8List.fromList(replierZidPtr.asTypedList(16))),
outReplierEid.value,
);
// Length-carried, exactly like the key expression below and for the same
// reason: a rendered MIME string is an arbitrary byte sequence and canon
// carries an interior NUL in one byte-exact. 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.
final encodingBytes = encodingPtr == nullptr
? null
: Uint8List.fromList(
encodingPtr.cast<Uint8>().asTypedList(outEncodingLen.value),
);
final encodingStr = encodingBytes == null
? null
: utf8.decode(encodingBytes, allowMalformed: true);
if (outIsOk.value == 0) {
return RecvData(
Reply.error(
ReplyError(
payloadBytes: payloadBytes,
payload: utf8.decode(payloadBytes, allowMalformed: true),
encoding: encodingStr,
encodingBytes: encodingBytes,
),
replierId: replierId,
),
);
}
// Empty != absent: a present-but-empty attachment comes back as a
// non-null pointer at length 0 (the shim mallocs >= 1 byte), so the
// discriminator is the POINTER, never the length.
final attachmentBytes = attachmentPtr == nullptr
? null
: Uint8List.fromList(
attachmentPtr.asTypedList(outAttachmentLen.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 = keyExprPtr == nullptr
? ''
: utf8.decode(
keyExprPtr.asTypedList(outKeyExprLen.value),
allowMalformed: true,
);
// Seed [10a]: adopt the retained slot. Ownership passes to the ZBytes,
// whose dispose() and finalizer both release with Dart's allocator --
// which is exactly why the slot is Dart-allocated.
ZBytes? retained;
if (retainSlot != nullptr && outHasRetained.value != 0) {
retained = ZBytes.fromNative(retainSlot.cast());
slotAdopted = true;
}
return RecvData(
Reply.ok(
Sample(
keyExpr: keyExprStr,
payload: utf8.decode(payloadBytes, allowMalformed: true),
payloadBytes: payloadBytes,
kind: outKind.value == 0 ? SampleKind.put : SampleKind.delete,
attachment: attachmentBytes == null
? null
: utf8.decode(attachmentBytes, allowMalformed: true),
attachmentBytes: attachmentBytes,
encoding: encodingStr,
encodingBytes: encodingBytes,
timestamp: timestampPtr == nullptr
? null
: Timestamp.fromRaw(
Uint8List.fromList(timestampPtr.asTypedList(24)),
),
priority: Priority.fromWire(outPriority.value),
congestionControl: CongestionControl.fromWire(outCongestion.value),
express: outExpress.value != 0,
payloadZBytes: retained,
),
replierId: replierId,
),
);
} finally {
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());
if (replierZidPtr != nullptr) malloc.free(replierZidPtr.cast());
calloc
..free(outIsOk)
..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)
..free(outReplierZid)
..free(outReplierEid);
if (retainSlot != nullptr && !slotAdopted) {
calloc.free(retainSlot);
}
}
}