/* ******************************************************************************* * INCLUDE ******************************************************************************* */ #include "MPEGMuxer.h" #include "../core/VideoEncoder.h" #include "../core/AudioEncoder.h" #include "../core/SampleFormat.h" #include "../core/System.h" namespace Video { namespace MediaServer { /* ******************************************************************************* * TYPEDEFS ******************************************************************************* */ struct CodecIDItem { Core::CodecID id; int type; }; /* ******************************************************************************* * LOCAL VARIABLES ******************************************************************************* */ static const std::array codecMap = {{ {Core::CodecID::H264, PSI_STREAM_H264}, {Core::CodecID::H265, PSI_STREAM_H265}, {Core::CodecID::MJPEG, PSI_STREAM_JPEG_2000}, {Core::CodecID::MP3, PSI_STREAM_MP3}, {Core::CodecID::AAC, PSI_STREAM_AAC}, {Core::CodecID::OPUS, PSI_STREAM_AUDIO_OPUS}, {Core::CodecID::G711A, PSI_STREAM_AUDIO_G711A}, {Core::CodecID::G711U, PSI_STREAM_AUDIO_G711U}, }}; /* ******************************************************************************* * CLASS DEFINE ******************************************************************************* */ MPEGMuxer::Stream::Stream(const std::string& name, int id, Core::PacketQueue::Ptr pQueue) : _name(name), _id(id), _pQueue(pQueue) { } MPEGMuxer::Stream::~Stream() { } /* ******************************************************************************* * CLASS DEFINE ******************************************************************************* */ static int isMulticastAddr(const char* ip) { int r; struct addrinfo* ai; r = getaddrinfo(ip, 0, NULL, &ai); if (0 != r) return 0; if (AF_INET == ai->ai_addr->sa_family) { const struct sockaddr_in* in = (const struct sockaddr_in*)ai->ai_addr; assert(sizeof(struct sockaddr_in) == ai->ai_addrlen); r = (ntohl(in->sin_addr.s_addr) & 0xf0000000) == 0xe0000000 ? 1 : 0; } else if (AF_INET6 == ai->ai_addr->sa_family) { const struct sockaddr_in6* in6 = (const struct sockaddr_in6*)ai->ai_addr; assert(sizeof(struct sockaddr_in6) == ai->ai_addrlen); r = in6->sin6_addr.s6_addr[0] == 0xff ? 1 : 0; } else { assert(0); r = 0; } freeaddrinfo(ai); return r; } Video::MediaServer::MPEGMuxer::MPEGMuxer(const std::string& url) : Core::Muxer(Poco::format("MediaServer::MPEGMuxer(%s)", url), url) { } MPEGMuxer::~MPEGMuxer() { } void MPEGMuxer::connect() { Poco::URI uri(url()); std::string scheme = uri.getScheme(); std::string host = uri.getHost(); short port = uri.getPort(); _socket = socket_udp_bind_ipv4(nullptr, 50609); if (socket_invalid == _socket) throw Poco::InvalidAccessException(Poco::format("%s: Create socket FAIL", name())); if (isMulticastAddr(host.c_str())) socket_udp_multicast(_socket, host.c_str(), nullptr, 16); memset(&_destAddr, 0, sizeof(_destAddr)); socket_addr_from_ipv4(&_destAddr, host.c_str(), port); static mpeg_muxer_func_t func = {[](void* param, size_t bytes) -> void* { MPEGMuxer* pMuxer = (MPEGMuxer*)param; if (pMuxer->_writeIndex + bytes > sizeof(pMuxer->_buffer)) pMuxer->flush(); return &pMuxer->_buffer[pMuxer->_writeIndex]; }, [](void* param, void* packet) -> void {}, [](void* param, int stream, void* packet, size_t bytes) -> int { MPEGMuxer* pMuxer = (MPEGMuxer*)param; pMuxer->_writeIndex += bytes; return 0; }}; _pMuxer = mpeg_muxer_create(0, &func, this); if (!_pMuxer) throw Poco::InvalidAccessException(Poco::format("%s: Create MPEG-Muxer FAIL!", name())); // mpeg_muxer_add_program(_pMuxer, 1, "XXXX", 4); /* 多节目方式 */ previousComponents().foreach ( [&](Core::Component::Ptr pPrevious) -> void { auto pVideoEncoder = pPrevious.cast(); if (!pVideoEncoder.isNull()) { int id = mpeg_muxer_add_stream(_pMuxer, getCodecID(pVideoEncoder->getCodecID()), nullptr, 0); _streams.push_back(new Stream(Poco::format("%s/Stream(%s)", name(), pVideoEncoder->name()), id, packetQueues().find(pPrevious))); return; } auto pAudioEncoder = pPrevious.cast(); if (!pAudioEncoder.isNull()) { int id = mpeg_muxer_add_stream(_pMuxer, getCodecID(pAudioEncoder->getCodecID()), nullptr, 0); _streams.push_back(new Stream(Poco::format("%s/Stream(%s)", name(), pAudioEncoder->name()), id, packetQueues().find(pPrevious))); } }); } void MPEGMuxer::mux() { std::sort(_streams.begin(), _streams.end(), [](auto a, auto b) { return *a < *b; }); auto& pStream = *_streams.begin(); Core::Packet::Ptr pPacket = pStream->queue()->dequeue(4000); if (pStream->serialCodeWatcher().isChanged(pPacket->serialCode())) throw Poco::ApplicationException(Poco::format("%s: Stream changed, restart...", name())); pStream->setTimeline(pPacket->pts()); int flags = 0; if (pPacket->isVideo()) flags |= pPacket->isKeyFrame() ? MPEG_FLAG_IDR_FRAME : 0; mpeg_muxer_input(_pMuxer, pStream->id(), flags, pPacket->pts() * 90000 / 1000000, pPacket->dts() * 90000 / 1000000, pPacket->data(), pPacket->length()); #if (TRACE_VIDEO_FRAME_RATE) if (pPacket->isVideo()) { count++; uint64_t now = Core::System::timestamp(); int64_t delta = now - timeline; if (delta >= 1000000) { timeline = now; std::cout << " ==> " << name() << " fps: " << (double)count / (delta / 1000000) << std::endl; count = 0; } } #endif } void MPEGMuxer::disconnect() { if (_writeIndex > 0 && _socket != socket_invalid) flush(); if (_pMuxer) { mpeg_muxer_destroy(_pMuxer); _pMuxer = nullptr; } if (_socket != socket_invalid) { socket_close(_socket); _socket = socket_invalid; } _streams.clear(); } void MPEGMuxer::flush() { socket_sendto_by_time(_socket, _buffer, _writeIndex, 0, (const struct sockaddr*)&_destAddr, sizeof(_destAddr), 1000); _writeIndex = 0; } const int MPEGMuxer::getCodecID(Core::CodecID id) { for (const auto& m : codecMap) { if (id == m.id) return m.type; } throw Poco::NotFoundException( Poco::format("Video::MediaServer: Can't find %s in codec-map", Core::codecIDToString(id))); } } }