Skip to content

Streaming Examples

C: PUB/SUB

Requires a running NATS server: run one (e.g. nats-server -js).

dp_pub_*/dp_sub_* live in the optional stream archive, not the core library — an undefined reference to dp_pub_create means libdoppler_stream.a is missing from the link line. Compile any snippet on this page with both archives (either order works; the stream layer is pure C and adds only -lpthread):

PREFIX="$HOME/.local/doppler"
cc app.c -I "$PREFIX/include" \
   "$PREFIX/lib/libdoppler_stream.a" \
   "$PREFIX/lib/libdoppler.a" \
   -lm -lpthread -o app

(CMake and pkg-config faces of the same link line: C Quick Start.)

Transmitter

#include <doppler/stream/stream.h>
#include <complex.h>
#include <math.h>
#include <stdio.h>

int main(void) {
    dp_pub_t *tx = dp_pub_create("nats://127.0.0.1:4222/iq", CF64);

    double _Complex samples[1000];
    for (int i = 0; i < 1000; i++) {
        double phase = 2.0 * M_PI * 1000.0 * i / 1e6;
        samples[i] = cos(phase) + sin(phase) * _Complex_I;
    }

    int rc = dp_pub_send_cf64(tx, samples, 1000, 1e6, 2.4e9);
    if (rc != DP_OK)
        fprintf(stderr, "send refused: %s\n", dp_ctx_last_error(tx));
    dp_pub_destroy(tx);
    return rc == DP_OK ? 0 : 1;
}

Receiver

#include <doppler/stream/stream.h>
#include <stdio.h>

int main(void) {
    dp_sub_t *rx = dp_sub_create("nats://127.0.0.1:4222/iq");

    dp_msg_t *msg;
    dp_header_t hdr;

    dp_sub_recv(rx, &msg, &hdr);
    printf("Received %zu samples, type=%s\n",
           dp_msg_num_samples(msg),
           dp_sample_type_str(dp_msg_sample_type(msg)));

    dp_msg_free(msg);
    dp_sub_destroy(rx);
    return 0;
}

C: PUSH/PULL pipeline

The PUSH/PULL work-queue tier is backed by NATS JetStream, so pushed frames survive a restart of the consumer and are load-balanced round-robin across every connected Pull.

#include <doppler/stream/stream.h>
#include <stdio.h>

// Producer
dp_push_t *push = dp_push_create("nats://127.0.0.1:4222/work", CF64);
if (dp_push_send_cf64(push, samples, count, 1e6, 2.4e9) != DP_OK)
    fprintf(stderr, "send refused: %s\n", dp_ctx_last_error(push));
dp_push_destroy(push);

// Consumer
dp_pull_t *pull = dp_pull_create("nats://127.0.0.1:4222/work");
dp_msg_t *msg;
dp_header_t hdr;
dp_pull_recv(pull, &msg, &hdr);
double _Complex *data = (double _Complex *)dp_msg_data(msg);
// use data ...
dp_msg_ack(msg);   // processed: otherwise it is redelivered, and stays queued
dp_msg_free(msg);
dp_pull_destroy(pull);

For complete, runnable examples see native/examples/.

Python: Publisher / Subscriber

Requires a running nats-server (e.g. nats-server -js).

from doppler.stream import Publisher, Subscriber, CF64
import numpy as np

samples = np.array([1+2j, 3+4j, 5+6j], dtype=np.complex128)

with Publisher("nats://127.0.0.1:4222/iq", CF64) as pub:
    pub.send(samples, sample_rate=1e6, center_freq=2.4e9)
    pub.drain()  # close() alone flushes best-effort and reports nothing

with Subscriber("nats://127.0.0.1:4222/iq") as sub:
    data, hdr = sub.recv(timeout_ms=500)
    print(f"Got {hdr['num_samples']} samples, seq={hdr['sequence']}")

For a complete runnable example with live dashboard and graceful shutdown:

python -m doppler.examples.transmitter nats://127.0.0.1:4222/iq
python -m doppler.examples.receiver nats://127.0.0.1:4222/iq

Python: Push / Pull pipeline

The PUSH/PULL work-queue tier is backed by NATS JetStream: pushed frames are durably queued and load-balanced round-robin across every connected Pull worker.

from doppler.stream import Push, Pull, CF64
import numpy as np

samples = np.ones(512, dtype=np.complex128)

# Multiple Pull workers share frames round-robin via a JetStream
# work-queue consumer group.
with Push("nats://127.0.0.1:4222/work", CF64) as push:
    push.send(samples, sample_rate=1e6, center_freq=2.4e9)

with Pull("nats://127.0.0.1:4222/work") as pull:
    data, hdr = pull.recv(timeout_ms=500)
    print(f"Got {hdr['num_samples']} samples at {hdr['sample_rate'] / 1e6:.2f} MHz")
    pull.ack(data)  # processed: an unacked frame is redelivered and stays queued

Run multiple workers for parallel processing:

# Terminal 1 — sender
python -m doppler.examples.pipeline_send nats://127.0.0.1:4222/work

# Terminals 2 and 3 — two parallel workers
python -m doppler.examples.pipeline_recv nats://127.0.0.1:4222/work 0
python -m doppler.examples.pipeline_recv nats://127.0.0.1:4222/work 1

Python: Requester / Replier

REQ/REP models a remote DSP service: the client sends a signal block, the server processes it and returns the result. The exchange is strictly alternating — send then recv on the Requester, recv then send on the Replier.

from doppler.stream import Requester, Replier, CF64
import numpy as np

ep = "nats://127.0.0.1:4222/ctrl"

# Server side — run in a thread or separate process
with Replier(ep, CF64) as rep:
    request, hdr = rep.recv(timeout_ms=5000)
    result = request * 0.5                    # example: apply -6 dB gain
    rep.send(result, sample_rate=hdr["sample_rate"])

# Client side
x = np.ones(1024, dtype=np.complex128)
with Requester(ep, CF64) as req:
    req.send(x, sample_rate=int(1e6), center_freq=int(2.4e9))
    reply, hdr = req.recv(timeout_ms=2000)
    print(f"Reply: {len(reply)} samples, seq={hdr['sequence']}")

Complete standalone examples:

# Terminal 1 — server (start first)
python -m doppler.examples.replier nats://127.0.0.1:4222/ctrl --gain 0.5

# Terminal 2 — client
python -m doppler.examples.requester nats://127.0.0.1:4222/ctrl --count 20

Network configurations

Every doppler stream endpoint is a NATS client — the transmitter and the receiver both connect to a nats-server broker (nats://host:port/subject); neither one binds a listening socket, so "transmitter" and "receiver" are peers of the broker rather than of each other.

Single machine (localhost)

# Start a broker once, in its own terminal:
nats-server -js

These commands run from a clone of the repository at the release you installed; the make targets and the ./build/ binaries are not in the wheel. Skip this if you already have one:

git clone --branch v0.81.0 https://github.com/doppler-dsp/doppler && cd doppler

// Transmitter and receiver both connect to the local broker
dp_pub_create("nats://127.0.0.1:4222/iq", CF64);
dp_sub_create("nats://127.0.0.1:4222/iq");

# Run the examples
./build/native/examples/transmitter
./build/native/examples/receiver

Two machines over LAN

The broker needs to be reachable from both the transmitter and the receiver — run it on either machine (or a third) and point both clients at its address.

Step 1: Find the broker machine's IP address:

# On the broker machine:
ip addr show | grep inet
# or:
hostname -I

Step 2: Open the firewall port on the broker machine:

# On the broker machine:
sudo ufw allow 4222/tcp
sudo ufw status

Step 3: Start the broker, listening on all interfaces:

# On the broker machine:
nats-server -js -a 0.0.0.0

Step 4: Point both the transmitter and the receiver at the broker's IP:

# On Machine A (transmitter) — replace with the broker's actual IP:
./build/native/examples/transmitter nats://192.168.1.100:4222/iq

# On Machine B (receiver) — same broker IP:
./build/native/examples/receiver nats://192.168.1.100:4222/iq

Loopback (fastest, same machine only)

NATS is TCP-only — there is no unix-domain-socket transport — so the fastest same-machine path is still a loopback connection to a broker on 127.0.0.1:

dp_pub_create("nats://127.0.0.1:4222/iq", CF64);
dp_sub_create("nats://127.0.0.1:4222/iq");

Docker Compose

The repo's docker-compose.yml runs the full pipeline (see the Docker guide). Each service uses the lean stream-services image, which carries the statically-linked streaming binaries on PATH:

services:
  nats:
    image: nats:2.10
    command: "-js"           # no ports: — reached only over doppler-net
    networks: [doppler-net]
  transmitter:
    build: &stream-build
      context: .
      dockerfile: deploy/docker/Dockerfile.examples
      target: stream-services
    image: doppler-stream-services:compose
    depends_on: [nats]
    command: transmitter nats://nats:4222/iq cf64
    networks: [doppler-net]
  receiver-1:
    build: *stream-build
    image: doppler-stream-services:compose   # the one shared image
    depends_on: [transmitter]
    command: receiver nats://nats:4222/iq    # uses Docker DNS
    networks: [doppler-net]
  # receiver-2 (PUB/SUB fan-out) and spectrum-analyzer follow the same shape
networks:
  doppler-net:
    driver: bridge

The stream-services image is built in a glibc 2.28 toolchain image that nothing pulls from a registry, so run make docker-stream before the first docker compose up.

Troubleshooting

Receiver can't connect to the broker

Symptom: Receiver hangs at "Waiting for packets..." when the broker is on a different machine.

Solution:

  1. Verify the receiver is using the broker's IP, not its own:

    # On the receiver machine, check you're connecting to the broker's IP:
./build/native/examples/receiver nats://192.168.1.100:4222/iq
# NOT nats://127.0.0.1:4222/iq (that's the receiver's own machine!)

  2. Check network connectivity:

    # From the receiver machine:
ping 192.168.1.100            # verify basic connectivity
nc -zv 192.168.1.100 4222     # test if the broker's port is reachable
# or:
telnet 192.168.1.100 4222

  3. Verify the firewall on the broker machine:

    # On the broker machine:
sudo ufw status              # check if port 4222 is allowed
sudo ufw allow 4222/tcp      # open it if needed

  4. Check for cloud/network firewalls:

    • AWS security groups, Azure NSGs, GCP firewall rules
    • Router port forwarding if crossing networks

Packets are being dropped

Symptom: Dashboard shows "Dropped: N" packets.

The count is the forward gaps in the header's sequence. A late join and a publisher restart add nothing: both receivers anchor on the first frame they see, and a sequence that goes back is a restart, not a loss.

Causes & solutions:

  • Two publishers on one subject: each numbers its own frames, so their interleaved sequences read as forward gaps

    • Solution: one publisher per subject (a frame carries no sender id yet, #2089)

  • Network congestion: Too much data for the link

    • Solution: Reduce sample rate or packet size in transmitter code

  • Receiver is too slow: Processing can't keep up with arrival rate

    • Solution: Profile your receiver code, optimize processing

"No server available for connection" error

Symptom: Transmitter or receiver fails immediately with a NATS No server available for connection status — there is no nats-server listening at the endpoint's host:port.

Solutions:

  1. Start (or restart) the broker:

    nats-server -js

  2. Verify it's actually listening on the port you expect:

    # Find the process bound to port 4222:
sudo lsof -i :4222
# or:
sudo netstat -tulpn | grep 4222

  3. Use a different port if something else already owns 4222:

    nats-server -js -p 4223
./build/native/examples/transmitter nats://192.168.1.100:4223/iq
./build/native/examples/receiver nats://192.168.1.100:4223/iq

No output / silent failure

It is not a library path: the streaming examples link doppler and the vendored nats.c statically, so there is no libdoppler.so for them to miss and no LD_LIBRARY_PATH to set. Silence means no messages are arriving: check that both ends name the same broker and subject, and that nats-server is running (see Receiver can't connect to the broker).

Performance tips

  • Run the broker on the same machine for the lowest latency: a loopback nats://127.0.0.1:4222/… connection beats a cross-machine one.
  • Batch samples: Larger packets = less overhead (but more latency)
  • Pin threads to cores: For real-time processing (see pthread_setaffinity_np)

Getting help

If you're still stuck:

  1. Check existing issues: https://github.com/doppler-dsp/doppler/issues
  2. Include in your bug report:
    • Output of ./build/native/examples/transmitter --help and ./build/native/examples/receiver --help
    • Network topology (same machine, LAN, cloud, containers)
    • nats-server logs (run it in the foreground with -DV to log connection attempts; -js only enables JetStream)
    • OS and toolchain versions (uname -a, cmake --version) — note doppler statically embeds the vendored nats.c client, so there is no system NATS client library to query