zebra-report/test/functional.c

/*
 * functional.c — end-to-end TX/RX pipeline and auto-negotiate tests
 *
 * Requires two live PulseAudio sink-inputs:
 *   ZEBRA_DATA_SINK  — TX modulates this, RX reads it  (data channel)
 *   ZEBRA_CTRL_SINK  — RX modulates this, TX reads it  (handshake channel)
 *
 * Run ./tx -l to find sink indices, then:
 *   ZEBRA_DATA_SINK=15815 ZEBRA_CTRL_SINK=15923 ./test/functional
 *
 * Compile: gcc -Wall -O2 -Iinclude $(pkg-config --cflags libpulse) \
 *               -o test/functional test/functional.c src/pulse.c \
 *               $(pkg-config --libs libpulse) -lrt -lpthread
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
#include <time.h>
#include "zebra.h"
#include "modem.h"
#include "test.h"

static uint32_t data_sink = 0;
static uint32_t ctrl_sink = 0;

/* ------------------------------------------------------------------ *
 * shared RX byte collector                                             *
 * ------------------------------------------------------------------ */

#define COLLECT_MAX 256

typedef struct {
    char       buf[COLLECT_MAX];
    int        len;
    int        want;      /* stop after this many bytes */
    sem_t      done;
} collector_t;

static void collect_byte(uint8_t b, void *ud) {
    collector_t *c = ud;
    if (c->len < COLLECT_MAX - 1) c->buf[c->len++] = (char)b;
    if (c->len >= c->want) sem_post(&c->done);
}

/* ------------------------------------------------------------------ *
 * test 1: fixed-baud loopback                                          *
 * TX and RX both run against the same data_sink at 50 baud.           *
 * ------------------------------------------------------------------ */

typedef struct {
    zebra_pulse_t *z;
    uint32_t       sink;
    uint8_t        channels;
    int            baud;
    const char    *msg;
} tx_arg_t;

typedef struct {
    zebra_pulse_t *z;
    uint32_t       sink;
    int            baud;
    collector_t   *col;
} rx_arg_t;

static void *tx_thread(void *arg) {
    tx_arg_t *a = arg;
    /* brief delay so RX loop is running before TX fires */
    struct timespec d = {0, 200000000L};   /* 200ms */
    nanosleep(&d, NULL);

    long period_ns = 1000000000L / a->baud;
    zebra_set_volume_fast(a->z, a->sink, a->channels, ZEBRA_VOL_MARK);

    struct timespec next;
    clock_gettime(CLOCK_MONOTONIC, &next);
    ts_add_ns(&next, period_ns);

    for (int i = 0; a->msg[i]; i++)
        zebra_send_byte(a->z, a->sink, a->channels, (uint8_t)a->msg[i], &next, period_ns);

    zebra_set_volume_fast(a->z, a->sink, a->channels, ZEBRA_VOL_MARK);
    return NULL;
}

static void *rx_thread(void *arg) {
    rx_arg_t *a = arg;
    zebra_rx_run(a->z, a->sink, a->baud, collect_byte, a->col);
    return NULL;
}

static void t_loopback_fixed_baud(void) {
    const char *msg  = "ZEBRA\n";
    const int   baud = 50;
    const int   timeout_s = 10;

    collector_t col;
    memset(&col, 0, sizeof(col));
    col.want = (int)strlen(msg);
    sem_init(&col.done, 0, 0);

    zebra_pulse_t ztx, zrx;
    T_EQ(zebra_pulse_connect(&ztx, "zebra-test-tx"), 0);
    T_EQ(zebra_pulse_connect(&zrx, "zebra-test-rx"), 0);

    /* resolve channel count for TX */
    uint8_t ch = 2;
    {
        zebra_sink_t buf[64];
        int n = zebra_list_sinks(&ztx, buf, 64);
        for (int i = 0; i < n; i++)
            if (buf[i].index == data_sink)
                { ch = buf[i].channels ? buf[i].channels : 2; break; }
    }

    tx_arg_t txa = { .z = &ztx, .sink = data_sink, .channels = ch,
                     .baud = baud, .msg = msg };
    rx_arg_t rxa = { .z = &zrx, .sink = data_sink, .baud = baud, .col = &col };

    pthread_t tx_tid, rx_tid;
    pthread_create(&rx_tid, NULL, rx_thread, &rxa);
    pthread_create(&tx_tid, NULL, tx_thread, &txa);

    /* wait for collector or timeout */
    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec += timeout_s;
    int timed_out = sem_timedwait(&col.done, &deadline);

    pthread_cancel(rx_tid);
    pthread_detach(rx_tid);
    pthread_join(tx_tid, NULL);

    T_EQ(timed_out, 0);   /* did not time out */
    col.buf[col.len] = '\0';
    T_EQ(memcmp(col.buf, msg, strlen(msg)), 0);
    fprintf(stderr, "  [rx] received %d bytes: %.*s", col.len, col.len, col.buf);

    zebra_pulse_disconnect(&ztx);
    zebra_pulse_disconnect(&zrx);
    sem_destroy(&col.done);
}

/* ------------------------------------------------------------------ *
 * test 2: auto-negotiate handshake                                     *
 * RX benchmarks, sends offer on ctrl_sink.                            *
 * TX receives offer, then sends data on data_sink at negotiated baud. *
 * RX receives data at negotiated baud.                                 *
 * ------------------------------------------------------------------ */

typedef struct {
    zebra_pulse_t *z_data;    /* for TX data */
    zebra_pulse_t *z_ctrl;    /* for reading handshake */
    uint32_t       data_sink;
    uint32_t       ctrl_sink;
    uint8_t        data_ch;
    const char    *msg;
    int            timeout_ms;
} autoneg_tx_arg_t;

typedef struct {
    zebra_pulse_t *z_data;    /* for RX data */
    zebra_pulse_t *z_ctrl;    /* for sending handshake */
    uint32_t       data_sink;
    uint32_t       ctrl_sink;
    uint8_t        ctrl_ch;
    collector_t   *col;
} autoneg_rx_arg_t;

static void *autoneg_tx_thread(void *arg) {
    autoneg_tx_arg_t *a = arg;
    uint16_t neg_baud = 0;
    fprintf(stderr, "  [tx] waiting for handshake on ctrl_sink %u...\n", a->ctrl_sink);
    if (zebra_recv_handshake(a->z_ctrl, a->ctrl_sink, a->timeout_ms, &neg_baud) < 0) {
        fprintf(stderr, "  [tx] handshake timeout — no frame received\n");
        return NULL;
    }
    fprintf(stderr, "  [tx] OFFER received: %u baud — waiting for READY\n", neg_baud);
    if (zebra_recv_ready(a->z_ctrl, a->ctrl_sink, ZEBRA_HS_READY_WAIT) < 0) {
        fprintf(stderr, "  [tx] READY timeout — aborting\n");
        return NULL;
    }
    fprintf(stderr, "  [tx] READY received — sending '%s' at %u baud on data_sink %u\n",
            a->msg, neg_baud, a->data_sink);
    long period_ns = 1000000000L / neg_baud;
    zebra_set_volume_fast(a->z_data, a->data_sink, a->data_ch, ZEBRA_VOL_MARK);

    struct timespec next;
    clock_gettime(CLOCK_MONOTONIC, &next);
    ts_add_ns(&next, period_ns);

    for (int i = 0; a->msg[i]; i++)
        zebra_send_byte(a->z_data, a->data_sink, a->data_ch,
                        (uint8_t)a->msg[i], &next, period_ns);

    zebra_set_volume_fast(a->z_data, a->data_sink, a->data_ch, ZEBRA_VOL_MARK);
    fprintf(stderr, "  [tx] send complete\n");
    return NULL;
}

static void *autoneg_rx_thread(void *arg) {
    autoneg_rx_arg_t *a = arg;
    int baud = zebra_benchmark_baud(a->z_data, a->data_sink);
    fprintf(stderr, "  [rx] benchmark: %d baud\n", baud);
    fprintf(stderr, "  [rx] sending handshake on ctrl_sink %u ch=%u\n",
            a->ctrl_sink, a->ctrl_ch);
    zebra_send_handshake(a->z_ctrl, a->ctrl_sink, a->ctrl_ch, (uint16_t)baud);
    fprintf(stderr, "  [rx] OFFER sent — sending READY (3x) then entering receive loop\n");
    zebra_send_ready(a->z_ctrl, a->ctrl_sink, a->ctrl_ch);

    fprintf(stderr, "  [rx] entering receive loop at %d baud on data_sink %u\n",
            baud, a->data_sink);
    zebra_rx_run(a->z_data, a->data_sink, baud, collect_byte, a->col);
    return NULL;
}

static void t_autoneg_pipeline(void) {
    const char *msg = "PING\n";
    const int timeout_s  = 30;
    const int timeout_ms = 15000;   /* handshake timeout for TX */

    collector_t col;
    memset(&col, 0, sizeof(col));
    col.want = (int)strlen(msg);
    sem_init(&col.done, 0, 0);

    /* four connections: TX-data, TX-ctrl, RX-data, RX-ctrl */
    zebra_pulse_t z_tx_data, z_tx_ctrl, z_rx_data, z_rx_ctrl;
    T_EQ(zebra_pulse_connect(&z_tx_data, "zebra-test-tx-data"), 0);
    T_EQ(zebra_pulse_connect(&z_tx_ctrl, "zebra-test-tx-ctrl"), 0);
    T_EQ(zebra_pulse_connect(&z_rx_data, "zebra-test-rx-data"), 0);
    T_EQ(zebra_pulse_connect(&z_rx_ctrl, "zebra-test-rx-ctrl"), 0);

    /* resolve channel counts */
    uint8_t data_ch = 2, ctrl_ch = 2;
    {
        zebra_sink_t buf[64];
        int n = zebra_list_sinks(&z_tx_data, buf, 64);
        for (int i = 0; i < n; i++) {
            if (buf[i].index == data_sink) data_ch = buf[i].channels ? buf[i].channels : 2;
            if (buf[i].index == ctrl_sink) ctrl_ch = buf[i].channels ? buf[i].channels : 2;
        }
    }

    autoneg_tx_arg_t txa = {
        .z_data = &z_tx_data, .z_ctrl = &z_tx_ctrl,
        .data_sink = data_sink, .ctrl_sink = ctrl_sink,
        .data_ch = data_ch, .msg = msg, .timeout_ms = timeout_ms
    };
    autoneg_rx_arg_t rxa = {
        .z_data = &z_rx_data, .z_ctrl = &z_rx_ctrl,
        .data_sink = data_sink, .ctrl_sink = ctrl_sink,
        .ctrl_ch = ctrl_ch, .col = &col
    };

    pthread_t tx_tid, rx_tid;
    pthread_create(&tx_tid, NULL, autoneg_tx_thread, &txa);
    pthread_create(&rx_tid, NULL, autoneg_rx_thread, &rxa);

    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec += timeout_s;
    int timed_out = sem_timedwait(&col.done, &deadline);

    pthread_cancel(rx_tid);
    pthread_detach(rx_tid);   /* don't join — PA wait may not respond to cancel */
    pthread_join(tx_tid, NULL);

    T_EQ(timed_out, 0);
    col.buf[col.len] = '\0';
    T_EQ(memcmp(col.buf, msg, strlen(msg)), 0);
    fprintf(stderr, "  [rx] received %d bytes: %.*s", col.len, col.len, col.buf);

    zebra_pulse_disconnect(&z_tx_data);
    zebra_pulse_disconnect(&z_tx_ctrl);
    zebra_pulse_disconnect(&z_rx_data);
    zebra_pulse_disconnect(&z_rx_ctrl);
    sem_destroy(&col.done);
}

/* ------------------------------------------------------------------ *
 * main                                                                 *
 * ------------------------------------------------------------------ */

int main(void) {
    const char *d = getenv("ZEBRA_DATA_SINK");
    const char *c = getenv("ZEBRA_CTRL_SINK");

    if (!d || !c) {
        fprintf(stderr,
            "ZEBRA_DATA_SINK and ZEBRA_CTRL_SINK must be set.\n"
            "  run: ./tx -l  to list available sink indices\n"
            "  example: ZEBRA_DATA_SINK=15815 ZEBRA_CTRL_SINK=15923 ./test/functional\n"
            "\n"
            "  DATA_SINK: TX modulates it, RX reads it  (e.g. 'zebra report' Firefox tab)\n"
            "  CTRL_SINK: RX modulates it, TX reads it  (e.g. 'X' Firefox tab)\n");
        return 1;
    }

    data_sink = (uint32_t)atoi(d);
    ctrl_sink = (uint32_t)atoi(c);

    if (data_sink == ctrl_sink) {
        fprintf(stderr, "error: DATA_SINK and CTRL_SINK must be different\n");
        return 1;
    }

    fprintf(stderr, "data_sink=%u  ctrl_sink=%u\n\n", data_sink, ctrl_sink);

    T_SECTION("loopback: fixed baud (50 baud)");
    T_RUN(t_loopback_fixed_baud);

    T_SECTION("auto-negotiate pipeline");
    T_RUN(t_autoneg_pipeline);

    T_SUMMARY();
}