Python Streaming API¶

Create a Publisher and connect to endpoint's NATS broker.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Publisher, CF64
>>> with Publisher("nats://127.0.0.1:4222/t19102", CF64) as pub:
...     type(pub).__name__
'Publisher'

Call :meth:close on context-manager exit.

Broadcast one block of samples to all connected Subscribers.

Constructs a dp_header_t with the supplied metadata (plus a per-socket monotonically increasing sequence number), stages one [header][raw sample bytes] buffer (NATS has no scatter/ gather send, so header and data are copied into one contiguous buffer first), and publishes it. The call releases the GIL while blocked in the underlying NATS client, so other Python threads can run concurrently.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Publisher, CF64
>>> import numpy as np, time
>>> pub = Publisher("nats://127.0.0.1:4222/iq", CF64)
>>> pub.send(np.ones(4, dtype=np.complex128),
...          sample_rate=int(48000),
...          center_freq=int(433e6))
>>> pub.close()

Shut down gracefully: stop new work, finish, flush, close.

This waits for the connection to close, which is the part worth having: the underlying drain returns immediately and finishes in the background, so a process that exits when it returns abandons exactly the work the drain was for.

Drain last, after you have stopped producing. It cannot be reversed, and a send issued while one is in progress races its phases -- it may slip through, or be refused. Because this waits, a single-threaded caller need not reason about that: afterwards a send raises RuntimeError deterministically.

Against :meth:flush: flush asks whether the server has what you published and leaves the publisher usable; drain ends it. A drained shutdown needs no flush of its own -- the drain's final phase is that flush.

Size timeout_ms to the slowest thing the drain waits for, with margin. doppler's receive is synchronous, so there is no handler to finish and the wait is dominated by flushing what is buffered; the default is generous for a link that is keeping up.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Publisher, CF64
>>> pub = Publisher("nats://127.0.0.1:4222/iq", CF64)
>>> pub.drain()      # stopped producing, so shut down
>>> pub.close()

Tell subscribers the stream has ended.

A subscriber's :meth:Subscriber.recv raises :class:EOFError instead of waiting out a timeout -- which means only "nothing yet", and cannot be told from "nothing ever" without this.

Send it before :meth:drain, not after: a drain cannot be reversed and refuses sends once it reaches its publish-flushing phase. The order is stop producing, send_eos(), drain(), close().

Examples:

>>> from doppler.stream import Publisher, CF32
>>> pub = Publisher("nats://127.0.0.1:4222/demo", CF32)
>>> pub.send_eos()
>>> pub.drain()

Wait until the server has everything published so far.

:meth:send hands the frame to the client and returns; the client writes it in the background, so "the send returned" is not "the server has it". This waits for a round trip.

You do not need it before :meth:close: the NATS client flushes what is buffered when the connection closes. But it does so best-effort with a 500 ms cap and no way to report failure, so a backlog that cannot drain in half a second is dropped silently -- and on a link slower than loopback that is not a large backlog. Call this when losing the tail would matter, and you get a budget you chose and an answer you can act on. It is also the only way to ask the question without closing.

Parameters:

Raises:

Examples:

>>> import numpy as np
>>> from doppler.stream import Publisher, CF64
>>> with Publisher("nats://127.0.0.1:4222/iq", CF64) as pub:
...     pub.send(np.zeros(8, dtype=np.complex128))
...     pub.flush()

Destroy the underlying NATS handle and release all resources.

Calls dp_pub_destroy(). Safe to call multiple times — subsequent calls are no-ops. After close() the object must not be used for sending.

Examples:

>>> from doppler.stream import Publisher, CF64
>>> pub = Publisher("nats://127.0.0.1:4222/t19103", CF64)
>>> pub.close()
>>> pub.close()  # idempotent — no error

Create a Subscriber socket and connect to endpoint.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Subscriber
>>> with Subscriber("nats://127.0.0.1:4222/t19106") as sub:
...     type(sub).__name__
'Subscriber'

Call :meth:close on context-manager exit.

Receive one signal frame from the connected Publisher.

Blocks until a frame arrives or the optional timeout expires. The returned NumPy array is a zero-copy view into the received NATS message buffer; the buffer is freed when the array is garbage-collected.

Parameters:

Returns:

Raises:

Examples:

>>> from doppler.stream import Subscriber
>>> with Subscriber("nats://127.0.0.1:4222/iq") as sub:
...     samples, hdr = sub.recv(timeout_ms=1000)
...     print(samples.dtype, hdr["sample_rate"])

What reassembling chunked frames has given up on this subscriber.

A frame above the server's max_payload (1 MiB by default) arrives as chunks, and recv() rebuilds one frame at a time. A frame it can no longer complete is given up and counted here rather than hidden: the next frame still arrives intact. A frame recv() DOES return is complete and exact. The counts run from the subscriber's creation and are never reset. Read them from the thread that calls recv(): a recv() on another thread releases the GIL and updates them without a lock.

Returns:

Raises:

Examples:

>>> from doppler.stream import Subscriber
>>> sub = Subscriber("nats://127.0.0.1:4222/t19124")
>>> sub.reasm_stats()
{'abandoned': 0, 'rejected': 0, 'mid_frame_timeouts': 0}
>>> sub.close()

Destroy the underlying NATS handle and release all resources.

Calls dp_sub_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Subscriber
>>> sub = Subscriber("nats://127.0.0.1:4222/t19107")
>>> sub.close()
>>> sub.close()  # idempotent — no error

Create a Push socket and bind to endpoint.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Push, CF64
>>> with Push("nats://127.0.0.1:4222/t19110", CF64) as push:
...     type(push).__name__
'Push'

Call :meth:close on context-manager exit.

Durably publish one block of samples to the work-queue.

Blocks until the broker acks that the frame is persisted (and replicated, on a clustered broker) — not until a Pull worker is ready; the frame waits in the queue for the next available worker. Releases the GIL while blocked on the broker round trip.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> import numpy as np, time
>>> push = Push("nats://127.0.0.1:4222/work", CF64)
>>> pull = Pull("nats://127.0.0.1:4222/work")
>>> time.sleep(0.05)
>>> push.send(np.array([1+2j, 3+4j],
...           dtype=np.complex128),
...           sample_rate=int(48000))
>>> samples, hdr = pull.recv(timeout_ms=2000)
>>> push.close(); pull.close()

Destroy the work-queue stream backing this endpoint.

Every frame still in the queue goes with it, acked or not.

Administrative, and deliberately never automatic: a work queue is shared infrastructure and outliving any one producer is the feature, so closing a Push must not end it. Call this only when you own the queue's lifetime — a test that invented the subject, or a tool tearing down what it provisioned. A work queue drops a frame only when a consumer acks it, so frames nobody consumed are kept forever otherwise (doppler#1136).

Only a Push provisions the stream, so a Pull cannot attach to the endpoint again until another Push re-creates it.

Raises:

Examples:

>>> from doppler.stream import Push, CF64
>>> push = Push("nats://127.0.0.1:4222/t19112", CF64)
>>> push.delete_stream()
>>> push.close()

Destroy the underlying NATS handle and release all resources.

Calls dp_push_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Push, CF64
>>> push = Push("nats://127.0.0.1:4222/t19111", CF64)
>>> push.close()
>>> push.close()  # idempotent — no error

Create a Pull socket and connect to endpoint.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> _ = Push("nats://127.0.0.1:4222/t19114", CF64)  # provisions queue
>>> with Pull("nats://127.0.0.1:4222/t19114") as pull:
...     type(pull).__name__
'Pull'

Call :meth:close on context-manager exit.

Receive one signal frame from the connected Push socket.

Blocks until a frame arrives or the optional timeout expires. The returned NumPy array is a zero-copy view into the received NATS message buffer; the buffer is freed when the array is garbage-collected.

Parameters:

Returns:

Raises:

Examples:

>>> from doppler.stream import Pull
>>> with Pull("nats://127.0.0.1:4222/work") as pull:
...     samples, hdr = pull.recv(timeout_ms=1000)
...     print(samples.dtype, hdr["num_samples"])

Acknowledge a frame consumed from the JetStream work-queue.

Delivery is at-least-once: a frame stays pending until acked and is redelivered to another worker if this one dies first. Pass the array returned by :meth:recv once it has been fully processed, then drop the array. Ack before :meth:close: once the Pull that received the frame is closed, nothing can be sent.

A frame that did not come from a Pull -- a :class:Subscriber's, say -- has nothing to acknowledge: its ack is a no-op that never raises, whether or not its receiver is closed.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Pull
>>> pull = Pull("nats://localhost:4222/work")
>>> samples, hdr = pull.recv(timeout_ms=1000)
>>> pull.ack(samples)

Destroy the underlying NATS handle and release all resources.

Calls dp_pull_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Push, Pull, CF64
>>> _ = Push("nats://127.0.0.1:4222/t19115", CF64)  # provisions queue
>>> pull = Pull("nats://127.0.0.1:4222/t19115")
>>> pull.close()
>>> pull.close()  # idempotent — no error

Create a Requester socket and connect to endpoint.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Requester, CF64
>>> with Requester("nats://127.0.0.1:4222/t19118", CF64) as req:
...     type(req).__name__
'Requester'

Call :meth:close on context-manager exit.

Publish a request frame with this Requester's reply-to inbox.

Use strictly alternating with :meth:recv (send, recv, send, recv, ...) — sending again before consuming the previous reply works mechanically but leaves an unread message in the inbox for the next :meth:recv to pick up, which will desync the request/reply pairing.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Requester, CF64
>>> import numpy as np
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> req.send(np.ones(4, dtype=np.complex128),
...          sample_rate=int(1e6))

Receive the reply frame from the :class:Replier.

Waits on this Requester's reply-to inbox. Call after :meth:send; calling without a prior send just blocks (or times out) on an empty inbox rather than raising immediately.

Parameters:

Returns:

Raises:

Examples:

>>> from doppler.stream import Requester, CF64
>>> req = Requester("nats://127.0.0.1:4222/ctrl", CF64)
>>> req.send(np.ones(4, dtype=np.complex128))
>>> reply, hdr = req.recv(timeout_ms=2000)

Destroy the underlying NATS handle and release all resources.

Calls dp_req_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Requester, CF64
>>> req = Requester("nats://127.0.0.1:4222/t19119", CF64)
>>> req.close()
>>> req.close()  # idempotent — no error

Create a Replier socket and bind to endpoint.

Parameters:

Return self for use as a context manager.

Examples:

>>> from doppler.stream import Replier, CF64
>>> with Replier("nats://127.0.0.1:4222/t19122", CF64) as rep:
...     type(rep).__name__
'Replier'

Call :meth:close on context-manager exit.

Receive one request frame from a :class:Requester.

Blocks until a request arrives or the timeout expires. Captures the request's reply-to inbox so the next :meth:send reaches the right Requester.

Parameters:

Returns:

Raises:

Examples:

>>> from doppler.stream import Replier, CF64
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> request, hdr = rep.recv(timeout_ms=5000)
>>> rep.send(request,
...          sample_rate=hdr["sample_rate"])

Publish the reply to the request's captured reply-to inbox.

Must be called after :meth:recv; calling without a prior recv raises :exc:RuntimeError immediately (there is no reply-to target yet). After this call returns, the Replier is ready to recv the next request.

Parameters:

Raises:

Examples:

>>> from doppler.stream import Replier, CF64
>>> import numpy as np
>>> rep = Replier("nats://127.0.0.1:4222/ctrl", CF64)
>>> request, hdr = rep.recv(timeout_ms=5000)
>>> rep.send(np.zeros_like(request),
...          sample_rate=hdr["sample_rate"])

Destroy the underlying NATS handle and release all resources.

Calls dp_rep_destroy(). Safe to call multiple times.

Examples:

>>> from doppler.stream import Replier, CF64
>>> rep = Replier("nats://127.0.0.1:4222/t19123", CF64)
>>> rep.close()
>>> rep.close()  # idempotent — no error