type = AVIO_DATA_MARKER_SYNC_POINT;
}
+ HTTPD::StreamID stream_id{ HTTPD::MAIN_STREAM, 0 };
if (type == AVIO_DATA_MARKER_HEADER) {
stream_mux_header.append((char *)buf, buf_size);
- global_httpd->set_header(HTTPD::MAIN_STREAM, stream_mux_header);
+ global_httpd->set_header(stream_id, stream_mux_header);
} else {
- global_httpd->add_data(HTTPD::MAIN_STREAM, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
+ global_httpd->add_data(stream_id, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
}
return buf_size;
}
#define MAX_FPS 60
#define FAKE_FPS 25 // Must be an integer.
-#define MAX_VIDEO_CARDS 16
+// #define MAX_VIDEO_CARDS 16 // defined in shared_defs.h.
#define MAX_ALSA_CARDS 16
#define MAX_BUSES 256 // Audio buses.
type = AVIO_DATA_MARKER_SYNC_POINT;
}
+ HTTPD::StreamID stream_id{ HTTPD::MAIN_STREAM, 0 };
if (type == AVIO_DATA_MARKER_HEADER) {
stream_mux_header.append((char *)buf, buf_size);
- httpd->set_header(HTTPD::MAIN_STREAM, stream_mux_header);
+ httpd->set_header(stream_id, stream_mux_header);
} else {
- httpd->add_data(HTTPD::MAIN_STREAM, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
+ httpd->add_data(stream_id, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
}
return buf_size;
}
//
// Only bother doing MJPEG encoding if there are any connected clients
// that want the stream.
- if (httpd.get_num_connected_multicam_clients() > 0) {
+ if (httpd.get_num_connected_multicam_clients() > 0 ||
+ httpd.get_num_connected_siphon_clients(card_index) > 0) {
vector<int32_t> converted_samples = convert_audio_to_fixed32(audio_frame.data + audio_offset, num_samples, audio_format, 2);
lock_guard<mutex> lock(card_mutex);
if (card->new_raw_audio.empty()) {
#include "mjpeg_encoder.h"
+#include <assert.h>
#include <jpeglib.h>
#include <unistd.h>
#if __SSE2__
int MJPEGEncoder::write_packet2_thunk(void *opaque, uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time)
{
- MJPEGEncoder *engine = (MJPEGEncoder *)opaque;
- return engine->write_packet2(buf, buf_size, type, time);
+ WritePacket2Context *ctx = (WritePacket2Context *)opaque;
+ return ctx->mjpeg_encoder->write_packet2(ctx->stream_id, buf, buf_size, type, time);
}
-int MJPEGEncoder::write_packet2(uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time)
+int MJPEGEncoder::write_packet2(HTTPD::StreamID stream_id, uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time)
{
+ string *mux_header = &streams[stream_id].mux_header;
if (type == AVIO_DATA_MARKER_HEADER) {
- mux_header.append((char *)buf, buf_size);
- httpd->set_header(HTTPD::MULTICAM_STREAM, mux_header);
+ mux_header->append((char *)buf, buf_size);
+ httpd->set_header(stream_id, *mux_header);
} else {
- httpd->add_data(HTTPD::MULTICAM_STREAM, (char *)buf, buf_size, /*keyframe=*/true, AV_NOPTS_VALUE, AVRational{ AV_TIME_BASE, 1 });
+ httpd->add_data(stream_id, (char *)buf, buf_size, /*keyframe=*/true, AV_NOPTS_VALUE, AVRational{ AV_TIME_BASE, 1 });
}
return buf_size;
}
MJPEGEncoder::MJPEGEncoder(HTTPD *httpd, const string &va_display)
: httpd(httpd)
{
- // Set up the mux. We don't use the Mux wrapper, because it's geared towards
- // a situation with only one video stream (and possibly one audio stream)
- // with known width/height, and we don't need the extra functionality it provides.
- avctx.reset(avformat_alloc_context());
- avctx->oformat = av_guess_format("nut", nullptr, nullptr);
-
- uint8_t *buf = (uint8_t *)av_malloc(MUX_BUFFER_SIZE);
- avctx->pb = avio_alloc_context(buf, MUX_BUFFER_SIZE, 1, this, nullptr, nullptr, nullptr);
- avctx->pb->write_data_type = &MJPEGEncoder::write_packet2_thunk;
- avctx->flags = AVFMT_FLAG_CUSTOM_IO;
-
- for (unsigned card_idx = 0; card_idx < global_flags.card_to_mjpeg_stream_export.size(); ++card_idx) {
- add_video_stream(avctx.get());
+ create_ffmpeg_context(HTTPD::StreamID{ HTTPD::MULTICAM_STREAM, 0 });
+ for (unsigned stream_idx = 0; stream_idx < MAX_VIDEO_CARDS; ++stream_idx) {
+ create_ffmpeg_context(HTTPD::StreamID{ HTTPD::SIPHON_STREAM, stream_idx });
}
- for (unsigned card_idx = 0; card_idx < global_flags.card_to_mjpeg_stream_export.size(); ++card_idx) {
- add_audio_stream(avctx.get());
- }
- finalize_mux(avctx.get());
+
+ add_stream(HTTPD::StreamID{ HTTPD::MULTICAM_STREAM, 0 });
// Initialize VA-API.
string error;
MJPEGEncoder::~MJPEGEncoder()
{
- av_free(avctx->pb->buffer);
+ for (auto &id_and_stream : streams) {
+ av_free(id_and_stream.second.avctx->pb->buffer);
+ }
global_metrics.remove("mjpeg_frames", {{ "status", "dropped" }, { "reason", "zero_size" }});
global_metrics.remove("mjpeg_frames", {{ "status", "dropped" }, { "reason", "interlaced" }});
{
// Only bother doing MJPEG encoding if there are any connected clients
// that want the stream.
- if (httpd->get_num_connected_multicam_clients() == 0) {
+ if (httpd->get_num_connected_multicam_clients() == 0 &&
+ httpd->get_num_connected_siphon_clients(card_index) == 0) {
return false;
}
// Will call back in the receiver thread.
encode_jpeg_va(move(qf));
} else {
+ update_siphon_streams();
+
+ HTTPD::StreamID multicam_id{ HTTPD::MULTICAM_STREAM, 0 };
+ HTTPD::StreamID siphon_id{ HTTPD::SIPHON_STREAM, qf.card_index };
+ assert(streams.count(multicam_id));
+
// Write audio before video, since Futatabi expects it.
if (qf.audio.size() > 0) {
- write_audio_packet(qf.pts, stream_index, qf.audio);
+ write_audio_packet(streams[multicam_id].avctx.get(), qf.pts, stream_index + global_flags.card_to_mjpeg_stream_export.size(), qf.audio);
+ if (streams.count(siphon_id)) {
+ write_audio_packet(streams[siphon_id].avctx.get(), qf.pts, /*stream_index=*/1, qf.audio);
+ }
}
// Encode synchronously, in the same thread.
vector<uint8_t> jpeg = encode_jpeg_libjpeg(qf);
- write_mjpeg_packet(qf.pts, stream_index, jpeg.data(), jpeg.size());
+ write_mjpeg_packet(streams[multicam_id].avctx.get(), qf.pts, stream_index, jpeg.data(), jpeg.size());
+ if (streams.count(siphon_id)) {
+ write_mjpeg_packet(streams[siphon_id].avctx.get(), qf.pts, /*stream_index=*/0, jpeg.data(), jpeg.size());
+ }
}
}
free(tmp_cr);
}
-void MJPEGEncoder::write_mjpeg_packet(int64_t pts, unsigned card_index, const uint8_t *jpeg, size_t jpeg_size)
+void MJPEGEncoder::write_mjpeg_packet(AVFormatContext *avctx, int64_t pts, unsigned stream_index, const uint8_t *jpeg, size_t jpeg_size)
{
AVPacket pkt;
memset(&pkt, 0, sizeof(pkt));
pkt.buf = nullptr;
pkt.data = const_cast<uint8_t *>(jpeg);
pkt.size = jpeg_size;
- pkt.stream_index = card_index;
+ pkt.stream_index = stream_index;
pkt.flags = AV_PKT_FLAG_KEY;
AVRational time_base = avctx->streams[pkt.stream_index]->time_base;
pkt.pts = pkt.dts = av_rescale_q(pts, AVRational{ 1, TIMEBASE }, time_base);
pkt.duration = 0;
- if (av_write_frame(avctx.get(), &pkt) < 0) {
+ if (av_write_frame(avctx, &pkt) < 0) {
fprintf(stderr, "av_write_frame() failed\n");
abort();
}
}
-void MJPEGEncoder::write_audio_packet(int64_t pts, unsigned card_index, const vector<int32_t> &audio)
+void MJPEGEncoder::write_audio_packet(AVFormatContext *avctx, int64_t pts, unsigned stream_index, const vector<int32_t> &audio)
{
AVPacket pkt;
memset(&pkt, 0, sizeof(pkt));
pkt.buf = nullptr;
pkt.data = reinterpret_cast<uint8_t *>(const_cast<int32_t *>(&audio[0]));
pkt.size = audio.size() * sizeof(audio[0]);
- pkt.stream_index = card_index + global_flags.card_to_mjpeg_stream_export.size();
+ pkt.stream_index = stream_index;
pkt.flags = AV_PKT_FLAG_KEY;
AVRational time_base = avctx->streams[pkt.stream_index]->time_base;
pkt.pts = pkt.dts = av_rescale_q(pts, AVRational{ 1, TIMEBASE }, time_base);
size_t num_stereo_samples = audio.size() / 2;
pkt.duration = av_rescale_q(num_stereo_samples, AVRational{ 1, OUTPUT_FREQUENCY }, time_base);
- if (av_write_frame(avctx.get(), &pkt) < 0) {
+ if (av_write_frame(avctx, &pkt) < 0) {
fprintf(stderr, "av_write_frame() failed\n");
abort();
}
frames_encoding.pop();
}
+ update_siphon_streams();
+
assert(global_flags.card_to_mjpeg_stream_export.count(qf.card_index)); // Or should_encode_mjpeg_for_card() would have returned false.
int stream_index = global_flags.card_to_mjpeg_stream_export[qf.card_index];
+ HTTPD::StreamID multicam_id{ HTTPD::MULTICAM_STREAM, 0 };
+ HTTPD::StreamID siphon_id{ HTTPD::SIPHON_STREAM, qf.card_index };
+ assert(streams.count(multicam_id));
+ assert(streams[multicam_id].avctx != nullptr);
+
// Write audio before video, since Futatabi expects it.
if (qf.audio.size() > 0) {
- write_audio_packet(qf.pts, stream_index, qf.audio);
+ write_audio_packet(streams[multicam_id].avctx.get(), qf.pts, stream_index + global_flags.card_to_mjpeg_stream_export.size(), qf.audio);
+ if (streams.count(siphon_id)) {
+ write_audio_packet(streams[siphon_id].avctx.get(), qf.pts, /*stream_index=*/1, qf.audio);
+ }
}
VAStatus va_status = vaSyncSurface(va_dpy->va_dpy, qf.resources.surface);
CHECK_VASTATUS(va_status, "vaMapBuffer");
const uint8_t *coded_buf = reinterpret_cast<uint8_t *>(segment->buf);
- write_mjpeg_packet(qf.pts, stream_index, coded_buf, segment->size);
+ write_mjpeg_packet(streams[multicam_id].avctx.get(), qf.pts, stream_index, coded_buf, segment->size);
+ if (streams.count(siphon_id)) {
+ write_mjpeg_packet(streams[siphon_id].avctx.get(), qf.pts, /*stream_index=*/0, coded_buf, segment->size);
+ }
va_status = vaUnmapBuffer(va_dpy->va_dpy, qf.resources.data_buffer);
CHECK_VASTATUS(va_status, "vaUnmapBuffer");
return dest.dest;
}
+
+void MJPEGEncoder::add_stream(HTTPD::StreamID stream_id)
+{
+ AVFormatContextWithCloser avctx;
+
+ // Set up the mux. We don't use the Mux wrapper, because it's geared towards
+ // a situation with only one video stream (and possibly one audio stream)
+ // with known width/height, and we don't need the extra functionality it provides.
+ avctx.reset(avformat_alloc_context());
+ avctx->oformat = av_guess_format("nut", nullptr, nullptr);
+
+ uint8_t *buf = (uint8_t *)av_malloc(MUX_BUFFER_SIZE);
+ avctx->pb = avio_alloc_context(buf, MUX_BUFFER_SIZE, 1, &ffmpeg_contexts[stream_id], nullptr, nullptr, nullptr);
+ avctx->pb->write_data_type = &MJPEGEncoder::write_packet2_thunk;
+ avctx->flags = AVFMT_FLAG_CUSTOM_IO;
+
+ if (stream_id.type == HTTPD::MULTICAM_STREAM) {
+ for (unsigned card_idx = 0; card_idx < global_flags.card_to_mjpeg_stream_export.size(); ++card_idx) {
+ add_video_stream(avctx.get());
+ }
+ for (unsigned card_idx = 0; card_idx < global_flags.card_to_mjpeg_stream_export.size(); ++card_idx) {
+ add_audio_stream(avctx.get());
+ }
+ } else {
+ assert(stream_id.type == HTTPD::SIPHON_STREAM);
+ add_video_stream(avctx.get());
+ add_audio_stream(avctx.get());
+ }
+ finalize_mux(avctx.get());
+
+ Stream s;
+ s.avctx = move(avctx);
+ streams[stream_id] = move(s);
+}
+
+void MJPEGEncoder::update_siphon_streams()
+{
+ // Bring the list of streams into sync with what the clients need.
+ for (auto it = streams.begin(); it != streams.end(); ) {
+ if (it->first.type != HTTPD::SIPHON_STREAM) {
+ ++it;
+ continue;
+ }
+ if (httpd->get_num_connected_siphon_clients(it->first.index) == 0) {
+ av_free(it->second.avctx->pb->buffer);
+ streams.erase(it++);
+ } else {
+ ++it;
+ }
+ }
+ for (unsigned stream_idx = 0; stream_idx < MAX_VIDEO_CARDS; ++stream_idx) {
+ HTTPD::StreamID stream_id{ HTTPD::SIPHON_STREAM, stream_idx };
+ if (streams.count(stream_id) == 0 && httpd->get_num_connected_siphon_clients(stream_idx) > 0) {
+ add_stream(stream_id);
+ }
+ }
+}
+
+void MJPEGEncoder::create_ffmpeg_context(HTTPD::StreamID stream_id)
+{
+ ffmpeg_contexts.emplace(stream_id, WritePacket2Context{ this, stream_id });
+}
#ifndef _MJPEG_ENCODER_H
#define _MJPEG_ENCODER_H 1
+#include "defs.h"
#include "shared/ffmpeg_raii.h"
+#include "shared/httpd.h"
#include "ref_counted_frame.h"
extern "C" {
#include <movit/effect.h>
#include <va/va.h>
-class HTTPD;
struct jpeg_compress_struct;
struct VADisplayWithCleanup;
struct VectorDestinationManager;
void va_receiver_thread_func();
void encode_jpeg_va(QueuedFrame &&qf);
std::vector<uint8_t> encode_jpeg_libjpeg(const QueuedFrame &qf);
- void write_mjpeg_packet(int64_t pts, unsigned card_index, const uint8_t *jpeg, size_t jpeg_size);
- void write_audio_packet(int64_t pts, unsigned card_index, const std::vector<int32_t> &audio);
+ void write_mjpeg_packet(AVFormatContext *avctx, int64_t pts, unsigned stream_index, const uint8_t *jpeg, size_t jpeg_size);
+ void write_audio_packet(AVFormatContext *avctx, int64_t pts, unsigned stream_index, const std::vector<int32_t> &audio);
void init_jpeg_422(unsigned width, unsigned height, const movit::RGBTriplet &white_balance, VectorDestinationManager *dest, jpeg_compress_struct *cinfo);
std::vector<uint8_t> get_jpeg_header(unsigned width, unsigned height, const movit::RGBTriplet &white_balance, jpeg_compress_struct *cinfo);
+ void add_stream(HTTPD::StreamID stream_id); // Can only be called from the constructor, or the thread owning <streams>.
+ void update_siphon_streams(); // Same.
+ void create_ffmpeg_context(HTTPD::StreamID stream_id);
+ struct WritePacket2Context {
+ MJPEGEncoder *mjpeg_encoder;
+ HTTPD::StreamID stream_id;
+ };
+ std::map<HTTPD::StreamID, WritePacket2Context> ffmpeg_contexts; // Statically set up, so we never need to worry about dangling pointers.
static int write_packet2_thunk(void *opaque, uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time);
- int write_packet2(uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time);
+ int write_packet2(HTTPD::StreamID stream_id, uint8_t *buf, int buf_size, AVIODataMarkerType type, int64_t time);
std::thread encoder_thread, va_receiver_thread;
std::queue<QueuedFrame> frames_encoding; // Under mu. Used for VA-API only.
std::condition_variable any_frames_encoding;
- AVFormatContextWithCloser avctx;
+ struct Stream {
+ AVFormatContextWithCloser avctx;
+ std::string mux_header;
+ };
+ std::map<HTTPD::StreamID, Stream> streams; // Owned by the VA-API receiver thread if VA-API is active, or the encoder thread if not.
HTTPD *httpd;
- std::string mux_header;
std::atomic<bool> should_quit{false};
bool running = false;
if (type == AVIO_DATA_MARKER_HEADER) {
stream_mux_header.append((char *)buf, buf_size);
- httpd->set_header(HTTPD::MAIN_STREAM, stream_mux_header);
+ httpd->set_header(HTTPD::StreamID{ HTTPD::MAIN_STREAM, 0 }, stream_mux_header);
} else {
- httpd->add_data(HTTPD::MAIN_STREAM, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
+ httpd->add_data(HTTPD::StreamID{ HTTPD::MAIN_STREAM, 0 }, (char *)buf, buf_size, type == AVIO_DATA_MARKER_SYNC_POINT, time, AVRational{ AV_TIME_BASE, 1 });
}
return buf_size;
}
{
global_metrics.add("num_connected_clients", &metric_num_connected_clients, Metrics::TYPE_GAUGE);
global_metrics.add("num_connected_multicam_clients", &metric_num_connected_multicam_clients, Metrics::TYPE_GAUGE);
+ for (unsigned stream_idx = 0; stream_idx < MAX_VIDEO_CARDS; ++stream_idx) {
+ global_metrics.add("num_connected_siphon_clients",
+ {{ "card", to_string(stream_idx) }},
+ &metric_num_connected_siphon_clients[stream_idx], Metrics::TYPE_GAUGE);
+ }
}
HTTPD::~HTTPD()
}
}
-void HTTPD::add_data(StreamType stream_type, const char *buf, size_t size, bool keyframe, int64_t time, AVRational timebase)
+void HTTPD::set_header(StreamID stream_id, const string &data)
+{
+ lock_guard<mutex> lock(streams_mutex);
+ header[stream_id] = data;
+ add_data_locked(stream_id, data.data(), data.size(), Stream::DATA_TYPE_HEADER, AV_NOPTS_VALUE, AVRational{ 1, 0 });
+}
+
+void HTTPD::add_data(StreamID stream_id, const char *buf, size_t size, bool keyframe, int64_t time, AVRational timebase)
{
lock_guard<mutex> lock(streams_mutex);
+ add_data_locked(stream_id, buf, size, keyframe ? Stream::DATA_TYPE_KEYFRAME : Stream::DATA_TYPE_OTHER, time, timebase);
+}
+
+void HTTPD::add_data_locked(StreamID stream_id, const char *buf, size_t size, Stream::DataType data_type, int64_t time, AVRational timebase)
+{
for (Stream *stream : streams) {
- if (stream->get_stream_type() == stream_type) {
- stream->add_data(buf, size, keyframe ? Stream::DATA_TYPE_KEYFRAME : Stream::DATA_TYPE_OTHER, time, timebase);
+ if (stream->get_stream_id() == stream_id) {
+ stream->add_data(buf, size, data_type, time, timebase);
}
}
}
} else {
framing = HTTPD::Stream::FRAMING_RAW;
}
- HTTPD::StreamType stream_type;
+ HTTPD::StreamID stream_id;
if (strcmp(url, "/multicam.mp4") == 0) {
- stream_type = HTTPD::StreamType::MULTICAM_STREAM;
+ stream_id.type = HTTPD::StreamType::MULTICAM_STREAM;
+ stream_id.index = 0;
+ } else if (strncmp(url, "/feeds/", 7) == 0) {
+ stream_id.type = HTTPD::StreamType::SIPHON_STREAM;
+ stream_id.index = atoi(url + 7);
} else {
- stream_type = HTTPD::StreamType::MAIN_STREAM;
+ stream_id.type = HTTPD::StreamType::MAIN_STREAM;
+ stream_id.index = 0;
}
if (strcmp(url, "/metrics") == 0) {
return ret;
}
- HTTPD::Stream *stream = new HTTPD::Stream(this, framing, stream_type);
- stream->add_data(header[stream_type].data(), header[stream_type].size(), Stream::DATA_TYPE_HEADER, AV_NOPTS_VALUE, AVRational{ 1, 0 });
+ HTTPD::Stream *stream = new HTTPD::Stream(this, framing, stream_id);
+ const string &hdr = header[stream_id];
+ stream->add_data(hdr.data(), hdr.size(), Stream::DATA_TYPE_HEADER, AV_NOPTS_VALUE, AVRational{ 1, 0 });
{
lock_guard<mutex> lock(streams_mutex);
streams.insert(stream);
}
++metric_num_connected_clients;
- if (stream_type == HTTPD::StreamType::MULTICAM_STREAM) {
+ if (stream_id.type == HTTPD::StreamType::MULTICAM_STREAM) {
++metric_num_connected_multicam_clients;
}
+ if (stream_id.type == HTTPD::StreamType::SIPHON_STREAM) {
+ ++metric_num_connected_siphon_clients[stream_id.index];
+ }
*con_cls = stream;
// Does not strictly have to be equal to MUX_BUFFER_SIZE.
{
HTTPD::Stream *stream = (HTTPD::Stream *)cls;
HTTPD *httpd = stream->get_parent();
- if (stream->get_stream_type() == HTTPD::StreamType::MULTICAM_STREAM) {
+ if (stream->get_stream_id().type == HTTPD::StreamType::MULTICAM_STREAM) {
--httpd->metric_num_connected_multicam_clients;
}
+ if (stream->get_stream_id().type == HTTPD::StreamType::SIPHON_STREAM) {
+ --httpd->metric_num_connected_siphon_clients[stream->get_stream_id().index];
+ }
{
lock_guard<mutex> lock(httpd->streams_mutex);
delete stream;
// A class dealing with stream output to HTTP.
+#include <assert.h>
#include <atomic>
#include <condition_variable>
#include <deque>
#include <stdint.h>
#include <string>
#include <sys/types.h>
+#include <map>
#include <unordered_map>
#include <utility>
#include <libavutil/rational.h>
}
+#include "shared/shared_defs.h"
+
struct MHD_Connection;
struct MHD_Daemon;
enum StreamType {
MAIN_STREAM,
MULTICAM_STREAM,
- NUM_STREAM_TYPES
+ SIPHON_STREAM // The only one that can have stream_index != 0.
+ };
+ struct StreamID {
+ StreamType type;
+ unsigned index;
+
+ bool operator< (const StreamID &other) const {
+ if (type != other.type)
+ return type < other.type;
+ return index < other.index;
+ }
+ bool operator== (const StreamID &other) const {
+ return (type == other.type && index == other.index);
+ }
};
-
- // Should be called before start().
- void set_header(StreamType stream_type, const std::string &data) {
- header[stream_type] = data;
- }
// Should be called before start() (due to threading issues).
enum CORSPolicy {
void start(int port);
void stop();
- void add_data(StreamType stream_type, const char *buf, size_t size, bool keyframe, int64_t time, AVRational timebase);
+ void set_header(StreamID stream_id, const std::string &data);
+ void add_data(StreamID stream_id, const char *buf, size_t size, bool keyframe, int64_t time, AVRational timebase);
int64_t get_num_connected_clients() const
{
return metric_num_connected_clients.load();
int64_t get_num_connected_multicam_clients() const {
return metric_num_connected_multicam_clients.load();
}
+ int64_t get_num_connected_siphon_clients(unsigned stream_idx) const {
+ assert(stream_idx < MAX_VIDEO_CARDS);
+ return metric_num_connected_siphon_clients[stream_idx].load();
+ }
private:
static int answer_to_connection_thunk(void *cls, MHD_Connection *connection,
FRAMING_RAW,
FRAMING_METACUBE
};
- Stream(HTTPD *parent, Framing framing, StreamType stream_type)
- : parent(parent), framing(framing), stream_type(stream_type) {}
+ Stream(HTTPD *parent, Framing framing, StreamID stream_id)
+ : parent(parent), framing(framing), stream_id(stream_id) {}
static ssize_t reader_callback_thunk(void *cls, uint64_t pos, char *buf, size_t max);
ssize_t reader_callback(uint64_t pos, char *buf, size_t max);
void add_data(const char *buf, size_t size, DataType data_type, int64_t time, AVRational timebase);
void stop();
HTTPD *get_parent() const { return parent; }
- StreamType get_stream_type() const { return stream_type; }
+ StreamID get_stream_id() const { return stream_id; }
private:
HTTPD *parent;
size_t used_of_buffered_data = 0; // How many bytes of the first element of <buffered_data> that is already used. Protected by <buffer_mutex>.
size_t buffered_data_bytes = 0; // The sum of all size() in buffered_data. Protected by <buffer_mutex>.
size_t seen_keyframe = false;
- StreamType stream_type;
+ StreamID stream_id;
};
+ void add_data_locked(StreamID stream_id, const char *buf, size_t size, Stream::DataType data_type, int64_t time, AVRational timebase);
+
MHD_Daemon *mhd = nullptr;
std::mutex streams_mutex;
std::set<Stream *> streams; // Not owned.
CORSPolicy cors_policy;
};
std::unordered_map<std::string, Endpoint> endpoints;
- std::string header[NUM_STREAM_TYPES];
+ std::map<StreamID, std::string> header;
// Metrics.
std::atomic<int64_t> metric_num_connected_clients{0};
std::atomic<int64_t> metric_num_connected_multicam_clients{0};
+ std::atomic<int64_t> metric_num_connected_siphon_clients[MAX_VIDEO_CARDS] {{0}};
};
#endif // !defined(_HTTPD_H)
// the output to be very uneven.
#define MUX_BUFFER_SIZE 10485760
+#define MAX_VIDEO_CARDS 16 // Only really used by Nageru.
+
#endif // !defined(_SHARED_DEFS_H)