554 lines
17 KiB
C++
554 lines
17 KiB
C++
/*
|
|
*******************************************************************************
|
|
* INCLUDE
|
|
*******************************************************************************
|
|
*/
|
|
#include "RTMPDemuxHandler.h"
|
|
#include "Logger.h"
|
|
#include "../core/System.h"
|
|
#include "util/Math.h"
|
|
#include <list>
|
|
|
|
namespace Video
|
|
{
|
|
namespace MediaServer
|
|
{
|
|
/*
|
|
*******************************************************************************
|
|
* CLASS DEFINE
|
|
*******************************************************************************
|
|
*/
|
|
RTMPDemuxHandler::RTMPDemuxHandler(const std::string& uri)
|
|
: Core::DemuxHandler(Poco::format("MediaServer::DemuxHandler(%s)", uri), uri)
|
|
{
|
|
}
|
|
|
|
RTMPDemuxHandler::~RTMPDemuxHandler()
|
|
{
|
|
}
|
|
|
|
void RTMPDemuxHandler::run(Core::VideoEncoder::Ptr pVideoEncoder, Core::AudioEncoder::Ptr pAudioEncoder,
|
|
const Post& post, std::atomic<bool>& isCancelled)
|
|
{
|
|
_pVideoEncoder = pVideoEncoder;
|
|
_pAudioEncoder = pAudioEncoder;
|
|
_post = post;
|
|
|
|
Poco::URI uriParser(uri());
|
|
std::string scheme = uriParser.getScheme();
|
|
std::string host = uriParser.getHost();
|
|
short port = uriParser.getPort();
|
|
port = port <= 0 ? 1935 : port;
|
|
std::string path = uriParser.getPath();
|
|
|
|
if (!path.empty() && path[0] == '/')
|
|
path = path.substr(1);
|
|
|
|
size_t pos = path.find_first_of('/');
|
|
if (pos == std::string::npos)
|
|
{
|
|
logger().error("%s: Invalid URL: %s", name(), uri());
|
|
Poco::Thread::sleep(2000);
|
|
return;
|
|
}
|
|
|
|
std::string turl = scheme + "://" + host + ":" + std::to_string(port) + "/" + path.substr(0, pos);
|
|
std::string appname = path.substr(0, pos);
|
|
std::string playpath = path.substr(pos + 1);
|
|
|
|
struct rtmp_client_handler_t handler;
|
|
handler.send = RTMPDemuxHandler::clientSend;
|
|
handler.onaudio = RTMPDemuxHandler::clientOnAudio;
|
|
handler.onvideo = RTMPDemuxHandler::clientOnVideo;
|
|
handler.onscript = RTMPDemuxHandler::clientOnScript;
|
|
|
|
const size_t buffsz = 2 * 1024 * 1024;
|
|
Poco::SharedPtr<uint8_t> pBuffer = new uint8_t[buffsz];
|
|
int rc;
|
|
while (!isCancelled)
|
|
{
|
|
_audioInfoParsed = false;
|
|
_videoInfoParsed = false;
|
|
|
|
while (!isCancelled)
|
|
{
|
|
_socket = socket_connect_host(host.c_str(), port, 2000);
|
|
if (_socket != socket_invalid)
|
|
{
|
|
socket_setnonblock(_socket, 0);
|
|
break;
|
|
}
|
|
|
|
Poco::Thread::sleep(2000);
|
|
continue;
|
|
}
|
|
|
|
_pCient = rtmp_client_create(appname.c_str(), playpath.c_str(), turl.c_str(), this, &handler);
|
|
if (!_pCient)
|
|
{
|
|
logger().error("%s: Create RTMP-Client FAIL!(URL: %s)", name(), uri());
|
|
Poco::Thread::sleep(2000);
|
|
goto rtmp_create_fail;
|
|
}
|
|
|
|
_pDemuxer = flv_demuxer_create(RTMPDemuxHandler::flvDemuxerHandler, this);
|
|
if (!_pDemuxer)
|
|
{
|
|
logger().error("%s: Create FLV Debuxer FAIL!(URL: %s)", name(), uri());
|
|
Poco::Thread::sleep(2000);
|
|
goto demuxer_fail;
|
|
}
|
|
|
|
rc = rtmp_client_start(_pCient, 1);
|
|
if (rc != 0)
|
|
{
|
|
logger().error("%s: Start RTMP-Client FAIL!(URL: %s)", name(), uri());
|
|
Poco::Thread::sleep(2000);
|
|
goto rmtp_start_fail;
|
|
}
|
|
|
|
while (!isCancelled)
|
|
{
|
|
rc = socket_recv(_socket, pBuffer.get(), buffsz, 0);
|
|
if (rc <= 0)
|
|
{
|
|
logger().error("%s: Socket recv FAIL: %d(URL: %s)", name(), rc, uri());
|
|
break;
|
|
}
|
|
|
|
rc = rtmp_client_input(_pCient, pBuffer.get(), rc);
|
|
if (rc != 0)
|
|
{
|
|
logger().error("%s: RTMP client input FAIL: %d(URL: %s)", name(), rc, uri());
|
|
break;
|
|
}
|
|
}
|
|
|
|
rtmp_client_stop(_pCient);
|
|
|
|
rmtp_start_fail:
|
|
flv_demuxer_destroy(_pDemuxer);
|
|
|
|
demuxer_fail:
|
|
rtmp_client_destroy(_pCient);
|
|
|
|
rtmp_create_fail:
|
|
socket_close(_socket);
|
|
_socket = socket_invalid;
|
|
Poco::Thread::sleep(2000);
|
|
}
|
|
}
|
|
|
|
int RTMPDemuxHandler::parseFLVMetaData(const uint8_t* data, size_t bytes)
|
|
{
|
|
double audiocodecid = 0;
|
|
double audiodatarate = 0;
|
|
double audiodelay = 0;
|
|
double audiosamplerate = 0;
|
|
double audiosamplesize = 0;
|
|
int stereo = 0;
|
|
double videocodecid = 0;
|
|
double videodatarate = 0;
|
|
double framerate = 0;
|
|
double height = 0;
|
|
double width = 0;
|
|
|
|
struct amf_object_item_t prop[16];
|
|
struct amf_object_item_t items[1];
|
|
char buffer[64];
|
|
|
|
#define AMF_OBJECT_ITEM_VALUE(v, amf_type, amf_name, amf_value, amf_size) \
|
|
{ \
|
|
v.type = amf_type; \
|
|
v.name = amf_name; \
|
|
v.value = amf_value; \
|
|
v.size = amf_size; \
|
|
}
|
|
int i = 0;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiocodecid", &audiocodecid, sizeof(audiocodecid));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiodatarate", &audiodatarate, sizeof(audiodatarate));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiodelay", &audiodelay, sizeof(audiodelay));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiosamplerate", &audiosamplerate, sizeof(audiosamplerate));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "audiosamplesize", &audiosamplesize, sizeof(audiosamplesize));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_BOOLEAN, "stereo", &stereo, sizeof(stereo));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "videocodecid", &videocodecid, sizeof(videocodecid));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "videodatarate", &videodatarate, sizeof(videodatarate));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "framerate", &framerate, sizeof(framerate));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "height", &height, sizeof(height));
|
|
i++;
|
|
AMF_OBJECT_ITEM_VALUE(prop[i], AMF_NUMBER, "width", &width, sizeof(width));
|
|
i++;
|
|
|
|
AMF_OBJECT_ITEM_VALUE(items[0], AMF_OBJECT, "onMetaData", prop, i);
|
|
|
|
const uint8_t* end = data + bytes;
|
|
if (AMF_STRING != data[0] || NULL == (data = AMFReadString(data + 1, end, 0, buffer, sizeof(buffer) - 1)))
|
|
return -1;
|
|
|
|
if (0 == strcmp(buffer, "@setDataFrame"))
|
|
{
|
|
if (AMF_STRING != data[0] || NULL == (data = AMFReadString(data + 1, end, 0, buffer, sizeof(buffer) - 1)))
|
|
return -1;
|
|
}
|
|
|
|
if (0 != strcmp(buffer, "onMetaData"))
|
|
return 0;
|
|
|
|
if (amf_read_items(data, end, items, sizeof(items) / sizeof(items[0])))
|
|
{
|
|
if (framerate > 0)
|
|
_pVideoEncoder->setFrameRate(framerate > 60 ? 60 : framerate);
|
|
|
|
if (videodatarate > 0)
|
|
_pVideoEncoder->setBitrate((int64_t)videodatarate);
|
|
|
|
if (audiodatarate > 0)
|
|
_pAudioEncoder->setBitrate((int64_t)audiodatarate);
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
inline const uint8_t* RTMPDemuxHandler::annexBStartcode(const uint8_t* data, size_t bytes)
|
|
{
|
|
size_t i;
|
|
for (i = 2; i + 1 < bytes; i++)
|
|
{
|
|
if (0x01 == data[i] && 0x00 == data[i - 1] && 0x00 == data[i - 2])
|
|
return data + i + 1;
|
|
}
|
|
|
|
return nullptr;
|
|
}
|
|
|
|
bool RTMPDemuxHandler::isH264BFrame(const void* h264, size_t bytes)
|
|
{
|
|
ptrdiff_t n;
|
|
const uint8_t *p, *next, *end;
|
|
|
|
end = (const uint8_t*)h264 + bytes;
|
|
p = annexBStartcode((const uint8_t*)h264, bytes);
|
|
|
|
while (p)
|
|
{
|
|
next = annexBStartcode(p, end - p);
|
|
if (next)
|
|
n = next - p - 3;
|
|
else
|
|
n = end - p;
|
|
|
|
while (n > 0 && 0 == p[n - 1])
|
|
n--;
|
|
|
|
assert(n > 0);
|
|
if (n > 0)
|
|
{
|
|
if ((p[0] & 0x1f) == 1)
|
|
{
|
|
bitstream_t stream;
|
|
bitstream_init(&stream, (const uint8_t*)(p + 1), n - 1);
|
|
bitstream_read_ue(&stream);
|
|
int32_t sliceType = bitstream_read_ue(&stream);
|
|
if (sliceType == 1)
|
|
return true;
|
|
}
|
|
}
|
|
p = next;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processH264Data(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
if (!_videoInfoParsed)
|
|
mpeg4_h264_annexb_nalu(data, bytes, RTMPDemuxHandler::h264NALhandler, this);
|
|
|
|
enqueueVideoPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processH265Data(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
if (!_videoInfoParsed)
|
|
mpeg4_h264_annexb_nalu(data, bytes, RTMPDemuxHandler::h265NALhandler, this);
|
|
|
|
enqueueVideoPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processASCData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
struct mpeg4_aac_t aac;
|
|
if (mpeg4_aac_audio_specific_config_load((const uint8_t*)data, bytes, &aac) >= 0)
|
|
{
|
|
_pAudioEncoder->setCodecID(Core::CodecID::AAC);
|
|
_pAudioEncoder->setFrameSize(1024);
|
|
_pAudioEncoder->setSampleFormat(Core::SampleFormat::FLTP);
|
|
_pAudioEncoder->setSampleRate((int)aac.sampling_frequency);
|
|
_pAudioEncoder->setChannelLayout(aac.channels == 2 ? Core::ChannelLayout::STEREO : Core::ChannelLayout::MONO);
|
|
_pAudioEncoder->commit();
|
|
logger().information("%s: Audio found: AAC %dHz %s", name(), (int)aac.sampling_frequency,
|
|
Core::channelLayoutToString(_pAudioEncoder->getChannelLayout()));
|
|
;
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processAACData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
enqueueAudioPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processOpusHeadData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
struct opus_head_t opus;
|
|
if (opus_head_load((const uint8_t*)data, bytes, &opus) >= 0)
|
|
{
|
|
_pAudioEncoder->setCodecID(Core::CodecID::OPUS);
|
|
_pAudioEncoder->setFrameSize(960);
|
|
_pAudioEncoder->setSampleFormat(Core::SampleFormat::S16);
|
|
_pAudioEncoder->setSampleRate((int)opus.input_sample_rate);
|
|
_pAudioEncoder->setChannelLayout(opus_head_channels(&opus) == 2 ? Core::ChannelLayout::STEREO
|
|
: Core::ChannelLayout::MONO);
|
|
_pAudioEncoder->commit();
|
|
logger().information("%s: Audio found: OPUS %dHz %s", name(), (int)opus.input_sample_rate,
|
|
Core::channelLayoutToString(_pAudioEncoder->getChannelLayout()));
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processOpusData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
enqueueAudioPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processG711AData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
if (!_audioInfoParsed)
|
|
{
|
|
_pAudioEncoder->setCodecID(Core::CodecID::G711A);
|
|
_pAudioEncoder->setFrameSize(1024);
|
|
_pAudioEncoder->setSampleFormat(Core::SampleFormat::S16P);
|
|
_pAudioEncoder->setSampleRate((int)8000);
|
|
_pAudioEncoder->setChannelLayout(Core::ChannelLayout::MONO);
|
|
_pAudioEncoder->commit();
|
|
logger().information("%s: Audio found: G711A %dHz %s", name(), 8000, std::string("MONO"));
|
|
|
|
_audioInfoParsed = true;
|
|
}
|
|
|
|
enqueueAudioPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
int RTMPDemuxHandler::processG711UData(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
if (!_audioInfoParsed)
|
|
{
|
|
_pAudioEncoder->setCodecID(Core::CodecID::G711U);
|
|
_pAudioEncoder->setFrameSize(1024);
|
|
_pAudioEncoder->setSampleFormat(Core::SampleFormat::S16P);
|
|
_pAudioEncoder->setSampleRate((int)8000);
|
|
_pAudioEncoder->setChannelLayout(Core::ChannelLayout::MONO);
|
|
_pAudioEncoder->commit();
|
|
logger().information("%s: Audio found: G711U %dHz %s", name(), 8000, std::string("MONO"));
|
|
|
|
_audioInfoParsed = true;
|
|
}
|
|
|
|
enqueueAudioPacket(data, bytes, pts, dts, flags);
|
|
return 0;
|
|
}
|
|
|
|
void RTMPDemuxHandler::enqueueVideoPacket(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
uint8_t* pBuffer = new uint8_t[bytes];
|
|
memcpy(pBuffer, data, bytes);
|
|
|
|
uint64_t localPTS = pts;
|
|
uint64_t localDTS = dts;
|
|
|
|
adjustTimestamp(localPTS, localDTS);
|
|
updateTimeline(localPTS);
|
|
|
|
_post(new Core::Packet(_pVideoEncoder->code(),
|
|
Core::Packet::FLAG_IS_VIDEO | (flags ? Core::Packet::FLAG_IS_KEY : 0), pBuffer, bytes,
|
|
localPTS, localDTS),
|
|
Core::MediaType::VIDEO);
|
|
}
|
|
|
|
void RTMPDemuxHandler::enqueueAudioPacket(const void* data, size_t bytes, uint32_t pts, uint32_t dts, int flags)
|
|
{
|
|
uint8_t* pBuffer = new uint8_t[bytes];
|
|
memcpy(pBuffer, data, bytes);
|
|
|
|
uint64_t localPTS = pts;
|
|
uint64_t localDTS = dts;
|
|
|
|
adjustTimestamp(localPTS, localDTS);
|
|
updateTimeline(localPTS);
|
|
_post(new Core::Packet(_pAudioEncoder->code(), flags, pBuffer, bytes, localPTS, localDTS), Core::MediaType::AUDIO);
|
|
}
|
|
|
|
int RTMPDemuxHandler::clientSend(void* param, const void* header, size_t len, const void* data, size_t bytes)
|
|
{
|
|
socket_bufvec_t vec[2];
|
|
socket_setbufvec(vec, 0, (void*)header, len);
|
|
socket_setbufvec(vec, 1, (void*)data, bytes);
|
|
return socket_send_v_all_by_time(((RTMPDemuxHandler*)param)->_socket, vec, bytes ? 2 : 1, 0, 2000);
|
|
}
|
|
|
|
int RTMPDemuxHandler::clientOnAudio(void* param, const void* data, size_t bytes, uint32_t timestamp)
|
|
{
|
|
return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_AUDIO, data, bytes, timestamp);
|
|
}
|
|
|
|
int RTMPDemuxHandler::clientOnVideo(void* param, const void* data, size_t bytes, uint32_t timestamp)
|
|
{
|
|
return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_VIDEO, data, bytes, timestamp);
|
|
}
|
|
|
|
int RTMPDemuxHandler::clientOnScript(void* param, const void* data, size_t bytes, uint32_t timestamp)
|
|
{
|
|
return flv_demuxer_input(((RTMPDemuxHandler*)param)->_pDemuxer, FLV_TYPE_SCRIPT, data, bytes, timestamp);
|
|
}
|
|
|
|
int RTMPDemuxHandler::flvDemuxerHandler(void* param, int codec, const void* data, size_t bytes, uint32_t pts,
|
|
uint32_t dts, int flags)
|
|
{
|
|
int rc;
|
|
RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;
|
|
|
|
switch (codec)
|
|
{
|
|
case FLV_SCRIPT_METADATA:
|
|
rc = pHandler->parseFLVMetaData((const uint8_t*)data, bytes);
|
|
break;
|
|
|
|
case FLV_VIDEO_AVCC:
|
|
break;
|
|
|
|
case FLV_VIDEO_H264:
|
|
rc = pHandler->processH264Data(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_VIDEO_HVCC:
|
|
break;
|
|
|
|
case FLV_VIDEO_H265:
|
|
rc = pHandler->processH265Data(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_ASC:
|
|
rc = pHandler->processASCData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_AAC:
|
|
rc = pHandler->processAACData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_OPUS_HEAD:
|
|
rc = pHandler->processOpusHeadData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_OPUS:
|
|
rc = pHandler->processOpusData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_G711A:
|
|
rc = pHandler->processG711AData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
case FLV_AUDIO_G711U:
|
|
rc = pHandler->processG711UData(data, bytes, pts, dts, flags);
|
|
break;
|
|
|
|
default:
|
|
logger().warning("%s: UNIMPLEMENTED FLV-CODEC type: %d(%#x)", pHandler->name(), codec, (uint32_t)codec);
|
|
break;
|
|
}
|
|
|
|
(void)rc;
|
|
return 0;
|
|
}
|
|
|
|
void RTMPDemuxHandler::h264NALhandler(void* param, const uint8_t* nalu, size_t bytes)
|
|
{
|
|
RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;
|
|
|
|
struct h264_sps_t sps;
|
|
if (h264_sps_parse(nalu, bytes, &sps) == 0)
|
|
{
|
|
int x, y, width, height;
|
|
|
|
h264_display_rect(&sps, &x, &y, &width, &height);
|
|
|
|
auto pEncoder = pHandler->_pVideoEncoder;
|
|
pEncoder->setCodecID(Core::CodecID::H264);
|
|
pEncoder->setWidth(width);
|
|
pEncoder->setHeight(height);
|
|
pEncoder->commit();
|
|
|
|
pHandler->_videoInfoParsed = true;
|
|
|
|
logger().information("%s: Video found: H.264 %dx%d", pHandler->name(), width, height);
|
|
}
|
|
}
|
|
|
|
void RTMPDemuxHandler::h265NALhandler(void* param, const uint8_t* nalu, size_t bytes)
|
|
{
|
|
RTMPDemuxHandler* pHandler = (RTMPDemuxHandler*)param;
|
|
|
|
struct h265_sps_t sps;
|
|
if (h265_sps_parse(nalu, bytes, &sps) == 0)
|
|
{
|
|
int x, y, width, height;
|
|
|
|
h265_display_rect(&sps, &x, &y, &width, &height);
|
|
|
|
auto pEncoder = pHandler->_pVideoEncoder;
|
|
pEncoder->setCodecID(Core::CodecID::H265);
|
|
pEncoder->setWidth(width);
|
|
pEncoder->setHeight(height);
|
|
pEncoder->commit();
|
|
|
|
pHandler->_videoInfoParsed = true;
|
|
|
|
logger().information("%s: Video found: H.265 %dx%d", pHandler->name(), width, height);
|
|
}
|
|
}
|
|
|
|
void RTMPDemuxHandler::adjustTimestamp(uint64_t& pts, uint64_t& dts)
|
|
{
|
|
pts *= 1000ull;
|
|
dts *= 1000ull;
|
|
|
|
auto drift = getTimeDrift(pts);
|
|
pts += drift;
|
|
dts += drift;
|
|
}
|
|
|
|
void RTMPDemuxHandler::updateTimeline(uint64_t ts)
|
|
{
|
|
if (getTimeline() < ts)
|
|
setTimeline(ts);
|
|
}
|
|
|
|
void RTMPDemuxHandler::seek(uint64_t ts)
|
|
{
|
|
}
|
|
}
|
|
}
|