demux: adaptive: declare codecs for MIME only ttml/webvtt
[vlc.git] / modules / access_output / rist.c
1 /*****************************************************************************
2  *  * rist.c: RIST (Reliable Internet Stream Transport) output module
3  *****************************************************************************
4  * Copyright (C) 2018, DVEO, the Broadcast Division of Computer Modules, Inc.
5  * Copyright (C) 2018, SipRadius LLC
6  *
7  * Authors: Sergio Ammirata <sergio@ammirata.net>
8  *          Daniele Lacamera <root@danielinux.net>
9  *
10  * This program is free software; you can redistribute it and/or modify it
11  * under the terms of the GNU Lesser General Public License as published by
12  * the Free Software Foundation; either version 2.1 of the License, or
13  * (at your option) any later version.
14  *
15  * This program is distributed in the hope that it will be useful,
16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18  * GNU Lesser General Public License for more details.
19  *
20  * You should have received a copy of the GNU Lesser General Public License
21  * along with this program; if not, write to the Free Software Foundation,
22  * Inc., 51 Franklin Street, Fifth Floor, Boston MA 02110-1301, USA.
23  *****************************************************************************/
24
25 #ifdef HAVE_CONFIG_H
26 # include "config.h"
27 #endif
28
29 #include <vlc_common.h>
30 #include <vlc_interrupt.h>
31 #include <vlc_fs.h>
32 #include <vlc_plugin.h>
33 #include <vlc_sout.h>
34 #include <vlc_block.h>
35 #include <vlc_network.h>
36 #include <vlc_threads.h>
37 #include <vlc_rand.h>
38 #ifdef HAVE_POLL
39 #include <poll.h>
40 #endif
41 #include <sys/time.h>
42 #ifdef HAVE_SYS_SOCKET_H
43 #include <sys/socket.h>
44 #endif
45 #include <bitstream/ietf/rtcp_rr.h>
46 #include <bitstream/ietf/rtcp_sr.h>
47 #include <bitstream/ietf/rtcp_fb.h>
48 #include <bitstream/ietf/rtcp_sdes.h>
49 #include <bitstream/ietf/rtp.h>
50
51 #include "../access/rist.h"
52
53 /* Uncomment the following to introduce induced packet loss for TESTING purposes only */
54 /*#define TEST_PACKET_LOSS*/
55
56 /* The default target packet size */
57 #define RIST_TARGET_PACKET_SIZE 1328
58 /* The default caching delay for output data */
59 #define DEFAULT_CACHING_DELAY 50
60 /* The default buffer size in ms */
61 #define DEFAULT_BUFFER_SIZE 0
62 /* Calculate and print stats once per second */
63 #define STATS_INTERVAL 1000 /*ms*/
64
65 #define MPEG_II_TRANSPORT_STREAM (0x21)
66 #define RIST_DEFAULT_PORT 1968
67
68 #define SOUT_CFG_PREFIX "sout-rist-"
69
70 static const char *const ppsz_sout_options[] = {
71     "packet-size",
72     "caching",
73     "buffer-size",
74     "ssrc",
75     "stream-name",
76     NULL
77 };
78
79 typedef struct
80 {
81     struct       rist_flow *flow;
82     uint16_t     rtp_counter;
83     char         receiver_name[MAX_CNAME];
84     uint64_t     last_rtcp_tx;
85     vlc_thread_t ristthread;
86     vlc_thread_t senderthread;
87     size_t       i_packet_size;
88     bool         b_mtu_warning;
89     bool         b_ismulticast;
90     vlc_mutex_t  lock;
91     vlc_mutex_t  fd_lock;
92     block_t      *p_pktbuffer;
93     uint64_t     i_ticks_caching;
94     uint32_t     ssrc;
95     block_fifo_t *p_fifo;
96     /* stats variables */
97     uint64_t     i_last_stat;
98     uint32_t     i_retransmit_packets;
99     uint32_t     i_total_packets;
100 } sout_access_out_sys_t;
101
102 static struct rist_flow *rist_init_tx()
103 {
104     struct rist_flow *flow = calloc(1, sizeof(struct rist_flow));
105     if (!flow)
106         return NULL;
107
108     flow->reset = 1;
109     flow->buffer = calloc(RIST_QUEUE_SIZE, sizeof(struct rtp_pkt));
110     if ( unlikely( flow->buffer == NULL ) )
111     {
112         free(flow);
113         return NULL;
114     }
115     flow->fd_out = -1;
116     flow->fd_rtcp = -1;
117     flow->fd_rtcp_m = -1;
118
119     return flow;
120 }
121
122 static struct rist_flow *rist_udp_transmitter(sout_access_out_t *p_access, char *psz_dst_server, 
123     int i_dst_port, bool b_ismulticast)
124 {
125     struct rist_flow *flow;
126     flow = rist_init_tx();
127     if (!flow)
128         return NULL;
129
130     flow->fd_out = net_ConnectDgram(p_access, psz_dst_server, i_dst_port, -1, IPPROTO_UDP );
131     if (flow->fd_out < 0)
132     {
133         msg_Err( p_access, "cannot open output socket" );
134         goto fail;
135     }
136
137     if (b_ismulticast) {
138         flow->fd_rtcp_m = net_OpenDgram(p_access, psz_dst_server, i_dst_port + 1,
139             NULL, 0, IPPROTO_UDP);
140         if (flow->fd_rtcp_m < 0)
141         {
142             msg_Err( p_access, "cannot open multicast nack socket" );
143             goto fail;
144         }
145     }
146
147     flow->fd_rtcp = net_ConnectDgram(p_access, psz_dst_server, i_dst_port + 1, -1, IPPROTO_UDP );
148     if (flow->fd_rtcp < 0)
149     {
150         msg_Err( p_access, "cannot open nack socket" );
151         goto fail;
152     }
153
154     char *psz_streamname = NULL;
155     psz_streamname = var_InheritString( p_access, SOUT_CFG_PREFIX "stream-name" );
156     if ( psz_streamname != NULL && psz_streamname[0] != '\0')
157     {
158         int name_length = snprintf(flow->cname, MAX_CNAME, "%s", psz_streamname);
159         if (name_length >= MAX_CNAME)
160             flow->cname[MAX_CNAME-1] = 0;
161         free( psz_streamname );
162     }
163     else
164         populate_cname(flow->fd_rtcp, flow->cname);
165
166     msg_Info(p_access, "our cname is %s", flow->cname);
167
168     return flow;
169
170 fail:
171     if (flow->fd_out != -1)
172         vlc_close(flow->fd_out);
173     if (flow->fd_rtcp != -1)
174         vlc_close(flow->fd_rtcp);
175     if (flow->fd_rtcp_m != -1)
176         vlc_close(flow->fd_rtcp_m);
177     free(flow->buffer);
178     free(flow);
179     return NULL;
180 }
181
182 static void rist_retransmit(sout_access_out_t *p_access, struct rist_flow *flow, uint16_t seq)
183 {
184     sout_access_out_sys_t *p_sys = p_access->p_sys;
185     struct rtp_pkt *pkt = &(flow->buffer[seq]);
186     if (pkt->buffer == NULL)
187     {
188         msg_Err(p_access, "RIST recovery: missing requested packet %d, buffer not yet full", seq);
189         return;
190     }
191
192     /* Mark SSID for retransmission (change the last bit of the ssrc to 1) */
193     pkt->buffer->p_buffer[11] |= (1 << 0);
194 #ifdef TEST_PACKET_LOSS
195 #   warning COMPILED WITH SELF INFLICTED PACKET LOSS
196         if ((flow->packets_count % 14) == 0) {
197             return;
198         }
199 #endif
200     uint32_t rtp_age = flow->hi_timestamp - pkt->rtp_ts;
201     uint64_t age = ts_get_from_rtp(rtp_age)/1000;
202     if (flow->rtp_latency > 0 && rtp_age > flow->rtp_latency)
203     {
204         msg_Err(p_access, "   Not Sending Nack #%d, too old (age %"PRId64" ms), current seq is:" \
205             " [%d]. Perhaps you should increase the buffer-size ...", seq, age, flow->wi);
206     }
207     else
208     {
209         msg_Dbg(p_access, "   Sending Nack #%d (age %"PRId64" ms), current seq is: [%d]", 
210             seq, age, flow->wi);
211         p_sys->i_retransmit_packets++;
212         vlc_mutex_lock( &p_sys->fd_lock );
213         if (rist_Write(flow->fd_out, pkt->buffer->p_buffer, pkt->buffer->i_buffer) 
214                 != (ssize_t)pkt->buffer->i_buffer) {
215             msg_Err(p_access, "Error sending retransmitted packet after 2 tries ...");
216         }
217
218         vlc_mutex_unlock( &p_sys->fd_lock );
219     }
220 }
221
222 static void process_nack(sout_access_out_t *p_access, uint8_t  ptype, uint16_t nrecords, 
223     struct rist_flow *flow, uint8_t *pkt)
224 {
225     sout_access_out_sys_t *p_sys = p_access->p_sys;
226     int i,j;
227
228     /*msg_Info(p_access, "   Nack (BbRR), %d record(s), Window: [%d:%d-->%d]", nrecords, 
229         flow->ri, flow->wi, flow->wi-flow->ri);*/
230
231     if (ptype == RTCP_PT_RTPFR)
232     {
233         uint8_t pi_ssrc[4];
234         rtcp_fb_get_ssrc_media_src(pkt, pi_ssrc);
235         if (memcmp(pi_ssrc, "RIST", 4) != 0)
236         {
237             msg_Info(p_access, "   Ignoring Nack with name %s", pi_ssrc);
238             return; /* Ignore app-type not RIST */
239         }
240
241         for (i = 0; i < (nrecords-2); i++) {
242             uint16_t missing;
243             uint16_t additional;
244             uint8_t *rtp_nack_record = (pkt + 12 + i * 4);
245             missing = rtcp_fb_nack_get_range_start(rtp_nack_record);
246             additional = rtcp_fb_nack_get_range_extra(rtp_nack_record);
247             /*msg_Info(p_access, "   Nack (Range), %d, current seq is: [%d]", missing, flow->wi);*/
248             vlc_mutex_lock( &p_sys->lock );
249             rist_retransmit(p_access, flow, missing);
250             for (j = 0; j < additional; j++) {
251                 rist_retransmit(p_access, flow, missing + j + 1);
252             }
253             vlc_mutex_unlock( &p_sys->lock );
254         }
255     }
256     else if (ptype == RTCP_PT_RTPFB)
257     {
258         for (i = 0; i < (nrecords-2); i++) {
259             uint16_t missing;
260             uint16_t bitmask;
261             uint8_t *rtp_nack_record = (pkt + 12 + i * 4);
262             missing = rtcp_fb_nack_get_packet_id(rtp_nack_record);
263             bitmask = rtcp_fb_nack_get_bitmask_lost(rtp_nack_record);
264             /*msg_Info(p_access, "  Nack (Bitmask), %d, current seq is: [%d]", missing, flow->wi);*/
265             vlc_mutex_lock( &p_sys->lock );
266             rist_retransmit(p_access, flow, missing);
267             for (j = 0; j < 16; j++) {
268                 if ((bitmask & (1 << j)) == (1 << j)) {
269                     rist_retransmit(p_access, flow, missing + j + 1);
270                 }
271             }
272             vlc_mutex_unlock( &p_sys->lock );
273         }
274     }
275     else
276     {
277         msg_Err(p_access, "   !!! Wrong feedback. Ptype is %02x!=%02x, FMT: %02x", ptype, 
278             RTCP_PT_RTPFR, rtcp_fb_get_fmt(pkt));
279     }
280 }
281
282 static void rist_rtcp_recv(sout_access_out_t *p_access, struct rist_flow *flow, uint8_t *pkt_raw, 
283     size_t len)
284 {
285     sout_access_out_sys_t *p_sys = p_access->p_sys;
286     uint8_t *pkt = pkt_raw;
287     uint8_t  ptype;
288     uint16_t processed_bytes = 0;
289     uint16_t records;
290
291     while (processed_bytes < len) {
292         pkt = pkt_raw + processed_bytes;
293         /* safety checks */
294         uint16_t bytes_left = len - processed_bytes + 1;
295         if ( bytes_left < 4 )
296         {
297             /* we must have at least 4 bytes */
298             msg_Err(p_access, "Rist rtcp packet must have at least 4 bytes, we have %d", 
299                 bytes_left);
300             return; 
301         }
302         else if (!rtp_check_hdr(pkt))
303         {
304             /* check for a valid rtp header */
305             msg_Err(p_access, "Malformed feedback packet starting with %02x, ignoring.", pkt[0]);
306             return;
307         }
308
309         ptype =  rtcp_get_pt(pkt);
310         records = rtcp_get_length(pkt);
311         uint16_t bytes = (uint16_t)(4 * (1 + records));
312         if (bytes > bytes_left)
313         {
314             /* check for a sane number of bytes */
315             msg_Err(p_access, "Malformed feedback packet, wrong len %d, expecting %u bytes in the" \
316                 " packet, got a buffer of %u bytes. ptype = %d", rtcp_get_length(pkt), bytes, 
317                 bytes_left, ptype);
318             return;
319         }
320
321         switch(ptype) {
322             case RTCP_PT_RTPFR:
323             case RTCP_PT_RTPFB:
324                 process_nack(p_access, ptype, records, flow, pkt);
325                 break;
326
327             case RTCP_PT_RR:
328                 /*
329                 if (p_sys->b_ismulticast == false)
330                     process_rr(f, pkt, len);
331                 */
332                 break;
333
334             case RTCP_PT_SDES:
335                 {
336                     if (p_sys->b_ismulticast == false)
337                     {
338                         int8_t name_length = rtcp_sdes_get_name_length(pkt);
339                         if (name_length > bytes_left)
340                         {
341                             /* check for a sane number of bytes */
342                             msg_Err(p_access, "Malformed SDES packet, wrong cname len %u, got a " \
343                                 "buffer of %u bytes.", name_length, bytes_left);
344                             return;
345                         }
346                         if (memcmp(pkt + RTCP_SDES_SIZE, p_sys->receiver_name, name_length) != 0)
347                         {
348                             memcpy(p_sys->receiver_name, pkt + RTCP_SDES_SIZE, name_length);
349                             msg_Info(p_access, "Receiver name: %s", p_sys->receiver_name);
350                         }
351                     }
352                 }
353                 break;
354
355             case RTCP_PT_SR:
356                 break;
357
358             default:
359                 msg_Err(p_access, "   Unrecognized RTCP packet with PTYPE=%02x!!", ptype);
360         }
361         processed_bytes += bytes;
362     }
363 }
364
365 static void rist_rtcp_send(sout_access_out_t *p_access)
366 {
367     sout_access_out_sys_t *p_sys = p_access->p_sys;
368     struct rist_flow *flow = p_sys->flow;
369     uint8_t rtcp_buf[RTCP_SR_SIZE + RTCP_SDES_SIZE + MAX_CNAME] = { };
370     struct timeval tv;
371     int r;
372     uint64_t fractions;
373     uint16_t namelen = strlen(flow->cname) + 1;
374     gettimeofday(&tv, NULL);
375
376     /* Populate SR for sender report */
377     uint8_t *p_sr = rtcp_buf;
378     rtp_set_hdr(p_sr);
379     rtcp_sr_set_pt(p_sr);
380     rtcp_sr_set_length(p_sr, 6);
381     rtcp_fb_set_int_ssrc_pkt_sender(p_sr, p_sys->ssrc);
382     rtcp_sr_set_ntp_time_msw(p_sr, tv.tv_sec + SEVENTY_YEARS_OFFSET);
383     fractions = (uint64_t)tv.tv_usec;
384     fractions <<= 32ULL;
385     fractions /= 1000000ULL;
386     rtcp_sr_set_ntp_time_lsw(p_sr, (uint32_t)fractions);
387     rtcp_sr_set_rtp_time(p_sr, rtp_get_ts(vlc_tick_now()));
388     vlc_mutex_lock( &p_sys->lock );
389     rtcp_sr_set_packet_count(p_sr, flow->packets_count);
390     rtcp_sr_set_octet_count(p_sr, flow->bytes_count);
391     vlc_mutex_unlock( &p_sys->lock );
392
393     /* Populate SDES for sender description */
394     uint8_t *p_sdes = (rtcp_buf + RTCP_SR_SIZE);
395     /* we need to make sure it is a multiple of 4, pad if necessary */
396     if ((namelen - 2) & 0x3)
397         namelen = ((((namelen - 2) >> 2) + 1) << 2) + 2;
398     rtp_set_hdr(p_sdes);
399     rtp_set_cc(p_sdes, 1); /* Actually it is source count in this case */
400     rtcp_sdes_set_pt(p_sdes);
401     rtcp_set_length(p_sdes, (namelen >> 2) + 2);
402     rtcp_sdes_set_cname(p_sdes, 1);
403     rtcp_sdes_set_name_length(p_sdes, strlen(flow->cname));
404     p_sdes += RTCP_SDES_SIZE;
405     strlcpy((char *)p_sdes, flow->cname, namelen);
406
407     /* Send the rtcp message */
408     r = send(flow->fd_rtcp, rtcp_buf, RTCP_SR_SIZE + RTCP_SDES_SIZE + namelen, 0);
409     (void)r;
410 }
411
412 static void *rist_thread(void *data)
413 {
414     sout_access_out_t *p_access = data;
415     sout_access_out_sys_t *p_sys = p_access->p_sys;
416     uint64_t now;
417     uint8_t pkt[RTP_PKT_SIZE];
418     struct pollfd pfd[2];
419     int ret;
420     ssize_t r;
421
422     int poll_sockets = 1;
423     pfd[0].fd = p_sys->flow->fd_rtcp;
424     pfd[0].events = POLLIN;
425     if (p_sys->b_ismulticast)
426     {
427         pfd[1].fd = p_sys->flow->fd_rtcp_m;
428         pfd[1].events = POLLIN;
429         poll_sockets++;
430     }
431
432     for (;;) {
433         ret = poll(pfd, poll_sockets, RTCP_INTERVAL >> 1);
434         int canc = vlc_savecancel();
435         if (ret > 0)
436         {
437             if (pfd[0].revents & POLLIN)
438             {
439                 r = rist_Read(p_sys->flow->fd_rtcp, pkt, RTP_PKT_SIZE);
440                 if (r == RTP_PKT_SIZE) {
441                     msg_Err(p_access, "Rist RTCP messsage is too big (%zd bytes) and was probably " \
442                         "cut, please keep it under %d bytes", r, RTP_PKT_SIZE);
443                 }
444                 if (unlikely(r == -1)) {
445                     msg_Err(p_access, "socket %d error: %s\n", p_sys->flow->fd_rtcp, 
446                         gai_strerror(errno));
447                 }
448                 else {
449                     rist_rtcp_recv(p_access, p_sys->flow, pkt, r);
450                 }
451             }
452             if (p_sys->b_ismulticast && (pfd[1].revents & POLLIN))
453             {
454                 r = rist_Read(p_sys->flow->fd_rtcp_m, pkt, RTP_PKT_SIZE);
455                 if (r == RTP_PKT_SIZE) {
456                     msg_Err(p_access, "Rist RTCP messsage is too big (%zd bytes) and was " \
457                         "probably cut, please keep it under %d bytes", r, RTP_PKT_SIZE);
458                 }
459                 if (unlikely(r == -1)) {
460                     msg_Err(p_access, "mcast socket %d error: %s\n", p_sys->flow->fd_rtcp_m,
461                         gai_strerror(errno));
462                 }
463                 else {
464                     rist_rtcp_recv(p_access, p_sys->flow, pkt, r);
465                 }
466             }
467         }
468
469         /* And, in any case: */
470         now = vlc_tick_now();
471         if ((now - p_sys->last_rtcp_tx) > VLC_TICK_FROM_MS(RTCP_INTERVAL))
472         {
473             rist_rtcp_send(p_access);
474             p_sys->last_rtcp_tx = now;
475         }
476         vlc_restorecancel (canc);
477     }
478
479     return NULL;
480 }
481
482 /****************************************************************************
483  * RTP send
484  ****************************************************************************/
485 static void* ThreadSend( void *data )
486 {
487     sout_access_out_t *p_access = data;
488     sout_access_out_sys_t *p_sys = p_access->p_sys;
489     vlc_tick_t i_caching = p_sys->i_ticks_caching;
490     struct rist_flow *flow = p_sys->flow;
491
492     for (;;)
493     {
494         ssize_t len = 0;
495         uint16_t seq = 0;
496         uint32_t pkt_ts = 0;
497         block_t *out = block_FifoGet( p_sys->p_fifo );
498
499         block_cleanup_push( out );
500         vlc_tick_wait (out->i_dts + i_caching);
501         vlc_cleanup_pop();
502
503         len = out->i_buffer;
504         int canc = vlc_savecancel();
505
506         seq = rtp_get_seqnum(out->p_buffer);
507         pkt_ts = rtp_get_timestamp(out->p_buffer);
508
509         vlc_mutex_lock( &p_sys->fd_lock );
510 #ifdef TEST_PACKET_LOSS
511 #   warning COMPILED WITH SELF INFLICTED PACKET LOSS
512         if ((seq % 14) == 0) {
513             /*msg_Err(p_access, "Dropped packet with seq number %d ...", seq);*/
514         }
515         else
516         {
517             if (rist_Write(flow->fd_out, out->p_buffer, len) != len) {
518                 msg_Err(p_access, "Error sending data packet after 2 tries ...");
519             }
520         }
521 #else
522         if (rist_Write(flow->fd_out, out->p_buffer, len) != len) {
523             msg_Err(p_access, "Error sending data packet after 2 tries ...");
524         }
525 #endif
526         vlc_mutex_unlock( &p_sys->fd_lock );
527
528         /* Insert Into Queue */
529         vlc_mutex_lock( &p_sys->lock );
530         /* Always replace the existing one with the new one */
531         struct rtp_pkt *pkt;
532         pkt = &(flow->buffer[seq]);
533         if (pkt->buffer)
534         {
535             block_Release(pkt->buffer);
536             pkt->buffer = NULL;
537         }
538         pkt->rtp_ts = pkt_ts;
539         pkt->buffer = out;
540
541         if (flow->reset == 1)
542         {
543             msg_Info(p_access, "Traffic detected");
544             /* First packet in the queue */
545             flow->reset = 0;
546         }
547         flow->wi = seq;
548         flow->hi_timestamp = pkt_ts;
549         /* Stats for RTCP feedback packets */
550         flow->packets_count++;
551         flow->bytes_count += len;
552         flow->last_out = seq;
553         vlc_mutex_unlock( &p_sys->lock );
554
555         /* We print out the stats once per second */
556         uint64_t now = vlc_tick_now();
557         uint64_t interval = (now - p_sys->i_last_stat);
558         if ( interval > VLC_TICK_FROM_MS(STATS_INTERVAL) )
559         {
560             if (p_sys->i_retransmit_packets > 0)
561             {
562                 float quality = 100;
563                 if (p_sys->i_total_packets > 0)
564                     quality = (float)100 - (float)100*(float)(p_sys->i_retransmit_packets)
565                         /(float)p_sys->i_total_packets;
566                 msg_Info(p_access, "STATS: Total %u, Retransmitted %u, Link Quality %.2f%%", 
567                     p_sys->i_total_packets, p_sys->i_retransmit_packets, quality);
568             }
569             p_sys->i_last_stat = now;
570             p_sys->i_retransmit_packets = 0;
571             p_sys->i_total_packets = 0;
572         }
573         p_sys->i_total_packets++;
574
575         vlc_restorecancel (canc);
576     }
577     return NULL;
578 }
579
580 static void SendtoFIFO( sout_access_out_t *p_access, block_t *buffer )
581 {
582     sout_access_out_sys_t *p_sys = p_access->p_sys;
583     uint16_t seq = p_sys->rtp_counter++;
584
585     /* Set fresh rtp header data */
586     uint8_t *bufhdr = buffer->p_buffer;
587     rtp_set_hdr(bufhdr);
588     rtp_set_type(bufhdr, MPEG_II_TRANSPORT_STREAM);
589     rtp_set_seqnum(bufhdr, seq);
590     rtp_set_int_ssrc(bufhdr, p_sys->ssrc);
591     uint32_t pkt_ts = rtp_get_ts(buffer->i_dts);
592     rtp_set_timestamp(bufhdr, pkt_ts);
593
594     block_t *pkt = block_Duplicate(buffer);
595     block_FifoPut( p_sys->p_fifo, pkt );
596 }
597
598 static ssize_t Write( sout_access_out_t *p_access, block_t *p_buffer )
599 {
600     sout_access_out_sys_t *p_sys = p_access->p_sys;
601     int i_len = 0;
602
603     while( p_buffer )
604     {
605         block_t *p_next;
606         int i_block_split = 0;
607
608         if( !p_sys->b_mtu_warning && p_buffer->i_buffer > p_sys->i_packet_size )
609         {
610             msg_Warn( p_access, "Buffer data size (%zu) > configured packet size (%zu), you " \
611                 "should probably increase the configured packet size", p_buffer->i_buffer, 
612                 p_sys->i_packet_size );
613             p_sys->b_mtu_warning = true;
614         }
615
616         /* Temp buffer is already too large, flush */
617         if( p_sys->p_pktbuffer->i_buffer + p_buffer->i_buffer > p_sys->i_packet_size )
618         {
619             SendtoFIFO(p_access, p_sys->p_pktbuffer);
620             p_sys->p_pktbuffer->i_buffer = RTP_HEADER_SIZE;
621         }
622
623         i_len += p_buffer->i_buffer;
624
625         while( p_buffer->i_buffer )
626         {
627
628             size_t i_write = __MIN( p_buffer->i_buffer, p_sys->i_packet_size );
629
630             i_block_split++;
631
632             if( p_sys->p_pktbuffer->i_buffer == RTP_HEADER_SIZE )
633             {
634                 p_sys->p_pktbuffer->i_dts = p_buffer->i_dts;
635             }
636
637             memcpy( p_sys->p_pktbuffer->p_buffer + p_sys->p_pktbuffer->i_buffer,
638                     p_buffer->p_buffer, i_write );
639
640             p_sys->p_pktbuffer->i_buffer += i_write;
641             p_buffer->p_buffer += i_write;
642             p_buffer->i_buffer -= i_write;
643
644             /*  Flush if we reached the target size for the case of block size < target packet size.
645              *  Also flush when we are in block_split > 1 for the case when the block_size is
646              *  larger than the packet-size because we need to continue the inner loop */
647             if( p_sys->p_pktbuffer->i_buffer == p_sys->i_packet_size || i_block_split > 1 )
648             {
649                 SendtoFIFO(p_access, p_sys->p_pktbuffer);
650                 p_sys->p_pktbuffer->i_buffer = RTP_HEADER_SIZE;
651             }
652
653         }
654
655         p_next = p_buffer->p_next;
656         block_Release( p_buffer );
657         p_buffer = p_next;
658
659     }
660
661     if ( i_len <= 0 ) {
662         block_ChainRelease( p_buffer );
663     }
664     return i_len;
665 }
666
667 static int Control( sout_access_out_t *p_access, int i_query, va_list args )
668 {
669     VLC_UNUSED( p_access );
670
671     int i_ret = VLC_SUCCESS;
672
673     switch( i_query )
674     {
675         case ACCESS_OUT_CONTROLS_PACE:
676             *va_arg( args, bool * ) = false;
677             break;
678
679         default:
680             i_ret = VLC_EGENERIC;
681             break;
682     }
683
684     return i_ret;
685 }
686
687 static void Clean( sout_access_out_t *p_access )
688 {
689     sout_access_out_sys_t *p_sys = p_access->p_sys;
690
691     if( likely(p_sys->p_fifo != NULL) )
692         block_FifoRelease( p_sys->p_fifo );
693
694     if ( p_sys->flow )
695     {
696         if (p_sys->flow->fd_out >= 0) {
697             net_Close (p_sys->flow->fd_out);
698         }
699         if (p_sys->flow->fd_rtcp >= 0) {
700             net_Close (p_sys->flow->fd_rtcp);
701         }
702         if (p_sys->flow->fd_rtcp_m >= 0) {
703             net_Close (p_sys->flow->fd_rtcp_m);
704         }
705         for (int i=0; i<RIST_QUEUE_SIZE; i++) {
706             struct rtp_pkt *pkt = &(p_sys->flow->buffer[i]);
707             if (pkt->buffer)
708             {
709                 block_Release(pkt->buffer);
710                 pkt->buffer = NULL;
711             }
712         }
713         free(p_sys->flow->buffer);
714         free(p_sys->flow);
715     }
716
717     vlc_mutex_destroy( &p_sys->lock );
718     vlc_mutex_destroy( &p_sys->fd_lock );
719     if (p_sys->p_pktbuffer)
720         block_Release(p_sys->p_pktbuffer);
721 }
722
723 static void Close( vlc_object_t * p_this )
724 {
725     sout_access_out_t     *p_access = (sout_access_out_t*)p_this;
726     sout_access_out_sys_t *p_sys = p_access->p_sys;
727
728     vlc_cancel(p_sys->ristthread);
729     vlc_cancel(p_sys->senderthread);
730
731     vlc_join(p_sys->ristthread, NULL);
732     vlc_join(p_sys->senderthread, NULL);
733
734     Clean( p_access );
735 }
736
737 static int Open( vlc_object_t *p_this )
738 {
739     sout_access_out_t       *p_access = (sout_access_out_t*)p_this;
740     sout_access_out_sys_t   *p_sys = NULL;
741
742     if (var_Create ( p_access, "dst-port", VLC_VAR_INTEGER )
743      || var_Create ( p_access, "src-port", VLC_VAR_INTEGER )
744      || var_Create ( p_access, "dst-addr", VLC_VAR_STRING )
745      || var_Create ( p_access, "src-addr", VLC_VAR_STRING ) )
746     {
747         msg_Err( p_access, "Valid network information is required." );
748         return VLC_ENOMEM;
749     }
750
751     config_ChainParse( p_access, SOUT_CFG_PREFIX, ppsz_sout_options, p_access->p_cfg );
752
753     p_sys = vlc_obj_calloc( p_this, 1, sizeof( *p_sys ) );
754     if( unlikely( p_sys == NULL ) )
755         return VLC_ENOMEM;
756
757     int i_dst_port = RIST_DEFAULT_PORT;
758     char *psz_dst_addr;
759     char *psz_parser = psz_dst_addr = strdup( p_access->psz_path );
760     if( !psz_dst_addr )
761         return VLC_ENOMEM;
762
763     if ( psz_parser[0] == '[' )
764         psz_parser = strchr( psz_parser, ']' );
765
766     psz_parser = strchr( psz_parser ? psz_parser : psz_dst_addr, ':' );
767     if ( psz_parser != NULL )
768     {
769         *psz_parser++ = '\0';
770         i_dst_port = atoi( psz_parser );
771     }
772
773     vlc_mutex_init( &p_sys->lock );
774     vlc_mutex_init( &p_sys->fd_lock );
775
776     msg_Info(p_access, "Connecting RIST output to %s:%d and %s:%d", psz_dst_addr, i_dst_port, 
777         psz_dst_addr, i_dst_port+1);
778     p_sys->b_ismulticast = is_multicast_address(psz_dst_addr);
779     struct rist_flow *flow = rist_udp_transmitter(p_access, psz_dst_addr, i_dst_port,
780         p_sys->b_ismulticast);
781     free (psz_dst_addr);
782     if (!flow)
783         goto failed;
784
785     p_sys->flow = flow;
786     flow->latency = var_InheritInteger(p_access, SOUT_CFG_PREFIX "buffer-size");
787     flow->rtp_latency = rtp_get_ts(VLC_TICK_FROM_MS(flow->latency));
788     p_sys->ssrc = var_InheritInteger(p_access, SOUT_CFG_PREFIX "ssrc");
789     if (p_sys->ssrc == 0) {
790         vlc_rand_bytes(&p_sys->ssrc, 4);
791     }
792     /* Last bit of ssrc must be 0 for normal data and 1 for retries */
793     p_sys->ssrc &= ~(1 << 0);
794
795     msg_Info(p_access, "SSRC: 0x%08X", p_sys->ssrc);
796     p_sys->i_ticks_caching = VLC_TICK_FROM_MS(var_InheritInteger( p_access, 
797         SOUT_CFG_PREFIX "caching"));
798     p_sys->i_packet_size = var_InheritInteger(p_access, SOUT_CFG_PREFIX "packet-size" );
799     p_sys->p_fifo = block_FifoNew();
800     if( unlikely(p_sys->p_fifo == NULL) )
801         goto failed;
802     p_sys->p_pktbuffer = block_Alloc( p_sys->i_packet_size );
803     if( unlikely(p_sys->p_pktbuffer == NULL) )
804         goto failed;
805
806     p_sys->p_pktbuffer->i_buffer = RTP_HEADER_SIZE;
807
808     p_access->p_sys = p_sys;
809
810     if( vlc_clone(&p_sys->senderthread, ThreadSend, p_access, VLC_THREAD_PRIORITY_HIGHEST ) )
811     {
812         msg_Err(p_access, "Failed to create sender thread.");
813         goto failed;
814     }
815
816     if (vlc_clone(&p_sys->ristthread, rist_thread, p_access, VLC_THREAD_PRIORITY_INPUT))
817     {
818         msg_Err(p_access, "Failed to create worker thread.");
819         vlc_cancel(p_sys->senderthread);
820         vlc_join(p_sys->senderthread, NULL);
821         goto failed;
822     }
823
824     p_access->pf_write = Write;
825     p_access->pf_control = Control;
826
827     return VLC_SUCCESS;
828
829 failed:
830     Clean( p_access );
831     return VLC_EGENERIC;
832 }
833
834 #define CACHING_TEXT N_("RIST data output caching size (ms)")
835 #define CACHING_LONGTEXT N_( \
836     "Having this cache will guarantee that the packets going out are " \
837     "delivered at a spacing determined by the chain timestamps thus ensuring " \
838     "a near jitter free output. Be aware that this setting will also add to " \
839     "the overall latency of the stream." )
840
841 #define BUFFER_TEXT N_("RIST retry-buffer queue size (ms)")
842 #define BUFFER_LONGTEXT N_( \
843     "This must match the buffer size (latency) configured on the server side. If you " \
844     "are not sure, leave the default of 0 which will set it the maximum " \
845     "value and will use about 100MB of RAM" )
846
847 #define SSRC_TEXT N_("SSRC used in RTP output (default is random, i.e. 0)")
848 #define SSRC_LONGTEXT N_( \
849     "Use this setting to specify a known SSRC for the RTP header. This is only useful " \
850     "if your receiver acts on it. When using VLC as receiver, it is not." )
851
852 #define NAME_TEXT N_("Stream name")
853 #define NAME_LONGTEXT N_( \
854     "This Stream name will be sent to the receiver using the rist RTCP channel" )
855
856 /* Module descriptor */
857 vlc_module_begin()
858
859     set_shortname( N_("RIST") )
860     set_description( N_("RIST stream output") )
861     set_category( CAT_SOUT )
862     set_subcategory( SUBCAT_SOUT_ACO )
863
864     add_integer( SOUT_CFG_PREFIX "packet-size", RIST_TARGET_PACKET_SIZE,
865             N_("RIST target packet size (bytes)"), NULL, true )
866     add_integer( SOUT_CFG_PREFIX "caching", DEFAULT_CACHING_DELAY,
867             CACHING_TEXT, CACHING_LONGTEXT, true )
868     add_integer( SOUT_CFG_PREFIX "buffer-size", DEFAULT_BUFFER_SIZE,
869             BUFFER_TEXT, BUFFER_LONGTEXT, true )
870     add_integer( SOUT_CFG_PREFIX "ssrc", 0,
871             SSRC_TEXT, SSRC_LONGTEXT, true )
872     add_string( SOUT_CFG_PREFIX "stream-name", NULL, NAME_TEXT, NAME_LONGTEXT, true )
873
874     set_capability( "sout access", 0 )
875     add_shortcut( "rist", "tr06" )
876
877     set_callbacks( Open, Close )
878
879 vlc_module_end ()