aes67-ESP32-P4/main/hls.c

#include "hls.h"

#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "cJSON.h"
#include "esp_crt_bundle.h"
#include "esp_heap_caps.h"
#include "esp_http_client.h"
#include "esp_log.h"
#include "esp_timer.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"

#define URL_MAX         HLS_URL_MAX
#define PLAYLIST_MAX    (64 * 1024)
#define MAX_SEGMENTS    64
#define LIVE_BACK       3           // start this many segments behind the live edge
#define CHUNK           4096
#define MAX_REDIRECTS   5
#define SEG_MAX         (4 * 1024 * 1024)   // largest segment we accept (PSRAM)

static const char *TAG = "hls";

typedef struct {
    int target_s;
    long long media_seq;
    int count;
    char uri[MAX_SEGMENTS][URL_MAX];
} media_pl_t;

static hls_sink_t s_sink;
static char *s_text;                // playlist buffer (PSRAM)
static media_pl_t *s_pl;            // parsed media playlist (PSRAM)
static char s_url[URL_MAX];         // URL being played
static char s_want_url[URL_MAX];    // from hls_set_url(), guarded by s_url_lock
static portMUX_TYPE s_url_lock = portMUX_INITIALIZER_UNLOCKED;
static char s_media_url[URL_MAX];
static uint8_t *s_seg;              // current segment (PSRAM)
static volatile unsigned s_suspended;   // HLS_SUSPEND_* bits

// Download statistics for the status (hls_status), guarded by s_stat_lock.
static portMUX_TYPE s_stat_lock = portMUX_INITIALIZER_UNLOCKED;
static struct {
    uint32_t seg_bytes, seg_ms;     // last complete segment
    uint32_t errors;                // failed requests since boot
    char error[128];                // last one, with the uptime it happened at
} s_stat;

// Log a failed request and keep it for the status.
static void fail(const char *fmt, ...)
{
    char msg[96];
    va_list ap;
    va_start(ap, fmt);
    vsnprintf(msg, sizeof(msg), fmt, ap);
    va_end(ap);
    ESP_LOGW(TAG, "%s", msg);
    taskENTER_CRITICAL(&s_stat_lock);
    s_stat.errors++;
    snprintf(s_stat.error, sizeof(s_stat.error), "%s (at %lld s)", msg, esp_timer_get_time() / 1000000);
    taskEXIT_CRITICAL(&s_stat_lock);
}

void hls_status(cJSON *st)
{
    taskENTER_CRITICAL(&s_stat_lock);
    typeof(s_stat) t = s_stat;
    taskEXIT_CRITICAL(&s_stat_lock);
    cJSON *o = cJSON_AddObjectToObject(st, "hls");
    cJSON_AddNumberToObject(o, "seg_bytes", t.seg_bytes);
    cJSON_AddNumberToObject(o, "seg_ms", t.seg_ms);
    cJSON_AddNumberToObject(o, "seg_mbit", t.seg_ms ? t.seg_bytes * 8.0 / t.seg_ms / 1000 : 0);
    cJSON_AddNumberToObject(o, "errors", t.errors);
    if (t.error[0]) {
        cJSON_AddStringToObject(o, "error", t.error);
    }
}

void hls_suspend(unsigned reason, bool suspend)
{
    if (suspend) {
        __atomic_or_fetch(&s_suspended, reason, __ATOMIC_SEQ_CST);
    } else {
        __atomic_and_fetch(&s_suspended, ~reason, __ATOMIC_SEQ_CST);
    }
}

void hls_set_url(const char *url)
{
    taskENTER_CRITICAL(&s_url_lock);
    strlcpy(s_want_url, url, sizeof(s_want_url));
    taskEXIT_CRITICAL(&s_url_lock);
}

static void wanted_url(char *out, size_t n)
{
    taskENTER_CRITICAL(&s_url_lock);
    strlcpy(out, s_want_url, n);
    taskEXIT_CRITICAL(&s_url_lock);
}

static bool still_wanted(void)
{
    if (s_suspended) {
        return false;
    }
    char u[URL_MAX];
    wanted_url(u, sizeof(u));
    return strcmp(u, s_url) == 0;
}

/* ----- HTTP ----- */

// Open a GET and follow redirects. Returns the client with headers read, or NULL.
static esp_http_client_handle_t http_open(const char *url)
{
    esp_http_client_config_t cfg = {
        .url = url, .crt_bundle_attach = esp_crt_bundle_attach, .timeout_ms = 10000,
        .buffer_size = CHUNK, .buffer_size_tx = 1024, .user_agent = "p4-aes67",
        .disable_auto_redirect = true, .keep_alive_enable = true,
    };
    esp_http_client_handle_t c = esp_http_client_init(&cfg);
    if (!c) {
        return NULL;
    }
    for (int i = 0; i <= MAX_REDIRECTS; i++) {
        esp_err_t err = esp_http_client_open(c, 0);
        if (err != ESP_OK) {
            fail("open %s: %s", url, esp_err_to_name(err));
            break;
        }
        esp_http_client_fetch_headers(c);
        int st = esp_http_client_get_status_code(c);
        if (st == 301 || st == 302 || st == 303 || st == 307 || st == 308) {
            esp_http_client_flush_response(c, NULL);
            esp_http_client_set_redirection(c);
            esp_http_client_close(c);
            continue;
        }
        if (st == 200) {
            return c;
        }
        fail("GET %s: HTTP %d", url, st);
        break;
    }
    esp_http_client_cleanup(c);
    return NULL;
}

// GET into s_text (NUL-terminated). Returns length or -1.
static int http_get_text(const char *url)
{
    esp_http_client_handle_t c = http_open(url);
    if (!c) {
        return -1;
    }
    int n = 0, r;
    while (n < PLAYLIST_MAX - 1 && (r = esp_http_client_read(c, s_text + n, PLAYLIST_MAX - 1 - n)) > 0) {
        n += r;
    }
    s_text[n] = 0;
    esp_http_client_close(c);
    esp_http_client_cleanup(c);
    return n;
}

/* ----- playlists ----- */

// Absolute, host-relative ("/x") or path-relative ("x") reference against base.
static void resolve(char *out, size_t n, const char *base, const char *ref)
{
    if (strstr(ref, "://")) {
        strlcpy(out, ref, n);
        return;
    }
    char b[URL_MAX];
    strlcpy(b, base, sizeof(b));
    char *q = strchr(b, '?');
    if (q) {
        *q = 0;
    }
    if (ref[0] == '/') {
        char *host_end = strchr(strstr(b, "://") + 3, '/');
        if (host_end) {
            *host_end = 0;
        }
    } else {
        char *slash = strrchr(b, '/');
        if (slash) {
            slash[1] = 0;
        }
    }
    snprintf(out, n, "%s%s", b, ref);
}

// Next line of s_text (CRLF tolerant); NULL at the end.
static char *next_line(char **p)
{
    if (!*p || !**p) {
        return NULL;
    }
    char *line = *p, *e = strchr(line, '\n');
    *p = e ? e + 1 : NULL;
    if (e) {
        *e = 0;
    }
    size_t l = strlen(line);
    while (l && (line[l - 1] == '\r' || line[l - 1] == ' ')) {
        line[--l] = 0;
    }
    return line;
}

// Master playlist: pick the variant with the highest BANDWIDTH. false if s_text is a media playlist.
static bool pick_variant(const char *base, char *out, size_t n)
{
    if (!strstr(s_text, "#EXT-X-STREAM-INF")) {
        return false;
    }
    long best = -1, bw = -1;
    char *p = s_text, *line;
    while ((line = next_line(&p))) {
        if (!strncmp(line, "#EXT-X-STREAM-INF:", 18)) {
            const char *b = strstr(line, "BANDWIDTH=");
            bw = b ? atol(b + 10) : 0;
        } else if (line[0] && line[0] != '#' && bw >= 0) {
            if (bw > best) {
                best = bw;
                resolve(out, n, base, line);
            }
            bw = -1;
        }
    }
    ESP_LOGI(TAG, "variant %ld bit/s: %s", best, out);
    return best >= 0;
}

static bool parse_media(const char *base)
{
    s_pl->target_s = 10;
    s_pl->media_seq = 0;
    s_pl->count = 0;
    char *p = s_text, *line;
    bool is_m3u = false;
    while ((line = next_line(&p))) {
        if (!strcmp(line, "#EXTM3U")) {
            is_m3u = true;
        } else if (!strncmp(line, "#EXT-X-TARGETDURATION:", 22)) {
            s_pl->target_s = atoi(line + 22);
        } else if (!strncmp(line, "#EXT-X-MEDIA-SEQUENCE:", 22)) {
            s_pl->media_seq = atoll(line + 22);
        } else if (line[0] && line[0] != '#' && s_pl->count < MAX_SEGMENTS) {
            resolve(s_pl->uri[s_pl->count++], URL_MAX, base, line);
        }
    }
    return is_m3u && s_pl->count > 0;
}

/* ----- segments ----- */

static bool fetch_segment(long long seq, const char *url)
{
    int64_t t0 = esp_timer_get_time();
    esp_http_client_handle_t c = http_open(url);
    if (!c) {
        return false;
    }
    // Download completely first, then decode: the connection is not held open while the
    // decoder waits for space in the ring (a stalled TCP stream for ~10 s may get cut by the CDN).
    size_t total = 0;
    int r = 0;
    bool ok = true;
    while (total < SEG_MAX && (r = esp_http_client_read(c, (char *)s_seg + total, SEG_MAX - total)) > 0) {
        total += r;
    }
    if (r < 0 || total == SEG_MAX) {
        ok = false;
        fail("segment %lld: %s after %u bytes", seq, r < 0 ? "read error" : "too large", (unsigned)total);
    }
    esp_http_client_close(c);
    esp_http_client_cleanup(c);
    int64_t us = esp_timer_get_time() - t0;
    if (ok) {
        taskENTER_CRITICAL(&s_stat_lock);
        s_stat.seg_bytes = total;
        s_stat.seg_ms = us / 1000;
        taskEXIT_CRITICAL(&s_stat_lock);
    }
    ESP_LOGD(TAG, "segment %lld: %u bytes in %.2f s (%.1f Mbit/s)%s", seq, (unsigned)total, us / 1e6,
             us ? total * 8.0 / us : 0.0, ok ? "" : ", failed");
    if (ok && s_sink) {
        ok = s_sink(s_seg, total, true);   // blocks at playback speed while the ring is full
    }
    return ok;
}

static void hls_task(void *arg)
{
    while (1) {
        wanted_url(s_url, sizeof(s_url));
        if (!s_url[0] || s_suspended) {
            vTaskDelay(pdMS_TO_TICKS(1000));
            continue;
        }
        ESP_LOGI(TAG, "start: %s", s_url);
        // Master -> media playlist
        if (http_get_text(s_url) < 0) {
            vTaskDelay(pdMS_TO_TICKS(5000));
            continue;
        }
        if (!pick_variant(s_url, s_media_url, sizeof(s_media_url))) {
            strlcpy(s_media_url, s_url, sizeof(s_media_url));
        }
        long long next = -1;
        int errors = 0;
        while (still_wanted() && errors < 5) {
            int64_t t0 = esp_timer_get_time();
            if (http_get_text(s_media_url) < 0 || !parse_media(s_media_url)) {
                errors++;
                vTaskDelay(pdMS_TO_TICKS(2000));
                continue;
            }
            long long first = s_pl->media_seq, last = first + s_pl->count - 1;
            if (next < 0) {
                next = last - LIVE_BACK + 1 > first ? last - LIVE_BACK + 1 : first;
                ESP_LOGI(TAG, "live playlist: seq %lld..%lld, target %d s, starting at %lld",
                         first, last, s_pl->target_s, next);
            } else if (next < first) {
                ESP_LOGW(TAG, "fell behind the live window (%lld < %lld), skipping ahead", next, first);
                next = first;
            }
            bool got_new = false;
            for (; next <= last && still_wanted(); next++) {
                if (!fetch_segment(next, s_pl->uri[next - first])) {
                    errors++;
                    break;
                }
                errors = 0;
                got_new = true;
            }
            // No new segment: wait half the target duration before reloading (RFC 8216 6.3.4).
            if (!got_new) {
                int64_t wait = s_pl->target_s * 500000LL - (esp_timer_get_time() - t0);
                if (wait > 0) {
                    vTaskDelay(pdMS_TO_TICKS(wait / 1000));
                }
            }
        }
        if (errors >= 5) {
            ESP_LOGW(TAG, "too many errors, restarting from the playlist in 5 s");
            vTaskDelay(pdMS_TO_TICKS(5000));
        }
    }
}

esp_err_t hls_start(hls_sink_t sink)
{
    s_sink = sink;
    s_text = heap_caps_malloc(PLAYLIST_MAX, MALLOC_CAP_SPIRAM);
    s_pl = heap_caps_malloc(sizeof(media_pl_t), MALLOC_CAP_SPIRAM);
    s_seg = heap_caps_malloc(SEG_MAX, MALLOC_CAP_SPIRAM);
    if (!s_text || !s_pl || !s_seg) {
        return ESP_ERR_NO_MEM;
    }
    return xTaskCreate(hls_task, "hls", 10240, NULL, 4, NULL) == pdPASS ? ESP_OK : ESP_ERR_NO_MEM;
}