From: Steinar H. Gunderson Date: Sat, 24 Oct 2015 23:15:09 +0000 (+0200) Subject: Add a HTTP server for stream output. X-Git-Tag: 1.0.0~225 X-Git-Url: https://git.sesse.net/?p=nageru;a=commitdiff_plain;h=b4f16ea9f8969a3ba14be8cd9c88cfe00d19533b Add a HTTP server for stream output. --- diff --git a/Makefile b/Makefile index 7a88cfe..d056084 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ CXX=g++ -PKG_MODULES = Qt5Core Qt5Gui Qt5Widgets Qt5OpenGLExtensions Qt5OpenGL libusb-1.0 movit lua5.2 +PKG_MODULES = Qt5Core Qt5Gui Qt5Widgets Qt5OpenGLExtensions Qt5OpenGL libusb-1.0 movit lua5.2 libmicrohttpd CXXFLAGS := -O2 -march=native -g -std=gnu++11 -Wall -Wno-deprecated-declarations -Werror -fPIC $(shell pkg-config --cflags $(PKG_MODULES)) -pthread -DMOVIT_SHADER_DIR=\"$(shell pkg-config --variable=shaderdir movit)\" LDFLAGS=$(shell pkg-config --libs $(PKG_MODULES)) -lEGL -lGL -pthread -lva -lva-drm -lva-x11 -lX11 -lavformat -lavcodec -lavutil -lzita-resampler @@ -8,7 +8,7 @@ OBJS=glwidget.o main.o mainwindow.o window.o OBJS += glwidget.moc.o mainwindow.moc.o window.moc.o # Mixer objects -OBJS += h264encode.o mixer.o bmusb/bmusb.o pbo_frame_allocator.o context.o ref_counted_frame.o theme.o resampler.o +OBJS += h264encode.o mixer.o bmusb/bmusb.o pbo_frame_allocator.o context.o ref_counted_frame.o theme.o resampler.o httpd.o %.o: %.cpp $(CXX) -MMD -MP $(CPPFLAGS) $(CXXFLAGS) -o $@ -c $< diff --git a/h264encode.cpp b/h264encode.cpp index 5fa01bd..3880c18 100644 --- a/h264encode.cpp +++ b/h264encode.cpp @@ -26,6 +26,7 @@ #include #include "context.h" +#include "httpd.h" #include "timebase.h" class QOpenGLContext; @@ -1666,6 +1667,7 @@ int H264Encoder::save_codeddata(storage_task task) pkt.flags = 0; } //pkt.duration = 1; + httpd->add_packet(pkt); av_interleaved_write_frame(avctx, &pkt); } // Encode and add all audio frames up to and including the pts of this video frame. @@ -1704,6 +1706,7 @@ int H264Encoder::save_codeddata(storage_task task) pkt.pts = av_rescale_q(audio_pts + global_delay, AVRational{1, TIMEBASE}, avstream_audio->time_base); pkt.dts = pkt.pts; pkt.stream_index = 1; + httpd->add_packet(pkt); av_interleaved_write_frame(avctx, &pkt); } // TODO: Delayed frames. @@ -1820,11 +1823,10 @@ static int print_input() //H264Encoder::H264Encoder(SDL_Window *window, SDL_GLContext context, int width, int height, const char *output_filename) -H264Encoder::H264Encoder(QSurface *surface, int width, int height, const char *output_filename) - : current_storage_frame(0), surface(surface) +H264Encoder::H264Encoder(QSurface *surface, int width, int height, const char *output_filename, HTTPD *httpd) + : current_storage_frame(0), surface(surface), httpd(httpd) //: width(width), height(height), current_encoding_frame(0) { - av_register_all(); avctx = avformat_alloc_context(); avctx->oformat = av_guess_format(NULL, output_filename, NULL); strcpy(avctx->filename, output_filename); diff --git a/h264encode.h b/h264encode.h index e4ef1ab..48dd329 100644 --- a/h264encode.h +++ b/h264encode.h @@ -45,13 +45,14 @@ extern "C" { #include "ref_counted_frame.h" #include "ref_counted_gl_sync.h" +class HTTPD; class QSurface; #define SURFACE_NUM 16 /* 16 surfaces for source YUV */ class H264Encoder { public: - H264Encoder(QSurface *surface, int width, int height, const char *output_filename); + H264Encoder(QSurface *surface, int width, int height, const char *output_filename, HTTPD *httpd); ~H264Encoder(); //void add_frame(FrameAllocator::Frame frame, GLsync fence); @@ -111,6 +112,7 @@ private: AVFormatContext *avctx; AVStream *avstream_video; AVStream *avstream_audio; + HTTPD *httpd; }; #endif diff --git a/httpd.cpp b/httpd.cpp new file mode 100644 index 0000000..3485a83 --- /dev/null +++ b/httpd.cpp @@ -0,0 +1,178 @@ +#include +#include +#include + +extern "C" { +#include +} + +#include "httpd.h" +#include "timebase.h" + +using namespace std; + +HTTPD::HTTPD() {} + +void HTTPD::start(int port) +{ + MHD_start_daemon(MHD_USE_THREAD_PER_CONNECTION | MHD_USE_POLL_INTERNALLY | MHD_USE_DUAL_STACK, + port, + nullptr, nullptr, + &answer_to_connection_thunk, this, MHD_OPTION_END); +} + +void HTTPD::add_packet(const AVPacket &pkt) +{ + for (Stream *stream : streams) { + stream->add_packet(pkt); + } +} + +int HTTPD::answer_to_connection_thunk(void *cls, MHD_Connection *connection, + const char *url, const char *method, + const char *version, const char *upload_data, + size_t *upload_data_size, void **con_cls) +{ + HTTPD *httpd = (HTTPD *)cls; + return httpd->answer_to_connection(connection, url, method, version, upload_data, upload_data_size, con_cls); +} + +int HTTPD::answer_to_connection(MHD_Connection *connection, + const char *url, const char *method, + const char *version, const char *upload_data, + size_t *upload_data_size, void **con_cls) +{ + printf("url %s\n", url); + AVOutputFormat *oformat = av_guess_format("mpegts", nullptr, nullptr); + assert(oformat != nullptr); + HTTPD::Stream *stream = new HTTPD::Stream(oformat); + streams.push_back(stream); + MHD_Response *response = MHD_create_response_from_callback( + (size_t)-1, 1048576, &HTTPD::Stream::reader_callback_thunk, stream, &HTTPD::free_stream); + int ret = MHD_queue_response(connection, MHD_HTTP_OK, response); + //MHD_destroy_response(response); + + return ret; +} + +void HTTPD::free_stream(void *cls) +{ + HTTPD::Stream *stream = (HTTPD::Stream *)cls; + delete stream; +} + +HTTPD::Stream::Stream(AVOutputFormat *oformat) +{ + avctx = avformat_alloc_context(); + avctx->oformat = oformat; + uint8_t *buf = (uint8_t *)av_malloc(1048576); + avctx->pb = avio_alloc_context(buf, 1048576, 1, this, nullptr, &HTTPD::Stream::write_packet_thunk, nullptr); + avctx->flags = AVFMT_FLAG_CUSTOM_IO; + + // TODO: Unify with the code in h264encoder.cpp. + AVCodec *codec_video = avcodec_find_encoder(AV_CODEC_ID_H264); + avstream_video = avformat_new_stream(avctx, codec_video); + if (avstream_video == nullptr) { + fprintf(stderr, "avformat_new_stream() failed\n"); + exit(1); + } + avstream_video->time_base = AVRational{1, TIMEBASE}; + avstream_video->codec->width = 1280; // FIXME + avstream_video->codec->height = 720; // FIXME + avstream_video->codec->time_base = AVRational{1, TIMEBASE}; + avstream_video->codec->ticks_per_frame = 1; // or 2? + + AVCodec *codec_audio = avcodec_find_encoder(AV_CODEC_ID_MP3); + avstream_audio = avformat_new_stream(avctx, codec_audio); + if (avstream_audio == nullptr) { + fprintf(stderr, "avformat_new_stream() failed\n"); + exit(1); + } + avstream_audio->time_base = AVRational{1, TIMEBASE}; + avstream_audio->codec->bit_rate = 256000; + avstream_audio->codec->sample_rate = 48000; + avstream_audio->codec->sample_fmt = AV_SAMPLE_FMT_FLTP; + avstream_audio->codec->channels = 2; + avstream_audio->codec->channel_layout = AV_CH_LAYOUT_STEREO; + avstream_audio->codec->time_base = AVRational{1, TIMEBASE}; + + if (avformat_write_header(avctx, NULL) < 0) { + fprintf(stderr, "avformat_write_header() failed\n"); + exit(1); + } +} + +HTTPD::Stream::~Stream() +{ + avformat_free_context(avctx); +} + +ssize_t HTTPD::Stream::reader_callback_thunk(void *cls, uint64_t pos, char *buf, size_t max) +{ + HTTPD::Stream *stream = (HTTPD::Stream *)cls; + return stream->reader_callback(pos, buf, max); +} + +ssize_t HTTPD::Stream::reader_callback(uint64_t pos, char *buf, size_t max) +{ + unique_lock lock(buffer_mutex); + has_buffered_data.wait(lock, [this]{ return !buffered_data.empty(); }); + + ssize_t ret = 0; + while (max > 0 && !buffered_data.empty()) { + const string &s = buffered_data.front(); + assert(s.size() > used_of_buffered_data); + size_t len = s.size() - used_of_buffered_data; + if (max >= len) { + // Consume the entire (rest of the) string. + memcpy(buf, s.data() + used_of_buffered_data, len); + ret += len; + max -= len; + buffered_data.pop_front(); + used_of_buffered_data = 0; + } else { + // We don't need the entire string; just use the first part of it. + memcpy(buf, s.data() + used_of_buffered_data, max); + used_of_buffered_data += max; + ret += max; + max = 0; + } + } + + return ret; +} + +void HTTPD::Stream::add_packet(const AVPacket &pkt) +{ + AVPacket pkt_copy; + av_copy_packet(&pkt_copy, &pkt); + if (pkt.stream_index == 0) { + pkt_copy.pts = av_rescale_q(pkt.pts, AVRational{1, TIMEBASE}, avstream_video->time_base); + pkt_copy.dts = av_rescale_q(pkt.dts, AVRational{1, TIMEBASE}, avstream_video->time_base); + } else if (pkt.stream_index == 1) { + pkt_copy.pts = av_rescale_q(pkt.pts, AVRational{1, TIMEBASE}, avstream_audio->time_base); + pkt_copy.dts = av_rescale_q(pkt.dts, AVRational{1, TIMEBASE}, avstream_audio->time_base); + } else { + assert(false); + } + + if (av_interleaved_write_frame(avctx, &pkt_copy) < 0) { + fprintf(stderr, "av_interleaved_write_frame() failed\n"); + exit(1); + } +} + +int HTTPD::Stream::write_packet_thunk(void *opaque, uint8_t *buf, int buf_size) +{ + HTTPD::Stream *stream = (HTTPD::Stream *)opaque; + return stream->write_packet(buf, buf_size); +} + +int HTTPD::Stream::write_packet(uint8_t *buf, int buf_size) +{ + unique_lock lock(buffer_mutex); + buffered_data.emplace_back((char *)buf, buf_size); + has_buffered_data.notify_all(); + return buf_size; +} + diff --git a/httpd.h b/httpd.h new file mode 100644 index 0000000..0a092aa --- /dev/null +++ b/httpd.h @@ -0,0 +1,61 @@ +#ifndef _HTTPD_H +#define _HTTPD_H + +#include +#include +#include +#include +#include +#include + +extern "C" { +#include +} + +class HTTPD { +public: + HTTPD(); + void start(int port); + void add_packet(const AVPacket &pkt); + +private: + static int answer_to_connection_thunk(void *cls, MHD_Connection *connection, + const char *url, const char *method, + const char *version, const char *upload_data, + size_t *upload_data_size, void **con_cls); + + int answer_to_connection(MHD_Connection *connection, + const char *url, const char *method, + const char *version, const char *upload_data, + size_t *upload_data_size, void **con_cls); + + static void free_stream(void *cls); + + class Stream { + public: + Stream(AVOutputFormat *oformat); + ~Stream(); + + 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_packet(const AVPacket &pkt); + + private: + static int write_packet_thunk(void *opaque, uint8_t *buf, int buf_size); + int write_packet(uint8_t *buf, int buf_size); + + AVIOContext *avio; + AVFormatContext *avctx; + AVStream *avstream_video, *avstream_audio; + + std::mutex buffer_mutex; + std::condition_variable has_buffered_data; + std::deque buffered_data; // Protected by . + size_t used_of_buffered_data = 0; // How many bytes of the first element of that is already used. Protected by . + }; + + std::vector streams; // Not owned. +}; + +#endif // !defined(_HTTPD_H) diff --git a/main.cpp b/main.cpp index cc554cf..6e95d18 100644 --- a/main.cpp +++ b/main.cpp @@ -15,6 +15,7 @@ int main(int argc, char *argv[]) { setenv("QT_XCB_GL_INTEGRATION", "xcb_egl", 0); setlinebuf(stdout); + av_register_all(); QCoreApplication::setAttribute(Qt::AA_ShareOpenGLContexts, true); QApplication app(argc, argv); diff --git a/mixer.cpp b/mixer.cpp index 90a00b8..ca55be0 100644 --- a/mixer.cpp +++ b/mixer.cpp @@ -74,6 +74,8 @@ Mixer::Mixer(const QSurfaceFormat &format) : mixer_surface(create_surface(format)), h264_encoder_surface(create_surface(format)) { + httpd.start(9095); + CHECK(init_movit(MOVIT_SHADER_DIR, MOVIT_DEBUG_OFF)); check_error(); @@ -97,7 +99,7 @@ Mixer::Mixer(const QSurfaceFormat &format) display_chain->set_dither_bits(0); // Don't bother. display_chain->finalize(); - h264_encoder.reset(new H264Encoder(h264_encoder_surface, WIDTH, HEIGHT, "test.mp4")); + h264_encoder.reset(new H264Encoder(h264_encoder_surface, WIDTH, HEIGHT, "test.ts", &httpd)); for (int card_index = 0; card_index < NUM_CARDS; ++card_index) { printf("Configuring card %d...\n", card_index); diff --git a/mixer.h b/mixer.h index 45311aa..a177854 100644 --- a/mixer.h +++ b/mixer.h @@ -17,6 +17,7 @@ #include "theme.h" #include "resampler.h" #include "timebase.h" +#include "httpd.h" #define NUM_CARDS 2 @@ -94,6 +95,8 @@ private: void release_display_frame(DisplayFrame *frame); double pts() { return double(pts_int) / TIMEBASE; } + HTTPD httpd; + QSurface *mixer_surface, *h264_encoder_surface; std::unique_ptr resource_pool; std::unique_ptr theme;