stream_broadcaster.h

Go to the documentation of this file.

 
#pragma once
 
#include <array>
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <optional>
 
#include <esp_http_server.h>
#include <freertos/FreeRTOS.h>
#include <freertos/semphr.h>
#include <freertos/task.h>
 
#include "frame_slots.h"
#include "frame_store.h"
#include "viewer_set.h"
 
class StreamBroadcaster
{
public:
    bool start(httpd_handle_t server, FrameStore& store);
 
    void stop();
 
    static void on_session_closed(httpd_handle_t server, int sockfd);
 
    uint32_t frames_sent() const { return sent_.load(); }
 
    struct SendStats {
        uint64_t total_us = 0; 
        uint32_t slow = 0;     
        uint32_t failed = 0;   
        uint32_t max_us = 0;   
    };
    static constexpr uint32_t kSlowSendUs = 200000; 
 
    SendStats take_send_stats();
 
    size_t viewer_count() const;
 
private:
    struct SendJob {
        StreamBroadcaster* owner = nullptr;
        int fd = -1;
        size_t index = 0;        
        uint32_t generation = 0; 
        bool in_flight = false;
        std::optional<frames::FrameLease> lease;
    };
 
    static esp_err_t ws_handler(httpd_req_t* req);
    static esp_err_t pre_handshake(httpd_req_t* req);  // ws_pre_handshake_cb: origin check, before the upgrade
    static esp_err_t post_handshake(httpd_req_t* req); // registers the viewer once the session is a WebSocket
    static void stream_task(void* arg);
    static void send_job(void* arg);
    void run();
    void remove_viewer(int fd);
    void wake_task(); // notify the stream task that a send finished
 
    httpd_handle_t server_ = nullptr;
    FrameStore* store_ = nullptr;
    frames::ViewerSet viewers_;
    std::array<SendJob, frames::kMaxViewers> jobs_{};
    std::array<uint32_t, frames::kMaxViewers> generation_{}; // bumped per new viewer in a slot; guards viewers_
    TaskHandle_t task_ = nullptr;
    SemaphoreHandle_t mutex_ = nullptr; // guards viewers_, generation_ and jobs_[i].in_flight
    std::atomic<uint32_t> sent_{0};
    // Written only by send_job on the httpd task; read by take_send_stats() on loop().
    std::atomic<uint64_t> send_us_total_{0};
    std::atomic<uint32_t> send_slow_{0};
    std::atomic<uint32_t> send_failed_{0};
    std::atomic<uint32_t> send_us_max_{0};
 
    static StreamBroadcaster* instance_; // close_fn has no user context
};