1 #include "AmRtpMuxStream.h"
2 #include "AmConfig.h"
3 #include "AmRtpReceiver.h"
4
5 #include "log.h"
6
7 #include "sip/ip_util.h"
8 #include "sip/wheeltimer.h"
9
10 #include <unistd.h>
11 #include <sys/types.h>
12 #include <sys/socket.h>
13 #include <sys/ioctl.h>
14 #include <arpa/inet.h>
15 #include <netinet/in.h>
16
17 #define MUX_HDR_SETUP_LENGTH 5
18 #define MUX_HDR_COMPRESSED_LENGTH 3
19
20 #define RTP_MUX_MAX_FRAME_SIZE 256
21
22 static crc_t calc_crc4(u_int32 ts);
23 static u_int16 get_rtp_hdr_len(const rtp_hdr_t* hdr);
24 static bool rtp_hdr_changed(const rtp_hdr_t* hdr1, const rtp_hdr_t* hdr2);
25
26 static void decompress(const rtp_mux_hdr_compressed_t* rtp_mux_hdr_compressed, unsigned int ts_increment,
27 const rtp_hdr_t* old_rtp_hdr, unsigned char* rtp_restored_hdr);
28
AmRtpMuxStream()29 AmRtpMuxStream::AmRtpMuxStream()
30 : AmRtpStream(NULL, 0)
31 {
32 // fixme: init interface _if
33 }
34
~AmRtpMuxStream()35 AmRtpMuxStream::~AmRtpMuxStream()
36 {
37 }
38
get_rtp_hdr_len(const rtp_hdr_t * hdr)39 u_int16 get_rtp_hdr_len(const rtp_hdr_t* hdr) {
40 // fixme: check overflows
41 unsigned int hdr_len = sizeof(rtp_hdr_t) + (hdr->cc*4);
42 if(hdr->x != 0) {
43 // skip extension header
44 hdr_len +=
45 ntohs(((rtp_xhdr_t*) (hdr + hdr_len))->len)*4;
46 }
47 // if ((unsigned char*)(hdr + hdr_len) > (p.getBuffer()+s)) {
48 // ERROR("RTP packet with CC and xtension header too long!\n");
49 // }
50 return hdr_len;
51 }
52
calculate_ts_increment(rtp_mux_hdr_setup_t * setup_hdr)53 unsigned int calculate_ts_increment(rtp_mux_hdr_setup_t* setup_hdr) {
54 return setup_hdr->u ? setup_hdr->ts_inc * RTP_MUX_HDR_TS_MULTIPLIER_HIGH : setup_hdr->ts_inc * RTP_MUX_HDR_TS_MULTIPLIER_LOW;
55 }
56
recvPacket(int fd,unsigned char * pkt,size_t len)57 void AmRtpMuxStream::recvPacket(int fd, unsigned char* pkt, size_t len) {
58 AmRtpPacket p;
59 int s = p.recv(l_sd);
60
61 if (s<0) {
62 DBG("ERROR receiving packet on fd %d\n", fd);
63 return;
64 }
65
66 // DBG("received packet of length %d\n", s);
67 if (!s) return;
68
69 unsigned char* frame_ptr = p.getBuffer();
70 while ((unsigned char*) frame_ptr < p.getBuffer()+s) {
71 rtp_mux_hdr_t* mux_hdr = (rtp_mux_hdr_t*)frame_ptr;
72
73 // fixme: handle situation where no setup frame received previously
74 MuxStreamState& state = recv_streamstates[mux_hdr->sid];
75
76 if (mux_hdr->t == RTP_MUX_HDR_TYPE_SETUP) {
77 rtp_mux_hdr_setup_t* mux_hdr_setup = (rtp_mux_hdr_setup_t*)frame_ptr;
78 // DBG("received setup packed on stream ID %u for port %u, ts_increment %u, len %u hdr_len %u",
79 // mux_hdr_setup->sid, ntohs(mux_hdr_setup->dstport), mux_hdr_setup->ts_inc, mux_hdr->len, state.rtp_hdr_len);
80
81 // save params
82 state.dstport = ntohs(mux_hdr_setup->dstport);
83 state.ts_increment = calculate_ts_increment(mux_hdr_setup);
84
85 // fixme: handle RTCP
86
87 // save RTP header
88 rtp_hdr_t* hdr = (rtp_hdr_t*)(frame_ptr + sizeof(rtp_mux_hdr_setup_t));
89
90 state.rtp_hdr_len = get_rtp_hdr_len(hdr);
91 memcpy(state.rtp_hdr, hdr, state.rtp_hdr_len);
92
93 // DBG("setup packet for port %u ts_inc %u len %u hdr_len %u\n",
94 // state.dstport, state.ts_increment, mux_hdr->len, state.rtp_hdr_len);
95
96 AmRtpReceiver::instance()->recvdPacket(getLocalPort(), state.dstport, (unsigned char*)hdr, mux_hdr->len);
97
98 frame_ptr+= mux_hdr->len + sizeof(rtp_mux_hdr_setup_t) /* skip hdr + frame*/;
99 } else {
100
101 // decompress header
102 unsigned char rtp_pkt[MAX_RTP_PACKET_LEN];
103 decompress((rtp_mux_hdr_compressed_t*)frame_ptr, state.ts_increment, (const rtp_hdr_t*)state.rtp_hdr, rtp_pkt);
104 state.rtp_hdr_len = get_rtp_hdr_len((rtp_hdr_t*)rtp_pkt);
105 // save header
106 memcpy(state.rtp_hdr, rtp_pkt, state.rtp_hdr_len);
107 // copy payload
108 memcpy(rtp_pkt + state.rtp_hdr_len, frame_ptr+sizeof(rtp_mux_hdr_compressed_t), mux_hdr->len);
109
110 // DBG("received compressed packet for dstport %u of size %u\n", state.dstport, state.rtp_hdr_len + mux_hdr->len);
111 AmRtpReceiver::instance()->recvdPacket(getLocalPort(), state.dstport, rtp_pkt, state.rtp_hdr_len + mux_hdr->len);
112
113 frame_ptr+= mux_hdr->len + sizeof(rtp_mux_hdr_compressed_t) /* skip hdr + frame*/;
114 }
115 }
116 }
117
send(unsigned char * buffer,unsigned int b_size,const string & remote_ip,unsigned short remote_port,unsigned short rtp_dst_port)118 int _AmRtpMuxSender::send(unsigned char* buffer, unsigned int b_size,
119 const string& remote_ip, unsigned short remote_port, unsigned short rtp_dst_port) {
120 if (remote_ip.empty() || !remote_port) {
121 ERROR("internal error: need RTP MUX remote IP:port to send to\n");
122 return -1;
123 }
124
125 send_queues_mut.lock();
126 int res = send_queues[remote_ip+":"+int2str(remote_port)].send(buffer, b_size, remote_ip, remote_port, rtp_dst_port);
127 send_queues_mut.unlock();
128 return res;
129 }
130
close(const string & remote_ip,unsigned short remote_port,unsigned short rtp_dst_port)131 void _AmRtpMuxSender::close(const string& remote_ip, unsigned short remote_port, unsigned short rtp_dst_port) {
132 DBG("RTP MUX: closing queue to %s:%u for port %u\n", remote_ip.c_str(), remote_port, rtp_dst_port);
133 send_queues_mut.lock();
134 send_queues[remote_ip+":"+int2str(remote_port)].close(remote_ip, remote_port, rtp_dst_port);
135 send_queues_mut.unlock();
136 }
137
MuxStreamQueue()138 MuxStreamQueue::MuxStreamQueue()
139 : l_sd(0), end_ptr(buf), oldest_frame_i(false), mux_packet_id(0), is_setup(false)
140 {
141 memset(&r_saddr,0,sizeof(struct sockaddr_storage));
142 memset(&l_saddr,0,sizeof(struct sockaddr_storage));
143 }
144
init(const string & _remote_ip,unsigned short _remote_port)145 int MuxStreamQueue::init(const string& _remote_ip, unsigned short _remote_port) {
146 if (l_sd)
147 close();
148
149 remote_ip = _remote_ip;
150 remote_port = _remote_port;
151
152 int l_if = 0;
153 map<string,unsigned short>::iterator n_it=AmConfig::RTP_If_names.find(AmConfig::RtpMuxOutInterface);
154 if (n_it == AmConfig::RTP_If_names.end()) {
155 // WARN("rtp_mux_interface '%s' not found. Using default RTP interface instead\n", AmConfig::RtpMuxOutInterface.c_str());
156 } else {
157 l_if = n_it->second;
158 }
159
160 if (!am_inet_pton(AmConfig::RTP_Ifs[l_if].LocalIP.c_str(), &l_saddr)) {
161 ERROR("MuxStreamQueue::init: Invalid IP address: %s\n", AmConfig::RTP_Ifs[l_if].LocalIP.c_str());
162 return -1;
163 }
164
165 int retry = 10;
166 unsigned short port = 0;
167 for(;retry; --retry){
168
169 int sd=0;
170 if((sd = socket(l_saddr.ss_family, SOCK_DGRAM, 0)) == -1) {
171 ERROR("%s\n",strerror(errno));
172 return -2;
173 }
174
175 int true_opt = 1;
176 if(ioctl(sd, FIONBIO , &true_opt) == -1){
177 ERROR("%s\n",strerror(errno));
178 ::close(l_sd);
179 return -2;
180 }
181
182 l_sd = sd;
183
184 port = AmConfig::RTP_Ifs[l_if].getNextRtpPort();
185
186 am_set_port(&l_saddr,port);
187 if(bind(l_sd,(const struct sockaddr*)&l_saddr,SA_len(&l_saddr))) {
188 ::close(l_sd);
189 l_sd = 0;
190 DBG("bind: %s\n",strerror(errno));
191 continue;
192 }
193
194 break;
195 }
196
197 if (!retry){
198 ERROR("RTP MUX: could not find a free RTP port\n");
199 return -1;
200 }
201
202 int true_opt = 1;
203 // rco: does that make sense after bind() ????
204 if(setsockopt(l_sd, SOL_SOCKET, SO_REUSEADDR,
205 (void*)&true_opt, sizeof (true_opt)) == -1) {
206
207 ERROR("%s\n",strerror(errno));
208 ::close(l_sd);
209 l_sd = 0;
210 return -1;
211 }
212
213 if (!am_inet_pton(remote_ip.c_str(), &r_saddr)) {
214 WARN("Address not valid (host: %s).\n", remote_ip.c_str());
215 return -1;
216 }
217 am_set_port(&r_saddr, remote_port);
218
219 DBG("RTP MUX: opened connection <%s>:%u --> %s:%u\n",
220 AmConfig::RtpMuxOutInterface.empty() ? "default" : AmConfig::RtpMuxOutInterface.c_str(), port, remote_ip.c_str(), remote_port);
221 return 0;
222 }
223
close()224 void MuxStreamQueue::close() {
225 ::close(l_sd);
226 l_sd = 0;
227 }
228
calc_crc4(u_int32 ts)229 crc_t calc_crc4(u_int32 ts) {
230 crc_t res = crc_init();
231 res = crc_update(res, &ts, sizeof(u_int32));
232 return crc_finalize(res);
233 }
234
send(unsigned char * buffer,unsigned int b_size,const string & _remote_ip,unsigned short _remote_port,unsigned short rtp_dst_port)235 int MuxStreamQueue::send(unsigned char* buffer, unsigned int b_size,
236 const string& _remote_ip, unsigned short _remote_port,
237 unsigned short rtp_dst_port) {
238 if (b_size > RTP_MUX_MAX_FRAME_SIZE) {
239 ERROR("RTP MUX: trying to send packet of too large size (b_size = %u, max size %u)\n",
240 b_size, RTP_MUX_MAX_FRAME_SIZE);
241 return -1;
242 }
243
244 // initialize UDP connection to MUX IP:port
245 if ((remote_ip != _remote_ip) || (remote_port != _remote_port)) {
246 if (init(_remote_ip, _remote_port))
247 return -1;
248 }
249
250 // find stream_id for rtp_dst_port
251 unsigned char stream_id = 0;
252 map<unsigned short, unsigned char>::iterator mux_id_it = stream_ids.find(rtp_dst_port);
253 if (mux_id_it == stream_ids.end()) {
254 // set up a new stream_id for port rtp_dst_port
255 for (stream_id=0;stream_id<=256;stream_id++) {
256 if (streamstates.find(stream_id)==streamstates.end())
257 break;
258 }
259 if (stream_id==256) {
260 ERROR("trying to send more than 256 streams on RTP MUX connection to %s:%u\n", remote_ip.c_str(), remote_port);
261 // or: find the oldest stream and overwrite that one
262 return -1;
263 }
264 DBG("RTP MUX: setting up new MUX stream for dst port %u as stream_id %u on RTP MUX %s:%u\n",
265 rtp_dst_port, stream_id, remote_ip.c_str(), remote_port);
266 stream_ids[rtp_dst_port]=stream_id;
267 } else {
268 stream_id = mux_id_it->second;
269 }
270
271 MuxStreamState& stream_state = streamstates[stream_id];
272
273 bool send_setup_frame = false;
274 // RTP header changed -> send setup frame
275 if (rtp_hdr_changed((rtp_hdr_t*)stream_state.rtp_hdr, (rtp_hdr_t*)buffer)) {
276 // DBG("RTP hdr changed - sending setup frame\n");
277 send_setup_frame = true;
278 stream_state.setup_frame_ctr = MUX_SETUP_FRAME_REPEAT;
279 stream_state.ts_increment = DEFAULT_TS_INCREMENT;
280 }
281
282 // still in n initial setup period?
283 if (!send_setup_frame &&
284 stream_state.setup_frame_ctr && (stream_state.last_mux_packet_id != mux_packet_id)) {
285 stream_state.setup_frame_ctr--;
286 send_setup_frame = true;
287 }
288
289 // periodic update?
290 if (!send_setup_frame &&
291 (stream_state.last_mux_packet_id != mux_packet_id) &&
292 wheeltimer::instance()->interval_elapsed(stream_state.last_setup_frame_ts, MUX_PERIODIC_SETUP_FRAME_MS)) {
293 send_setup_frame = true;
294 }
295
296 rtp_hdr_t* rtp_hdr = (rtp_hdr_t*)buffer;
297 u_int16 rtp_hdr_len = get_rtp_hdr_len(rtp_hdr);
298 u_int16 data_len = b_size - rtp_hdr_len;
299 rtp_hdr_t* old_rtp_hdr = (rtp_hdr_t*)stream_state.rtp_hdr;
300
301 // length has changed (e.g. CSRC, extension hdr)
302 if (rtp_hdr_len != stream_state.rtp_hdr_len) {
303 send_setup_frame = true;
304 }
305
306 // ts_increment right?
307 if (!send_setup_frame &&
308 ntohl(rtp_hdr->ts) != (ntohl(old_rtp_hdr->ts) + stream_state.ts_increment)) {
309 u_int16 old_ts_increment = stream_state.ts_increment;
310 stream_state.ts_increment = ntohl(rtp_hdr->ts) - ntohl(old_rtp_hdr->ts);
311 DBG("corrected ts_increment %u -> %u; old_hdr_ts %u, ts = %u (FIXME: TS resync?)\n",
312 old_ts_increment, stream_state.ts_increment, ntohl(old_rtp_hdr->ts), ntohl(rtp_hdr->ts));
313 send_setup_frame = true;
314 }
315
316 if (!send_setup_frame) {
317 rtp_mux_hdr_compressed_t* rtp_mux_hdr_compressed = (rtp_mux_hdr_compressed_t*)end_ptr;
318 rtp_mux_hdr_compressed->t = RTP_MUX_HDR_TYPE_COMPRESSED;
319 rtp_mux_hdr_compressed->sid = stream_id;
320 rtp_mux_hdr_compressed->len = data_len;
321 rtp_mux_hdr_compressed->m = rtp_hdr->m;
322 rtp_mux_hdr_compressed->sn_lsb = ntohs(rtp_hdr->seq) & 7 /* 0b0111 */;
323 rtp_mux_hdr_compressed->ts_crc4 = calc_crc4(ntohl(rtp_hdr->ts));
324
325 // try decompressing
326 // DBG("test decompress\n");
327 unsigned char rtp_restored_hdr[MAX_RTP_HDR_LEN];
328 decompress(rtp_mux_hdr_compressed, stream_state.ts_increment, old_rtp_hdr, rtp_restored_hdr);
329
330 if (!memcmp(rtp_restored_hdr, rtp_hdr, rtp_hdr_len)) {
331 // compress -> decompress worked, send compressed
332 // skip compressed header
333 end_ptr+=sizeof(rtp_mux_hdr_compressed_t);
334 // copy payload (skipping RTP hdr)
335 memcpy(end_ptr, buffer+rtp_hdr_len, data_len);
336 end_ptr+= data_len;
337 // DBG("decompress matched, sending compressed frame (%lu mux hdr + %u payload bytes).\n",
338 // sizeof(rtp_mux_hdr_compressed_t), (unsigned int)(b_size-rtp_hdr_len));
339 } else {
340 // decompressed frame didn't match, sending setup frame
341 // DBG("decompressed frame didn't match, sending setup frame.\n");
342 send_setup_frame = true;
343 }
344 }
345
346 if (send_setup_frame) {
347 // setup frame header
348 rtp_mux_hdr_setup_t* setup_hdr = (rtp_mux_hdr_setup_t*)end_ptr;
349 setup_hdr->t = RTP_MUX_HDR_TYPE_SETUP;
350 setup_hdr->sid = stream_id;
351 setup_hdr->len = b_size;
352 setup_hdr->dstport = htons(rtp_dst_port);
353
354 // set ts increment
355 if (!(stream_state.ts_increment % RTP_MUX_HDR_TS_MULTIPLIER_HIGH)) {
356 // it's safe to divide by 160
357 setup_hdr->u = RTP_MUX_HDR_TS_MULTIPLIER_160;
358 setup_hdr->ts_inc = stream_state.ts_increment / RTP_MUX_HDR_TS_MULTIPLIER_HIGH;
359 } else {
360 // use div by 40
361 setup_hdr->u = RTP_MUX_HDR_TS_MULTIPLIER_40;
362 setup_hdr->ts_inc = stream_state.ts_increment / RTP_MUX_HDR_TS_MULTIPLIER_LOW;
363 }
364 if (stream_state.ts_increment != calculate_ts_increment(setup_hdr)) {
365 WARN("ts increment of %u can't be expressed by u %u and ts_inc %u (TS resync?)\n",
366 stream_state.ts_increment, setup_hdr->u, setup_hdr->ts_inc);
367 }
368
369 end_ptr += sizeof(rtp_mux_hdr_setup_t);
370
371 // copy complete frame (rtp hdr + payload)
372 memcpy(end_ptr, buffer, b_size);
373 end_ptr += b_size;
374
375 stream_state.last_setup_frame_ts = wheeltimer::instance()->wall_clock;
376 }
377
378 stream_state.last_mux_packet_id = mux_packet_id;
379
380 // save old RTP header
381 stream_state.rtp_hdr_len = get_rtp_hdr_len(rtp_hdr);
382 memcpy(stream_state.rtp_hdr, rtp_hdr, stream_state.rtp_hdr_len);
383
384 // DBG("queued frame on MUX stream for dst port %u, stream_id %u on RTP MUX %s:%u (hdr_len %u, len %u); send_setup_frame = %s\n",
385 // rtp_dst_port, stream_id, remote_ip.c_str(), remote_port, get_rtp_hdr_len((rtp_hdr_t*)buffer), b_size, send_setup_frame?"true":"false");
386
387 return sendQueue();
388 }
389
sendQueue(bool force)390 int MuxStreamQueue::sendQueue(bool force) {
391 bool force_send = force;
392
393 if (!oldest_frame_i) {
394 // record oldest TS in the queue
395 oldest_frame = wheeltimer::instance()->wall_clock;
396 oldest_frame_i = true;
397 } else if (wheeltimer::instance()->interval_elapsed(oldest_frame, AmConfig::RtpMuxMaxFrameAgeMs)) {
398 // enough time elapsed - flush queue
399 force_send = true;
400 }
401
402 unsigned int len = end_ptr - buf;
403
404 if (len >= AmConfig::RtpMuxMTUThreshold)
405 force_send = true;
406
407 if (!force_send) // can buffer more
408 return 0;
409
410 // DBG("sending frame on MUX stream \n");
411
412 int err = ::sendto(l_sd, buf, len, 0,
413 (const struct sockaddr *)&r_saddr,
414 SA_len(&r_saddr));
415
416 if(err == -1){
417 ERROR("RTP MUX: while sending RTP packet: '%s' trying to send %u bytes to %s:%u\n",
418 strerror(errno), len, am_inet_ntop(&r_saddr).c_str(), am_get_port(&r_saddr));
419 }
420
421 // reset queue
422 end_ptr = buf;
423 oldest_frame_i = false;
424
425 // increase, so send code above can see queue being cleared in between
426 mux_packet_id++;
427
428 return err;
429 }
430
close(const string & _remote_ip,unsigned short _remote_port,unsigned short rtp_dst_port)431 void MuxStreamQueue::close(const string& _remote_ip, unsigned short _remote_port, unsigned short rtp_dst_port) {
432 if ((remote_ip != _remote_ip) || (remote_port != _remote_port) || !rtp_dst_port) {
433 return;
434 }
435
436 // do we have that rtp_dst_port already?
437 map<unsigned short, unsigned char>::iterator mux_id_it = stream_ids.find(rtp_dst_port);
438 if (mux_id_it == stream_ids.end()) {
439 //we don't have it - all ok
440 return;
441 }
442 unsigned char stream_id = mux_id_it->second;
443
444 // flush remaining frames - might be including frames of other streams, but this
445 // destination might not be periodically handled in the future if this was the only
446 // stream to this destination
447 DBG("RTP MUX: flushing queue for %s:%u\n", remote_ip.c_str(), remote_port);
448 sendQueue(true);
449
450 // fixme: maybe the stream_id closing should be signaled?
451 // remove that stream here
452 DBG("RTP MUX: freeing stream_id %u to %s:%u\n", stream_id, remote_ip.c_str(), remote_port);
453 stream_ids.erase(mux_id_it);
454 streamstates.erase(stream_id);
455 }
456
decompress(const rtp_mux_hdr_compressed_t * rtp_mux_hdr_compressed,unsigned int ts_increment,const rtp_hdr_t * old_rtp_hdr,unsigned char * rtp_restored_hdr)457 void decompress(const rtp_mux_hdr_compressed_t* rtp_mux_hdr_compressed, unsigned int ts_increment,
458 const rtp_hdr_t* old_rtp_hdr, unsigned char* rtp_restored_hdr) {
459 u_int16 hdr_len = get_rtp_hdr_len(old_rtp_hdr);
460 // use old as template
461 memcpy(rtp_restored_hdr, old_rtp_hdr, hdr_len);
462 rtp_hdr_t* rtp_hdr = (rtp_hdr_t*)rtp_restored_hdr;
463 // copy marker bit
464 rtp_hdr->m = rtp_mux_hdr_compressed->m;
465 // use 3 lsb bits sent in compressed
466 u_int16 old_rtp_hdr_seq = ntohs(old_rtp_hdr->seq);
467 u_int16 rtp_hdr_seq = (ntohs(rtp_hdr->seq) & 0xFFF8/*(~ (u_int16)7)*/) + rtp_mux_hdr_compressed->sn_lsb;
468 // new Seqno smaller?
469 if (rtp_hdr_seq < old_rtp_hdr_seq) {
470 // next 3-bit
471 rtp_hdr_seq+=8;
472 }
473 // test different possible SNs
474 u_int16 test_sns[9] = {0, 8, 1, (u_int16)-8, 16, 32, 40, 48, (u_int16)-16};
475
476 bool found = false;
477 for (size_t i=0; i<9; i++) {
478 u_int16 sn_diff_test = test_sns[i];
479 u_int16 sn_test = rtp_hdr_seq + sn_diff_test;
480 u_int16 sn_diff = sn_test - old_rtp_hdr_seq;
481 u_int32 ts_test = ntohl(rtp_hdr->ts) + sn_diff * ts_increment;
482 // DBG("testing SN diff %u - sn_test %u, sn_diff %u, ts_test %u, crc4(ts) %u\n",
483 // sn_diff_test, sn_test, sn_diff, ts_test, calc_crc4(ts_test));
484 if (calc_crc4(ts_test) == rtp_mux_hdr_compressed->ts_crc4) {
485 // found correct TS -> SeqNo must be correct as well
486 rtp_hdr->ts = htonl(ts_test);
487 rtp_hdr->seq = htons(sn_test);
488 found = true;
489 break;
490 }
491 }
492 if (!found) {
493 WARN("couldn't reconstruct TS/Seqno: old seq %u, ts %u, ts_inc %u, sn_lsb %u, ts_crc4 %u\n",
494 old_rtp_hdr_seq, old_rtp_hdr->ts, ts_increment, rtp_mux_hdr_compressed->sn_lsb, rtp_mux_hdr_compressed->ts_crc4);
495 // using most likely one - or drop?
496 u_int16 sn_diff = rtp_hdr_seq - old_rtp_hdr_seq;
497 rtp_hdr->ts += htonl(sn_diff * ts_increment);
498 } else {
499 // DBG("found\n");
500 }
501 }
502
503 // return true if RTP header has changed significantly so will need to be sent a setup frame
rtp_hdr_changed(const rtp_hdr_t * hdr1,const rtp_hdr_t * hdr2)504 bool rtp_hdr_changed(const rtp_hdr_t* hdr1, const rtp_hdr_t* hdr2) {
505 // PT change SSRC change CSRC count change
506 if ((hdr1->pt != hdr2->pt) || (hdr1->ssrc != hdr2->ssrc) || (hdr1->cc != hdr2->cc) || (hdr1->x != hdr2->x))
507 return true;
508 // fixme: check CSRCs and extension headers' contents
509
510 // TS resync?
511 if( ts_less()(ntohl(hdr1->ts), ntohl(hdr2->ts) - RESYNC_MAX_DELAY/2) ||
512 !ts_less()(ntohl(hdr1->ts), ntohl(hdr2->ts) + RESYNC_MAX_DELAY) ) {
513 DBG("TS resync: hdr1->ts %u, hdr2->ts %u\n", ntohl(hdr1->ts), ntohl (hdr2->ts));
514 return true;
515 }
516
517 return false;
518 }
519