amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-10-06 20:23:43 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

整理flv复用器

Browse Source
This commit is contained in:
ziyue 2021-07-15 11:16:11 +08:00
parent 0d48275115
commit fe42ea30fc
3 changed files with 108 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

151
src/Rtmp/FlvMuxer.cpp
View File

@ -16,23 +16,17 @@
namespace mediakit {
FlvMuxer::FlvMuxer() {
}
FlvMuxer::~FlvMuxer() {
}
void FlvMuxer::start(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &media) {
if(!media){
void FlvMuxer::start(const EventPoller::Ptr &poller, const RtmpMediaSource::Ptr &media) {
if (!media) {
throw std::runtime_error("RtmpMediaSource 无效");
}
if(!poller->isCurrentThread()){
if (!poller->isCurrentThread()) {
weak_ptr<FlvMuxer> weakSelf = getSharedPtr();
//延时两秒启动录制，目的是为了等待config帧收集完毕
poller->doDelayTask(2000,[weakSelf,poller,media](){
poller->doDelayTask(2000, [weakSelf, poller, media]() {
auto strongSelf = weakSelf.lock();
if(strongSelf){
strongSelf->start(poller,media);
if (strongSelf) {
strongSelf->start(poller, media);
}
return 0;
});
@ -43,9 +37,9 @@ void FlvMuxer::start(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &
std::weak_ptr<FlvMuxer> weakSelf = getSharedPtr();
_ring_reader = media->getRing()->attach(poller);
_ring_reader->setDetachCB([weakSelf](){
_ring_reader->setDetachCB([weakSelf]() {
auto strongSelf = weakSelf.lock();
if(!strongSelf){
if (!strongSelf) {
return;
}
strongSelf->onDetach();
@ -53,15 +47,15 @@ void FlvMuxer::start(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &
//音频同步于视频
_stamp[0].syncTo(_stamp[1]);
_ring_reader->setReadCB([weakSelf](const RtmpMediaSource::RingDataType &pkt){
_ring_reader->setReadCB([weakSelf](const RtmpMediaSource::RingDataType &pkt) {
auto strongSelf = weakSelf.lock();
if(!strongSelf){
if (!strongSelf) {
return;
}
size_t i = 0;
auto size = pkt->size();
pkt->for_each([&](const RtmpPacket::Ptr &rtmp){
pkt->for_each([&](const RtmpPacket::Ptr &rtmp) {
strongSelf->onWriteRtmp(rtmp, ++i == size);
});
});
@ -75,38 +69,36 @@ BufferRaw::Ptr FlvMuxer::obtainBuffer(const void *data, size_t len) {
void FlvMuxer::onWriteFlvHeader(const RtmpMediaSource::Ptr &mediaSrc) {
//发送flv文件头
char flv_file_header[] = "FLV\x1\x5\x0\x0\x0\x9"; // have audio and have video
bool is_have_audio = false,is_have_video = false;
auto buffer = BufferRaw::create();
buffer->setCapacity(sizeof(FLVHeader));
buffer->setSize(sizeof(FLVHeader));
mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){
if(pkt->type_id == MSG_VIDEO){
is_have_video = true;
FLVHeader *header = (FLVHeader *) buffer->data();
memset(header, 0, sizeof(FLVHeader));
header->flv[0] = 'F';
header->flv[1] = 'L';
header->flv[2] = 'V';
header->version = 1;
header->length = htonl(9);
mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt) {
if (pkt->type_id == MSG_VIDEO) {
header->have_video = 1;
}
if(pkt->type_id == MSG_AUDIO){
is_have_audio = true;
if (pkt->type_id == MSG_AUDIO) {
header->have_audio = 1;
}
});
if (is_have_audio && is_have_video) {
flv_file_header[4] = 0x05;
} else if (is_have_audio && !is_have_video) {
flv_file_header[4] = 0x04;
} else if (!is_have_audio && is_have_video) {
flv_file_header[4] = 0x01;
} else {
flv_file_header[4] = 0x00;
}
//flv header
onWrite(obtainBuffer(flv_file_header, sizeof(flv_file_header) - 1), false);
onWrite(buffer, false);
auto size = htonl(0);
//PreviousTagSize0 Always 0
onWrite(obtainBuffer((char *)&size,4), false);
auto size = htonl(0);
onWrite(obtainBuffer((char *) &size, 4), false);
auto &metadata = mediaSrc->getMetaData();
if(metadata){
if (metadata) {
//在有metadata的情况下才发送metadata
//其实metadata没什么用，有些推流器不产生metadata
AMFEncoder invoke;
@ -115,96 +107,82 @@ void FlvMuxer::onWriteFlvHeader(const RtmpMediaSource::Ptr &mediaSrc) {
}
//config frame
mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt){
mediaSrc->getConfigFrame([&](const RtmpPacket::Ptr &pkt) {
onWriteRtmp(pkt, true);
});
}
#if defined(_WIN32)
#pragma pack(push, 1)
#endif // defined(_WIN32)
class RtmpTagHeader {
public:
uint8_t type = 0;
uint8_t data_size[3] = {0};
uint8_t timestamp[3] = {0};
uint8_t timestamp_ex = 0;
uint8_t streamid[3] = {0}; /* Always 0. */
}PACKED;
#if defined(_WIN32)
#pragma pack(pop)
#endif // defined(_WIN32)
void FlvMuxer::onWriteFlvTag(const RtmpPacket::Ptr &pkt, uint32_t time_stamp , bool flush) {
void FlvMuxer::onWriteFlvTag(const RtmpPacket::Ptr &pkt, uint32_t time_stamp, bool flush) {
onWriteFlvTag(pkt->type_id, pkt, time_stamp, flush);
}
void FlvMuxer::onWriteFlvTag(uint8_t type, const Buffer::Ptr &buffer, uint32_t time_stamp, bool flush) {
RtmpTagHeader header;
header.type = type;
set_be24(header.data_size, (uint32_t)buffer->size());
header.timestamp_ex = (uint8_t) ((time_stamp >> 24) & 0xff);
set_be24(header.data_size, (uint32_t) buffer->size());
header.timestamp_ex = (time_stamp >> 24) & 0xff;
set_be24(header.timestamp, time_stamp & 0xFFFFFF);
//tag header
onWrite(obtainBuffer((char *)&header, sizeof(header)), false);
onWrite(obtainBuffer((char *) &header, sizeof(header)), false);
//tag data
onWrite(buffer, false);
uint32_t size = htonl((uint32_t)(buffer->size() + sizeof(header)));
//PreviousTagSize
onWrite(obtainBuffer((char *)&size,4), flush);
uint32_t size = htonl((uint32_t) (buffer->size() + sizeof(header)));
onWrite(obtainBuffer((char *) &size, 4), flush);
}
void FlvMuxer::onWriteRtmp(const RtmpPacket::Ptr &pkt,bool flush) {
void FlvMuxer::onWriteRtmp(const RtmpPacket::Ptr &pkt, bool flush) {
int64_t dts_out;
_stamp[pkt->type_id % 2].revise(pkt->time_stamp, 0, dts_out, dts_out);
onWriteFlvTag(pkt, (uint32_t)dts_out,flush);
onWriteFlvTag(pkt, (uint32_t) dts_out, flush);
}
void FlvMuxer::stop() {
if(_ring_reader){
if (_ring_reader) {
_ring_reader.reset();
onDetach();
}
}
///////////////////////////////////////////////////////FlvRecorder/////////////////////////////////////////////////////
void FlvRecorder::startRecord(const EventPoller::Ptr &poller,const string &vhost, const string &app, const string &stream,const string &file_path) {
startRecord(poller,dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA,vhost,app,stream)),file_path);
void FlvRecorder::startRecord(const EventPoller::Ptr &poller, const string &vhost, const string &app, const string &stream, const string &file_path) {
startRecord(poller, dynamic_pointer_cast<RtmpMediaSource>(MediaSource::find(RTMP_SCHEMA, vhost, app, stream)), file_path);
}
void FlvRecorder::startRecord(const EventPoller::Ptr &poller,const RtmpMediaSource::Ptr &media, const string &file_path) {
void FlvRecorder::startRecord(const EventPoller::Ptr &poller, const RtmpMediaSource::Ptr &media,
const string &file_path) {
stop();
lock_guard<recursive_mutex> lck(_file_mtx);
//开辟文件写缓存
std::shared_ptr<char> fileBuf(new char[FILE_BUF_SIZE],[](char *ptr){
if(ptr){
delete [] ptr;
std::shared_ptr<char> fileBuf(new char[FILE_BUF_SIZE], [](char *ptr) {
if (ptr) {
delete[] ptr;
}
});
//新建文件
_file.reset(File::create_file(file_path.data(), "wb"), [fileBuf](FILE *fp){
if(fp){
_file.reset(File::create_file(file_path.data(), "wb"), [fileBuf](FILE *fp) {
if (fp) {
fflush(fp);
fclose(fp);
}
});
if (!_file){
throw std::runtime_error( StrPrinter << "打开文件失败:" << file_path);
if (!_file) {
throw std::runtime_error(StrPrinter << "打开文件失败:" << file_path);
}
//设置文件写缓存
setvbuf( _file.get(), fileBuf.get(),_IOFBF, FILE_BUF_SIZE);
start(poller,media);
setvbuf(_file.get(), fileBuf.get(), _IOFBF, FILE_BUF_SIZE);
start(poller, media);
}
void FlvRecorder::onWrite(const Buffer::Ptr &data, bool flush) {
lock_guard<recursive_mutex> lck(_file_mtx);
if(_file){
fwrite(data->data(),data->size(),1,_file.get());
if (_file) {
fwrite(data->data(), data->size(), 1, _file.get());
}
}
@ -214,14 +192,7 @@ void FlvRecorder::onDetach() {
}
std::shared_ptr<FlvMuxer> FlvRecorder::getSharedPtr() {
return shared_from_this();
return shared_from_this();
}
FlvRecorder::FlvRecorder() {
}
FlvRecorder::~FlvRecorder() {
}
}//namespace mediakit

17
src/Rtmp/FlvMuxer.h
View File

@ -19,11 +19,12 @@ using namespace toolkit;
namespace mediakit {
class FlvMuxer{
class FlvMuxer {
public:
typedef std::shared_ptr<FlvMuxer> Ptr;
FlvMuxer();
virtual ~FlvMuxer();
using Ptr = std::shared_ptr<FlvMuxer>;
FlvMuxer() = default;
virtual ~FlvMuxer() = default;
void stop();
protected:
@ -47,9 +48,10 @@ private:
class FlvRecorder : public FlvMuxer , public std::enable_shared_from_this<FlvRecorder>{
public:
typedef std::shared_ptr<FlvRecorder> Ptr;
FlvRecorder();
virtual ~FlvRecorder();
using Ptr = std::shared_ptr<FlvRecorder>;
FlvRecorder() = default;
~FlvRecorder() override = default;
void startRecord(const EventPoller::Ptr &poller, const RtmpMediaSource::Ptr &media, const string &file_path);
void startRecord(const EventPoller::Ptr &poller, const string &vhost, const string &app, const string &stream, const string &file_path);
@ -63,6 +65,5 @@ private:
recursive_mutex _file_mtx;
};
}//namespace mediakit
#endif //ZLMEDIAKIT_FLVMUXER_H

38
src/Rtmp/Rtmp.h
View File

@ -129,6 +129,44 @@ public:
uint8_t stream_index[4]; /* Note, this is little-endian while others are BE */
}PACKED;
class FLVHeader {
public:
//FLV
char flv[3];
//File version (for example, 0x01 for FLV version 1)
uint8_t version;
#if __BYTE_ORDER == __BIG_ENDIAN
//保留,置0
uint8_t : 5;
//是否有音频
uint8_t have_audio: 1;
//保留,置0
uint8_t : 1;
//是否有视频
uint8_t have_video: 1;
#else
//是否有视频
uint8_t have_video: 1;
//保留,置0
uint8_t : 1;
//是否有音频
uint8_t have_audio: 1;
//保留,置0
uint8_t : 5;
#endif
//The length of this header in bytes,固定为9
uint32_t length;
} PACKED;
class RtmpTagHeader {
public:
uint8_t type = 0;
uint8_t data_size[3] = {0};
uint8_t timestamp[3] = {0};
uint8_t timestamp_ex = 0;
uint8_t streamid[3] = {0}; /* Always 0. */
} PACKED;
#if defined(_WIN32)
#pragma pack(pop)
#endif // defined(_WIN32)