tryRecv method

RecvResult<Query> tryRecv()

Tries to take one query, without waiting.

Returns canon's own three-way outcome, undiluted:

• RecvData — a query was taken out of the buffer. Reply to it and then call Query.dispose, exactly as on the stream path.
• RecvEmpty — the queryable is alive and nothing is buffered right now. Back off and call again.
• RecvDisconnected — the producing end is gone: this queryable was undeclared, or its session closed. Terminal and sticky.

Throws ZenohException if the call itself failed — an allocation sized by the requester could not be satisfied. That is a fault, not a channel state, so it is thrown rather than returned.

Throws StateError if this queryable 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<Query> tryRecv() {
  if (_closed) throw StateError('PullQueryable has been closed');

  // THE STASH IS CONSULTED FIRST, ahead of the native channel.
  //
  // A query that completed the [stream] drive loop's pull into a paused or
  // cancelled subscription is held in a one-slot stash rather than disposed,
  // and this is its retrieval exit -- the one that keeps it REPLIABLE.
  // Taking it before the channel preserves the ordering [recv]'s own dartdoc
  // publishes: an interleaved `tryRecv` "is fine and *wins*".
  //
  // A caller who never touched [stream] has no gate, 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 outQuery = calloc<Int64>();
  final outKeyExpr = calloc<Pointer<Uint8>>();
  final outKeyExprLen = calloc<Size>();
  final outParameters = calloc<Pointer<Uint8>>();
  final outParametersLen = calloc<Size>();
  final outPayload = calloc<Pointer<Uint8>>();
  final outPayloadLen = calloc<Size>();
  final outAttachment = calloc<Pointer<Uint8>>();
  final outAttachmentLen = calloc<Size>();
  final outEncoding = calloc<Pointer<Char>>();
  final outEncodingLen = calloc<Size>();
  final outAccepts = calloc<Int8>();

  // 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> parametersPtr = nullptr;
  Pointer<Uint8> payloadPtr = nullptr;
  Pointer<Uint8> attachmentPtr = nullptr;
  Pointer<Char> encodingPtr = nullptr;

  try {
    final rc = bindings.zd_query_channel_try_recv(
      _handlerHandle,
      _kind.value,
      outQuery,
      outKeyExpr.cast(),
      outKeyExprLen,
      outParameters.cast(),
      outParametersLen,
      outPayload.cast(),
      outPayloadLen,
      outAttachment.cast(),
      outAttachmentLen,
      outEncoding.cast(),
      outEncodingLen,
      outAccepts,
    );

    // Canon's own codes, passed through by the shim and preserved here.
    // These two are STATES, not failures.
    if (rc == 1) return const RecvDisconnected<Query>();
    if (rc == 2) return const RecvEmpty<Query>();
    if (rc != 0) {
      throw ZenohException('Failed to receive from query channel', rc);
    }

    keyExprPtr = outKeyExpr.value;
    parametersPtr = outParameters.value;
    payloadPtr = outPayload.value;
    attachmentPtr = outAttachment.value;
    encodingPtr = outEncoding.value;

    // Both length-carried, never strlen-measured: the key expression grammar
    // and the selector's parameters segment each admit an interior NUL, and
    // canon carries one byte-exact.
    final keyExprStr = utf8.decode(
      keyExprPtr.asTypedList(outKeyExprLen.value),
      allowMalformed: true,
    );
    final parametersStr = utf8.decode(
      parametersPtr.asTypedList(outParametersLen.value),
      allowMalformed: true,
    );

    // Empty != absent on all three: the shim reports an absent value as a
    // null pointer and a present-but-empty one as a non-null pointer at
    // length 0, so the discriminator is the POINTER.
    final payloadBytes = payloadPtr == nullptr
        ? null
        : Uint8List.fromList(payloadPtr.asTypedList(outPayloadLen.value));
    final attachmentBytes = attachmentPtr == nullptr
        ? null
        : Uint8List.fromList(
            attachmentPtr.asTypedList(outAttachmentLen.value),
          );

    // Length-carried, exactly like the key expression and the parameters
    // above: a rendered MIME string is an arbitrary byte sequence and canon
    // carries an interior NUL in one byte-exact. NULL still means absent --
    // canon returns no encoding when the requester set none and sent no
    // payload — while a present-but-empty one is a non-NULL pointer at
    // length 0.
    final encodingBytes = encodingPtr == nullptr
        ? null
        : Uint8List.fromList(
            encodingPtr.cast<Uint8>().asTypedList(outEncodingLen.value),
          );

    return RecvData(
      Query(
        handle: outQuery.value,
        keyExpr: keyExprStr,
        parameters: parametersStr,
        payloadBytes: payloadBytes,
        attachmentBytes: attachmentBytes,
        encoding: encodingBytes == null
            ? null
            : utf8.decode(encodingBytes, allowMalformed: true),
        encodingBytes: encodingBytes,
        acceptsReplies: ReplyKeyExpr.fromWire(outAccepts.value),
      ),
    );
  } finally {
    if (keyExprPtr != nullptr) malloc.free(keyExprPtr.cast());
    if (parametersPtr != nullptr) malloc.free(parametersPtr.cast());
    if (payloadPtr != nullptr) malloc.free(payloadPtr.cast());
    if (attachmentPtr != nullptr) malloc.free(attachmentPtr.cast());
    if (encodingPtr != nullptr) malloc.free(encodingPtr.cast());
    calloc
      ..free(outQuery)
      ..free(outKeyExpr)
      ..free(outKeyExprLen)
      ..free(outParameters)
      ..free(outParametersLen)
      ..free(outPayload)
      ..free(outPayloadLen)
      ..free(outAttachment)
      ..free(outAttachmentLen)
      ..free(outEncoding)
      ..free(outEncodingLen)
      ..free(outAccepts);
  }
}