declarePullSubscriber method
- Object keyExpr, {
- ChannelKind kind = ChannelKind.ring,
- int capacity = 256,
- Locality? allowedOrigin,
- bool retainPayload = false,
Declares a pull subscriber on the given keyExpr.
Returns a PullSubscriber that buffers samples in a bounded channel of
the given kind and capacity. Use PullSubscriber.tryRecv to poll
without waiting. Call PullSubscriber.close when done.
kind chooses what happens when the buffer fills:
ChannelKind.ring (the default) drops the oldest sample and never
blocks the publisher; ChannelKind.fifo keeps every sample and blocks
the publisher instead.
⚠️ Never publish into a full fifo from this same session. The delivery happens on the publisher's own thread, so one thread acting as both producer and only consumer blocks inside the put permanently — measured at zenoh-c 1.8.0 and unrecoverable, because the call is a synchronous FFI call. Use a second session for the publisher, or ChannelKind.ring. See ChannelKind.fifo.
The kinds also differ at the END of the channel's life, and the difference is canon's own: when the producer dies, a fifo hands over the samples it still holds before reporting disconnected, while a ring discards them. Either way the drain window closes at PullSubscriber.close — drain only if the residue matters.
⚠️ That used to read "drain before you close", and it silently carried a CORRECTNESS claim it could not honour — closing an undrained, overflowing fifo once hung the calling isolate permanently. Fixed at the source; and draining could never have prevented it from one isolate anyway. What still matters is the teardown ORDER: close the pull handle before closing its session when a fifo may be in overflow. A session closed first still stalls — a measured residual of canon's own session teardown that this binding does not fix.
capacity is the channel's bound. Both this and kind are
binding-decided defaults, not canon's: canon forces the caller to
choose both — its constructors take a raw size_t and there is no
options-default to defer to — so there is no "canon decides" value that
omitting them could mean. ChannelKind.ring preserves every existing
caller's behaviour byte-for-byte and matches canon's own z_pull.c;
256 is the shipped default, unchanged.
A capacity of 0 declares and delivers on both kinds — measured at zenoh-c 1.8.0, not promised by canon, which documents nothing about capacity anywhere. ⚠️ On ChannelKind.fifo, capacity 0 is a rendezvous rather than a one-slot buffer, so PullSubscriber.tryRecv works there but PullSubscriber.recv cannot — see PullSubscriber.recv.
allowedOrigin restricts whose traffic this declaration accepts.
Omitting it — or passing null — means canon decides, which is
Locality.any. See Locality.
Throws ArgumentError if capacity is negative, or too large for this
target's size_t — outside canon's domain, refused rather than
silently transformed.
Throws ZenohException if the key expression is invalid.
Throws StateError if the session has been closed.
Implementation
PullSubscriber declarePullSubscriber(
Object keyExpr, {
ChannelKind kind = ChannelKind.ring,
int capacity = 256,
Locality? allowedOrigin,
/// Whether each pulled sample carries a retained
/// [Sample.payloadZBytes]. Off by default.
bool retainPayload = false,
}) {
// BEFORE ANY NATIVE CALL. A negative is outside canon's `size_t` domain
// entirely, and the carriage would otherwise reinterpret it as an
// enormous unsigned capacity -- a silent transform, not a refusal.
//
// `ArgumentError`, not `ZenohException`: nothing in zenoh failed here.
// The caller passed a value canon has no way to represent, and canon was
// never asked. (The shim carries the same check as a structural
// backstop, plus the `> SIZE_MAX` half that only an ILP32 target can
// reach; a 64-bit Dart int cannot exceed a 64-bit `size_t`, so this side
// has only the one boundary to guard.)
//
// No upper bound is invented. Rejecting a large-but-representable
// capacity would narrow canon's surface on no canon-intrinsic ground --
// the same reasoning that keeps zero.
if (capacity < 0) {
throw ArgumentError.value(
capacity,
'capacity',
'must be non-negative',
);
}
return _withKeyExprArg(keyExpr, 'keyExpr', (loanedSession, loanedKe) {
// ALLOCATE-LAST: both slots are claimed only after the key expression
// has been accepted. They used to be claimed before it, so a rejected
// key expression stranded them.
final subscriberHandle = calloc<Uint8>(bindings.zd_subscriber_sizeof());
// The two handler types are distinct, so the slot is sized for the kind
// we are about to declare -- and released through the same kind.
final handlerHandle = calloc<Uint8>(
bindings.zd_pull_handler_sizeof(kind.value),
);
// The readiness channel behind `PullSubscriber.recv()`. The shim posts
// an int64 ping when an armed waiter should look again, and a null
// sentinel from the closure's drop when the producer is gone.
final receivePort = ReceivePort();
// Our out-cell for a SHIM-owned block: the shim mallocs the tee context
// and `zd_pull_tee_drop` frees it. This cell is ours, and the outer
// `finally` below encloses every statement that can throw.
final teeOut = calloc<Pointer<Uint8>>();
try {
final rc = bindings.zd_declare_pull_subscriber(
subscriberHandle,
handlerHandle,
teeOut,
loanedSession.cast(),
loanedKe.cast(),
kind.value,
capacity,
allowedOrigin?.value ?? -1,
receivePort.sendPort.nativePort,
);
if (rc != 0) {
receivePort.close();
calloc
..free(subscriberHandle)
..free(handlerHandle);
// The declare channel's return space is SPLIT, and it is mapped
// here, once, at the single call site. Positives are the shim's own
// -- they start at 10 because canon owns 0-and-negative on this
// channel and a shim-owned -1 would let a canon EINVAL masquerade
// as our failure. Negatives are canon's, passed through with
// canon's own code.
if (rc == 10) {
throw ArgumentError.value(
capacity,
'capacity',
"must be non-negative and within this platform's size_t range",
);
}
if (rc == 11) {
throw ZenohException(
'Failed to allocate pull subscriber state',
rc,
);
}
throw ZenohException('Failed to declare pull subscriber', rc);
}
return PullSubscriber(
subscriberHandle,
handlerHandle,
teeOut.value,
receivePort,
keyExprString(keyExpr, 'keyExpr'),
kind,
retainPayload,
);
} finally {
calloc.free(teeOut);
}
});
}