recv method

Future<RecvResult<Query>> recv()

Waits for the next query.

Completes with:

• RecvData as soon as a query is available — immediately if one is already buffered, otherwise when the next one arrives.
• RecvDisconnected when the producing end goes: this queryable is closed, or its session closes. For a fifo, only after its buffered queries have been handed over first.

It never completes RecvEmpty, and it never hangs, for the same reason canon's own blocking recv is two-valued: it waits rather than reporting an empty buffer.

No thread is parked and no isolate is created — the native side signals readiness and this future consumes through the ordinary synchronous tryRecv.

Throws StateError if a recv() is already pending — one pull at a time per handle. An interleaved tryRecv is fine and wins: it reaches the channel first, and the pending recv() re-arms for the next arrival.

Throws StateError if this queryable has been closed.

Implementation

Future<RecvResult<Query>> recv() {
  if (_closed) throw StateError('PullQueryable has been closed');
  if (_pending != null) {
    throw StateError('a recv() is already pending on this PullQueryable');
  }

  final immediate = tryRecv();
  if (immediate is! RecvEmpty<Query>) {
    return Future<RecvResult<Query>>.value(immediate);
  }

  // REGISTER, THEN ARM, THEN LOOK AGAIN — the order closes the arm-vs-arrival
  // race. Dart is single-threaded, so no port message can be delivered into
  // the gap between the first look and the arming.
  final completer = Completer<RecvResult<Query>>();
  _pending = completer;
  bindings.zd_pull_tee_arm(_teeHandle);

  final afterArm = tryRecv();
  if (afterArm is! RecvEmpty<Query>) {
    _pending = null;
    // The arming stays set and the next delivery spends it on a single ping
    // that `_onWake` discards. Bounded at one, and self-clearing.
    return Future<RecvResult<Query>>.value(afterArm);
  }
  return completer.future;
}