I try to use GStreamer to stream H264 packets (encoded by NVENC), to VLC (or ffplay), with RT(S)P.
Everything works well on the GStreamer side, the pipeline is built, the caps are negociated correctly, Wireshark shows me that the SDP dynamically generated when the client connects, the data feeding callbacks are called… the PLAYING state is switched on…
but nothing is visible in VLC or ffplay, it get stalled.
I tried UDP, TCP transports configurations for the clients without success.
With wireshark, I can’t see any H264 data packet being emitted by gstreamer, even if buffers have been pushed.
I have read tons of tutorials, debug tricks, but I can’t get any further.
Could anyone help me understand what could be wrong ?
Below is my code (simulating input from a file containing my raw NALUs from the H264 encoder)
(and the video.h264 used as input to help debugging)
I should mention that the same raw H264 packets work correctly with Live555+VLC
#include "stdafx.h"
#include <atomic>
#include <condition_variable>
#include <filesystem>
#include <iostream>
#include <queue>
#include <sstream>
#include <string>
#include <thread>
#include <vector>
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>
#include <gst/rtsp-server/rtsp-server.h>
inline uint8_t get_nalu_type(const unsigned char* data, size_t size)
{
uint8_t result = 0;
// Look for start code pattern
size_t offset = size;
if (size >= 5 && data[0] == 0 && data[1] == 0 && data[2] == 0 && data[3] == 1)
offset = 4;
else if (size >= 4 && data[0] == 0 && data[1] == 0 && data[2] == 1)
offset = 3;
if (offset < size)
result = data[offset] & 0x1F;
return result;
}
//end get_nalu_type()
void print_nalu_type(const unsigned char* data, size_t size)
{
uint8_t nal_type = get_nalu_type(data, size);
std::cout << "NALU type: " << "0x" << std::hex << (int)nal_type << std::dec << " ";
switch(nal_type)
{
case 7: std::cout << "(SPS)" << std::endl; break;
case 8: std::cout << "(PPS)" << std::endl; break;
case 5: std::cout << "(IDR)" << std::endl; break;
case 1: std::cout << "(P-frame)" << std::endl; break;
default: std::cout << "(other)" << std::endl; break;
}//end switch(nal_type)
}
//end print_nalu_type()
class _EXRTSPStreamerGStreamer
{
public:
_EXRTSPStreamerGStreamer(unsigned __int16 aPort, const std::string& aStreamName)
: mount_point_("/"+aStreamName), port_(aPort), loop_(nullptr),
server_(nullptr), main_loop_thread_(), running_(false) {
}//end _EXRTSPStreamerGStreamer()
_EXRTSPStreamerGStreamer(const _EXRTSPStreamerGStreamer&) = delete;
_EXRTSPStreamerGStreamer(_EXRTSPStreamerGStreamer&&) noexcept = delete;
~_EXRTSPStreamerGStreamer() {
stop();
}//end ~_EXRTSPStreamerGStreamer()
public:
_EXRTSPStreamerGStreamer& operator=(const _EXRTSPStreamerGStreamer&) = delete;
_EXRTSPStreamerGStreamer& operator=(_EXRTSPStreamerGStreamer&&) noexcept = delete;
public:
bool start()
{
if (running_) return false;
loop_ = g_main_loop_new(nullptr, FALSE);
factory_ = gst_rtsp_media_factory_new();
server_ = gst_rtsp_server_new();
gst_rtsp_server_set_service(server_, std::to_string(port_).c_str());
// Launch pipeline with appsrc feeding raw H264, multicast enabled
const char *launch_desc =
"( appsrc name=mysrc emit-signals=true is-live=true format=time "
//"caps=video/x-h264,stream-format=byte-stream,alignment=au "
"caps=video/x-h264,stream-format=byte-stream,alignment=nal "
"! h264parse name=parser0 config-interval=1 "
"! rtph264pay name=pay0 pt=96 "
//"! application/x-rtp,media=video,encoding-name=H264,payload=96,clock-rate=90000 "
//"! udpsink name=sink host=127.0.0.1 port=8554 sync=false "
")";
gst_rtsp_media_factory_set_launch(factory_, launch_desc);
gst_rtsp_media_factory_set_shared(factory_, TRUE);
gst_rtsp_media_factory_set_latency(factory_, 0);
gst_rtsp_media_factory_set_transport_mode(factory_, GST_RTSP_TRANSPORT_MODE_PLAY);
gst_rtsp_media_factory_set_protocols(factory_, (GstRTSPLowerTrans) (GST_RTSP_LOWER_TRANS_UDP | GST_RTSP_LOWER_TRANS_TCP));
gst_rtsp_media_factory_set_media_gtype(factory_, gst_rtsp_media_get_type());
GstRTSPMountPoints *mounts = gst_rtsp_server_get_mount_points(server_);
g_object_ref(factory_);//should be taken by gst_rtsp_mount_points_add_factory, for some reason is not
gst_rtsp_mount_points_add_factory(mounts, mount_point_.c_str(), factory_);
g_object_unref(mounts);
// Setup callbacks to get appsrc element and push buffers
g_signal_connect(factory_, "media-configure", G_CALLBACK(media_configure_cb), this);
g_signal_connect(server_, "client-connected", G_CALLBACK(client_connected_cb), this);
this->serverAttachment = gst_rtsp_server_attach(server_, nullptr);
running_ = true;
// Start main loop thread
main_loop_thread_ = std::thread([this]() {
g_main_loop_run(loop_);
});
std::cout << "RTSP server started at rtsp://127.0.0.1:" << port_ << mount_point_ << std::endl;
return true;
}
void stop() {
if (!running_) return;
running_ = false;
if (this->serverAttachment != 0) {
g_source_remove(this->serverAttachment);
this->serverAttachment = 0;
}
// Stop main loop
if (loop_) {
g_main_loop_quit(loop_);
}
// Join thread
if (main_loop_thread_.joinable()) {
main_loop_thread_.join();
}
// Cleanup GStreamer objects
this->activeClientsCount = 0;
if (appsrc_) {
g_object_unref(appsrc_);
appsrc_ = nullptr;
}
if (pipeline_) {
g_object_unref(pipeline_);
pipeline_ = nullptr;
}
if (server_) {
g_object_unref(server_);
server_ = nullptr;
}
if (factory_) {
g_object_unref(factory_);
factory_ = nullptr;
}
if (loop_) {
g_main_loop_unref(loop_);
loop_ = nullptr;
}
// Clear queue
std::lock_guard<std::mutex> lock(queue_mutex_);
while (!buffer_queue_.empty())
{
gst_buffer_unref(buffer_queue_.front());
buffer_queue_.pop();
}
this->enoughDataFlag = false;
std::cout << "RTSP server stopped." << std::endl;
}
// Push raw H264 NAL unit (including SPS/PPS as needed)
void pushData(const uint8_t* data, size_t size, GstClockTime pts = GST_CLOCK_TIME_NONE) {
if (!running_) return;
if (!appsrc_) return;
if (!this->activeClientsCount) return;
if (this->enoughDataFlag) return;
guint64 level = 0;
g_object_get(this->appsrc_, "current-level-bytes", &level, NULL);
printf("Appsrc buffer level : %llu\r\n", (unsigned long long)level);
// Copy data to a GstBuffer
GstBuffer *buffer = gst_buffer_new_allocate(nullptr, size, nullptr);
gst_buffer_fill(buffer, 0, data, size);
if (pts == GST_CLOCK_TIME_NONE) {
pts = gst_util_get_timestamp();
}
GST_BUFFER_PTS(buffer) = pts;
GST_BUFFER_DTS(buffer) = GST_BUFFER_PTS(buffer);
GST_BUFFER_DURATION(buffer) = gst_util_uint64_scale_int(1, GST_SECOND, 30); // assuming 30 fps
const uint8_t naluType = get_nalu_type(data, size);
const bool isSPS = (naluType == 7);
const bool isPPS = (naluType == 8);
const bool isIFrame = (naluType == 5);
const bool isPFrame = (naluType == 1);
GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_LIVE);
if (isSPS || isPPS)
{
GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_HEADER);
GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_DECODE_ONLY);
}//end if (isSPS || isPPS)
else//if (!isSPS && !isPPS)
{
GST_BUFFER_FLAG_UNSET(buffer, GST_BUFFER_FLAG_HEADER);
GST_BUFFER_FLAG_UNSET(buffer, GST_BUFFER_FLAG_DECODE_ONLY);
}//end if (!isSPS && !isPPS)
if (isIFrame)
GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_RESYNC);
else//if (!isIFrame)
GST_BUFFER_FLAG_UNSET(buffer, GST_BUFFER_FLAG_RESYNC);
if (isPFrame)
GST_BUFFER_FLAG_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT);
else//if (!isPFrame)
GST_BUFFER_FLAG_UNSET(buffer, GST_BUFFER_FLAG_DELTA_UNIT);
// Push buffer into queue
{
std::lock_guard<std::mutex> lock(this->queue_mutex_);
buffer_queue_.push(buffer);
}
this->queue_cond_.notify_one();
}
void push_buffers_from_queue() {
bool stop = false;
while(!stop)
{
GstBuffer* buffer = nullptr;
{
std::unique_lock locker(this->queue_mutex_);
if (buffer_queue_.empty()) {
// Wait a little bit for new buffers, non-blocking to keep main loop responsive
this->queue_cond_.wait_for(locker, std::chrono::milliseconds(10));
}
if (!buffer_queue_.empty())
{
buffer = buffer_queue_.front();
buffer_queue_.pop();
}//end if (!buffer_queue_.empty())
}
if (buffer != nullptr)
{
GstFlowReturn ret = gst_app_src_push_buffer(GST_APP_SRC(appsrc_), buffer);
if (ret != GST_FLOW_OK) {
std::cerr << "gst_app_src_push_buffer failed: " << ret << std::endl;
gst_buffer_unref(buffer);
}
}
stop |= !buffer;
}
}
private:
std::string mount_point_;
int port_ = 0;
GMainLoop *loop_ = nullptr;
GstRTSPServer *server_ = nullptr;
guint serverAttachment = 0;
GstRTSPMediaFactory *factory_ = nullptr;
GstElement *appsrc_ = nullptr;
GstElement *pipeline_ = nullptr;
unsigned int activeClientsCount = 0;
std::atomic<bool> enoughDataFlag;
std::thread main_loop_thread_;
std::atomic<bool> running_;
std::mutex queue_mutex_;
std::condition_variable queue_cond_;
std::queue<GstBuffer *> buffer_queue_;
static void client_connected_cb(GstRTSPServer* server, GstRTSPClient* client, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
g_signal_connect(client, "options-request", G_CALLBACK(client_options_request_cb), self);
g_signal_connect(client, "play-request", G_CALLBACK(client_play_request_cb), self);
}
static void client_options_request_cb(GstRTSPClient* client, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
printf("client options-request called\n");
}
static void client_play_request_cb(GstRTSPClient* client, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
printf("client play-request called\n");
}
static void media_configure_cb(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
g_signal_connect(media, "prepared", G_CALLBACK(media_prepared_cb), self);
g_signal_connect(media, "unprepared", G_CALLBACK(media_unprepared_cb), self);
//g_signal_connect(media, "target-state", G_CALLBACK(media_target_state_cb), self);
GstElement* element = gst_rtsp_media_get_element(media);
gst_debug_bin_to_dot_file(GST_BIN(element), GST_DEBUG_GRAPH_SHOW_ALL, "rtsp-pipeline");
self->pipeline_ = (GstElement*)gst_object_ref(element);
g_signal_connect(element, "state-changed", G_CALLBACK(media_state_changed_cb), self);
if (self->appsrc_ != 0)
{
g_object_unref(self->appsrc_);
self->appsrc_ = 0;
}
self->appsrc_ = gst_bin_get_by_name_recurse_up(GST_BIN(element), "mysrc");
if (!self->appsrc_) {
printf("Failed to get appsrc 'mysrc'\n");
return;
}
++self->activeClientsCount;
g_object_set(self->appsrc_, "stream-type", GST_APP_STREAM_TYPE_STREAM, "format", GST_FORMAT_TIME, "is-live", TRUE, nullptr);
g_object_set(self->appsrc_, "block", FALSE, nullptr);
/*g_object_set(self->appsrc_, "min-percent", 0, nullptr);
g_object_set(self->appsrc_, "emit-signals", TRUE, nullptr);*/
//GstCaps *caps = gst_caps_from_string("video/x-h264,stream-format=byte-stream,alignment=au");
/*GstCaps *caps = gst_caps_from_string("video/x-h264,stream-format=byte-stream,alignment=nal");
g_object_set(self->appsrc_, "caps", caps, nullptr);
gst_caps_unref(caps);*/
{
GstPad* srcpad = gst_element_get_static_pad(self->appsrc_, "src");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_appsrc_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstPad* srcpad = gst_element_get_static_pad(self->appsrc_, "sink");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_appsrc_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* h264parse = gst_bin_get_by_name(GST_BIN(self->pipeline_), "parser0");
GstPad* srcpad = gst_element_get_static_pad(h264parse, "src");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_parser_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* h264parse = gst_bin_get_by_name(GST_BIN(self->pipeline_), "parser0");
GstPad* srcpad = gst_element_get_static_pad(h264parse, "sink");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_parser_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* h264parse = gst_bin_get_by_name(GST_BIN(self->pipeline_), "parser0");
GstPad* srcpad = gst_element_get_static_pad(h264parse, "src");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM, (GstPadProbeCallback)probe_parser_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* rtph264pay = gst_bin_get_by_name(GST_BIN(self->pipeline_), "pay0");
GstPad* srcpad = gst_element_get_static_pad(rtph264pay, "src");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_payloader_cb, self, nullptr);
auto ok2 = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM, (GstPadProbeCallback)probe_payloader_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* rtph264pay = gst_bin_get_by_name(GST_BIN(self->pipeline_), "pay0");
GstPad* srcpad = gst_element_get_static_pad(rtph264pay, "sink");
auto ok = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_BUFFER, (GstPadProbeCallback)probe_payloader_cb, self, nullptr);
auto ok2 = gst_pad_add_probe(srcpad, GST_PAD_PROBE_TYPE_DATA_DOWNSTREAM, (GstPadProbeCallback)probe_payloader_cb, self, nullptr);
gst_object_unref(srcpad);
}
{
GstElement* rtph264pay = gst_bin_get_by_name(GST_BIN(self->pipeline_), "pay0");
GstPad* pay_sink = gst_element_get_static_pad(rtph264pay, "sink");
GstCaps* allowed_caps = gst_pad_query_caps(pay_sink, nullptr);
gchar* allowed_str = gst_caps_to_string(allowed_caps);
printf("rtph264pay accepts %s\r\n", allowed_str);
g_free(allowed_str);
gst_object_unref(allowed_caps);
}
g_signal_connect(self->appsrc_, "need-data", G_CALLBACK(need_data_cb), self);
g_signal_connect(self->appsrc_, "enough-data", G_CALLBACK(enough_data_cb), self);
//GstStateChangeReturn ret = gst_element_set_state(self->pipeline_, GST_STATE_PLAYING);
//gst_rtsp_media_prepare(media, nullptr);
gst_object_unref(element);
}
static GstPadProbeReturn probe_appsrc_cb(GstPad *pad, GstPadProbeInfo *info, gpointer user_data) {
GstPadDirection dir = gst_pad_get_direction(pad);
printf("probe_appsrc_cb called pad[%s:%s]\r\n", GST_PAD_NAME(pad),
(dir == GST_PAD_SRC) ? "src" : (dir == GST_PAD_SINK) ? "sink" : "?");
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
return GST_PAD_PROBE_OK;
}//end probe_appsrc_cb()
static GstPadProbeReturn probe_parser_cb(GstPad *pad, GstPadProbeInfo *info, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
//if (GST_PAD_PROBE_INFO_TYPE(info) & GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM)
{
GstEvent *event = GST_PAD_PROBE_INFO_EVENT(info);
if (GST_EVENT_TYPE(event) == GST_EVENT_CAPS) {
GstCaps *caps;
gst_event_parse_caps(event, &caps);
gchar *caps_str = gst_caps_to_string(caps);
g_print("CAPS NEGOTIATED: %s\n", caps_str);
g_free(caps_str);
}
}
{
GstPadDirection dir = gst_pad_get_direction(pad);
printf("probe_parser_cb called pad[%s:%s]\r\n", GST_PAD_NAME(pad),
(dir == GST_PAD_SRC) ? "src" : (dir == GST_PAD_SINK) ? "sink" : "?");
}
return GST_PAD_PROBE_OK;
}//end probe_parser_cb()
static GstPadProbeReturn probe_payloader_cb(GstPad *pad, GstPadProbeInfo *info, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
{
GstElement* h264parse = gst_bin_get_by_name(GST_BIN(self->pipeline_), "parser0");
GstElement* rtph264pay = gst_bin_get_by_name(GST_BIN(self->pipeline_), "pay0");
{
GstEvent *event = GST_PAD_PROBE_INFO_EVENT(info);
if (GST_EVENT_TYPE(event) == GST_EVENT_CAPS) {
GstCaps *caps;
gst_event_parse_caps(event, &caps);
gchar *caps_str = gst_caps_to_string(caps);
g_print("CAPS NEGOTIATED: %s\n", caps_str);
g_free(caps_str);
}
}
{
GstPadDirection dir = gst_pad_get_direction(pad);
printf("probe_payloader_cb called pad[%s:%s]\r\n", GST_PAD_NAME(pad),
(dir == GST_PAD_SRC) ? "src" : (dir == GST_PAD_SINK) ? "sink" : "?");
}
}
return GST_PAD_PROBE_OK;
}//end probe_payloader_cb()
static void media_state_changed_cb(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data) {
GstState old_state, new_state;
GstElement* element = gst_rtsp_media_get_element(media);
gst_element_get_state(element, &old_state, &new_state, GST_CLOCK_TIME_NONE);
printf("Pipeline state changed from %s to %s\n", gst_element_state_get_name(old_state), gst_element_state_get_name(new_state));
gst_object_unref(element);
};
static void media_prepared_cb(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data) {
printf("media-prepared called\n");
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
GstElement *element = gst_rtsp_media_get_element(media);
/*const guint n_streams = gst_rtsp_media_n_streams(media);
for(guint i = 0 ; i<n_streams ; ++i)
{
GstRTSPStream* stream = gst_rtsp_media_get_stream(media, i);
if (stream != nullptr)
{
}
}*/
gst_element_set_state(element, GST_STATE_PLAYING);
gst_object_unref(element);
}
static void media_unprepared_cb(GstRTSPMedia *media, gpointer user_data) {
printf("media-unprepared called\n");
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
if (self != nullptr)
{
--self->activeClientsCount;
/*if (self->appsrc_ != 0)
{
g_object_unref(self->appsrc_);
self->appsrc_ = 0;
}*/
}
}
static void media_target_state_cb(GstRTSPMediaFactory *factory, GstRTSPMedia *media, gpointer user_data) {
printf("media-target-state called\n");
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
}
static void need_data_cb(GstAppSrc *src, guint length, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
self->enoughDataFlag = false;
self->push_buffers_from_queue();
}
static void enough_data_cb(GstAppSrc *src, gpointer user_data) {
_EXRTSPStreamerGStreamer *self = static_cast<_EXRTSPStreamerGStreamer *>(user_data);
self->enoughDataFlag = true;
}
};
int main(int argc, char* argv[])
{
if (argc <= 1)
{
std::cerr << "syntax : " << argv[0] << " <file of raw h264 samples>" << std::endl;
return -1;
}
const std::filesystem::path inputFilePath(argv[1]);
FILE* fp = fopen(inputFilePath.string().c_str(), "rb");
if (!fp)
{
std::cerr << "cannot open <" << inputFilePath << ">" << std::endl;
return -2;
}
fseek(fp, 0, SEEK_END);
const long fileSize = ftell(fp);
fseek(fp, 0, SEEK_SET);
std::vector<unsigned char> inputBytes(fileSize);
inputBytes.resize(fread(inputBytes.data(), sizeof(unsigned char), inputBytes.size(), fp));
fclose(fp);
if (inputBytes.empty())
{
std::cerr << "empty <" << inputFilePath << ">" << std::endl;
return -3;
}
const std::filesystem::path gstPluginPath = std::filesystem::path(argv[0]).parent_path();
g_setenv("GST_PLUGIN_PATH", gstPluginPath.string().c_str(), FALSE);
g_setenv("GST_DEBUG", "*:3", TRUE);
g_setenv("GOBJECT_DEBUG", "objects", TRUE);
g_setenv("GST_DEBUG_DUMP_DOT_DIR", std::filesystem::current_path().string().c_str(), TRUE);
gst_init(nullptr, nullptr);
gboolean ginitialized = gst_is_initialized();
_EXRTSPStreamerGStreamer server(8554, "test");
server.start();
constexpr const size_t nalu_startCodeLength = 4;
const unsigned char nalu_startCode[nalu_startCodeLength] = {0x00, 0x00, 0x00, 0x01};
const unsigned char* begin = inputBytes.data();
const unsigned char* end = begin+inputBytes.size();
const unsigned char* cur = std::search(begin, end, nalu_startCode, nalu_startCode+nalu_startCodeLength);
while(cur != end)
{
const unsigned char* next = std::search(cur+nalu_startCodeLength, end, nalu_startCode, nalu_startCode+nalu_startCodeLength);
const unsigned char* nalu = cur;
const size_t naluLength = next-cur;
print_nalu_type(nalu, naluLength);
server.pushData(nalu, naluLength);
std::this_thread::sleep_for(std::chrono::milliseconds(1000/30));
cur = next;
}
server.stop();
return 0;
}