external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

for #250, use buffer to cache bytes, for system will split the udp packet.

Browse source
This commit is contained in:
winlin 2015-01-29 23:54:04 +08:00
parent b38bae2a89
commit 35ab0ea627
2 changed files with 35 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

38
trunk/src/app/srs_app_mpegts_udp.cpp
View file

@ -35,6 +35,7 @@ using namespace std;
#include <srs_kernel_ts.hpp>
#include <srs_kernel_stream.hpp>
#include <srs_kernel_ts.hpp>
#include <srs_kernel_buffer.hpp>
#ifdef SRS_AUTO_STREAM_CASTER
@ -42,11 +43,13 @@ SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c)
{
stream = new SrsStream();
context = new SrsTsContext();
buffer = new SrsSimpleBuffer();
output = _srs_config->get_stream_caster_output(c);
}
SrsMpegtsOverUdp::~SrsMpegtsOverUdp()
{
srs_freep(buffer);
srs_freep(stream);
srs_freep(context);
}
@ -58,16 +61,36 @@ int SrsMpegtsOverUdp::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
std::string peer_ip = inet_ntoa(from->sin_addr);
int peer_port = ntohs(from->sin_port);
// append to buffer.
buffer->append(buf, nb_buf);
// find the sync byte of mpegts.
char* p = buffer->bytes();
for (int i = 0; i < buffer->length(); i++) {
if (p[i] != 0x47) {
continue;
}
if (i > 0) {
buffer->erase(i);
}
break;
}
// drop ts packet when size not modulus by 188
if (nb_buf < SRS_TS_PACKET_SIZE || (nb_buf % SRS_TS_PACKET_SIZE) != 0) {
srs_warn("udp: drop %s:%d packet %d bytes", peer_ip.c_str(), peer_port, nb_buf);
if (buffer->length() < SRS_TS_PACKET_SIZE) {
srs_info("udp: wait %s:%d packet %d/%d bytes",
peer_ip.c_str(), peer_port, nb_buf, buffer->length());
return ret;
}
srs_info("udp: got %s:%d packet %d bytes", peer_ip.c_str(), peer_port, nb_buf);
srs_info("udp: got %s:%d packet %d/%d bytes",
peer_ip.c_str(), peer_port, nb_buf, buffer->length());
// use stream to parse ts packet.
for (int i = 0; i < nb_buf; i += SRS_TS_PACKET_SIZE) {
if ((ret = stream->initialize(buf + i, SRS_TS_PACKET_SIZE)) != ERROR_SUCCESS) {
int nb_packet = buffer->length() / SRS_TS_PACKET_SIZE;
for (int i = 0; i < nb_packet; i++) {
char* p = buffer->bytes() + (i * SRS_TS_PACKET_SIZE);
if ((ret = stream->initialize(p, SRS_TS_PACKET_SIZE)) != ERROR_SUCCESS) {
return ret;
}
@ -80,6 +103,11 @@ int SrsMpegtsOverUdp::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
}
srs_info("mpegts: parse udp packet completed");
// erase consumed bytes
if (nb_packet > 0) {
buffer->erase(nb_packet * SRS_TS_PACKET_SIZE);
}
return ret;
}

2
trunk/src/app/srs_app_mpegts_udp.hpp
View file

@ -36,6 +36,7 @@ class sockaddr_in;
class SrsStream;
class SrsTsContext;
class SrsConfDirective;
class SrsSimpleBuffer;
#ifdef SRS_AUTO_STREAM_CASTER
@ -49,6 +50,7 @@ class SrsMpegtsOverUdp : public ISrsTsHandler
private:
SrsStream* stream;
SrsTsContext* context;
SrsSimpleBuffer* buffer;
std::string output;
public:
SrsMpegtsOverUdp(SrsConfDirective* c);