Minimalistic RTP demux

Currently, only MP2T payload and UDP transport work.

configure.ac

@@ -313,7 +313,7 @@ case "${host_os}" in
         VLC_ADD_LDFLAGS([vlc],[-mwindows])
         VLC_ADD_LIBS([activex mozilla],[-lgdi32])
         VLC_ADD_LIBS([cdda vcdx cddax sdl_image],[-lwinmm])
-        VLC_ADD_LIBS([access_http access_mms access_udp access_tcp access_ftp access_rtmp access_output_udp access_output_shout access_output_rtmp sap slp http stream_out_standard stream_out_rtp vod_rtsp access_realrtsp telnet rc netsync gnutls growl_udp flac ts audioscrobbler lua],[-lws2_32])
+        VLC_ADD_LIBS([access_http access_mms access_udp access_tcp access_ftp access_rtmp access_output_udp access_output_shout access_output_rtmp sap slp http stream_out_standard stream_out_rtp vod_rtsp access_realrtsp rtp telnet rc netsync gnutls growl_udp flac ts audioscrobbler lua],[-lws2_32])
     fi
     if test "${SYS}" = "mingwce"; then
         # add ws2 for closesocket, select, recv

modules/demux/Modules.am

@@ -12,6 +12,7 @@ SOURCES_mkv = mkv.cpp mp4/libmp4.c mp4/drms.c
 SOURCES_live555 = live555.cpp ../access/mms/asf.c ../access/mms/buffer.c
 SOURCES_nsv = nsv.c
 SOURCES_real = real.c
+SOURCES_rtp = rtp.c rtp.h rtpsession.c
 SOURCES_ts = ts.c ../mux/mpeg/csa.c
 SOURCES_ps = ps.c ps.h
 SOURCES_mod = mod.c

modules/demux/rtp.c

+/**
+ * @file rtp.c
+ * @brief Real-Time Protocol (RTP) demux module for VLC media player
+ */
+/*****************************************************************************
+ * Copyright (C) 2001-2005 the VideoLAN team
+ * Copyright © 2007-2008 Rémi Denis-Courmont
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2.0
+ * of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ ****************************************************************************/
+
+#ifdef HAVE_CONFIG_H
+# include <config.h>
+#endif
+#include <stdarg.h>
+#include <assert.h>
+
+#include <vlc_common.h>
+#include <vlc_demux.h>
+#include <vlc_aout.h>
+#include <vlc_network.h>
+#include <vlc_plugin.h>
+
+#include <vlc_codecs.h>
+
+#include "rtp.h"
+
+#define RTP_CACHING_TEXT N_("RTP de-jitter buffer length (msec)")
+#define RTP_CACHING_LONGTEXT N_( \
+    "How long to wait for late RTP packets (and delay the performance)." )
+
+#define RTP_MAX_SRC_TEXT N_("Maximum RTP sources")
+#define RTP_MAX_SRC_LONGTEXT N_( \
+    "How many distinct active RTP sources are allowed at a time." )
+
+#define RTP_TIMEOUT_TEXT N_("RTP source timeout (sec)")
+#define RTP_TIMEOUT_LONGTEXT N_( \
+    "How long to wait for any packet before a source is expired.")
+
+#define RTP_MAX_DROPOUT_TEXT N_("Maximum RTP sequence number dropout")
+#define RTP_MAX_DROPOUT_LONGTEXT N_( \
+    "RTP packets will be discarded if they are too much ahead (i.e. in the " \
+    "future) by this many packets from the last received packet." )
+
+#define RTP_MAX_MISORDER_TEXT N_("Maximum RTP sequence number misordering")
+#define RTP_MAX_MISORDER_LONGTEXT N_( \
+    "RTP packets will be discarded if they are too far behind (i.e. in the " \
+    "past) by this many packets from the last received packet." )
+
+static int  Open (vlc_object_t *);
+static void Close (vlc_object_t *);
+
+/*
+ * Module descriptor
+ */
+vlc_module_begin ();
+    set_shortname (_("RTP"));
+    set_description (_("(Experimental) Real-Time Protocol demuxer"));
+    set_category (CAT_INPUT);
+    set_subcategory (SUBCAT_INPUT_DEMUX);
+    set_capability ("access_demux", 10);
+    set_callbacks (Open, Close);
+
+    add_integer ("rtp-caching", 1000, NULL, RTP_CACHING_TEXT,
+                 RTP_CACHING_LONGTEXT, true);
+        change_integer_range (0, 65535);
+    add_integer ("rtp-max-src", 1, NULL, RTP_MAX_SRC_TEXT,
+                 RTP_MAX_SRC_LONGTEXT, true);
+        change_integer_range (1, 255);
+    add_integer ("rtp-timeout", 5, NULL, RTP_TIMEOUT_TEXT,
+                 RTP_TIMEOUT_LONGTEXT, true);
+    add_integer ("rtp-max-dropout", 3000, NULL, RTP_MAX_DROPOUT_TEXT,
+                 RTP_MAX_DROPOUT_LONGTEXT, true);
+        change_integer_range (0, 32767);
+    add_integer ("rtp-max-misorder", 100, NULL, RTP_MAX_MISORDER_TEXT,
+                 RTP_MAX_MISORDER_LONGTEXT, true);
+        change_integer_range (0, 32767);
+
+    add_shortcut ("rtp");
+vlc_module_end ();
+
+/*
+ * TODO: so much stuff
+ * - send RTCP-RR and RTCP-BYE
+ * - dynamic payload types (need SDP parser)
+ * - multiple medias (need SDP parser, and RTCP-SR parser for lip-sync)
+ * - support for access_filter in case of stream_Demux (MPEG-TS)
+ */
+
+/*
+ * Local prototypes
+ */
+static int Demux (demux_t *);
+static int Control (demux_t *, int i_query, va_list args);
+static int extract_port (char **phost);
+
+/**
+ * Probes and initializes.
+ */
+static int Open (vlc_object_t *obj)
+{
+    demux_t *demux = (demux_t *)obj;
+
+    if (strcmp (demux->psz_access, "rtp"))
+        return VLC_EGENERIC;
+
+    char *tmp = strdup (demux->psz_path);
+    char *shost = tmp;
+    if (shost == NULL)
+        return VLC_ENOMEM;
+
+    char *dhost = strchr (shost, '@');
+    if (dhost)
+        *dhost++ = '\0';
+
+    /* Parses the port numbers */
+    int sport = 0, dport = 0;
+    sport = extract_port (&shost);
+    if (dhost != NULL)
+        dport = extract_port (&dhost);
+    if (dport == 0)
+        dport = 5004; /* avt-profile-1 port */
+
+    /* Try to connect */
+    int fd = net_OpenDgram (obj, dhost, dport, shost, sport,
+                            AF_UNSPEC, IPPROTO_UDP);
+    free (tmp);
+    if (fd == -1)
+        return VLC_EGENERIC;
+
+    /* Initializes demux */
+    demux_sys_t *p_sys = malloc (sizeof (*p_sys));
+    if (p_sys == NULL)
+        goto error;
+
+    var_Create (obj, "rtp-caching", VLC_VAR_INTEGER | VLC_VAR_DOINHERIT);
+    p_sys->max_src      = var_CreateGetInteger (obj, "rtp-max-src");
+    p_sys->timeout      = var_CreateGetInteger (obj, "rtp-timeout");
+    p_sys->max_dropout  = var_CreateGetInteger (obj, "rtp-max-dropout");
+    p_sys->max_misorder = var_CreateGetInteger (obj, "rtp-max-misorder");
+    p_sys->autodetect   = true;
+
+    demux->pf_demux   = Demux;
+    demux->pf_control = Control;
+    demux->p_sys      = p_sys;
+
+    p_sys->session = rtp_session_create (demux);
+    if (p_sys->session == NULL)
+        goto error;
+
+    p_sys->fd = fd;
+    return VLC_SUCCESS;
+
+error:
+    net_Close (fd);
+    free (p_sys);
+    return VLC_SUCCESS;
+}
+
+
+/**
+ * Releases resources
+ */
+static void Close (vlc_object_t *obj)
+{
+    demux_t *demux = (demux_t *)obj;
+    demux_sys_t *p_sys = demux->p_sys;
+
+    rtp_session_destroy (demux, p_sys->session);
+    net_Close (p_sys->fd);
+    free (p_sys);
+}
+
+
+/**
+ * Extracts port number from "[host]:port" or "host:port" strings,
+ * and remove brackets from the host name.
+ * @param phost pointer to the string upon entry,
+ * pointer to the hostname upon return.
+ * @return port number, 0 if missing.
+ */
+static int extract_port (char **phost)
+{
+    char *host = *phost, *port;
+
+    if (host[0] == '[')
+    {
+        host = *++phost; /* skip '[' */
+        port = strchr (host, ']');
+        if (port)
+            *port++ = '\0'; /* skip ']' */
+    }
+    else
+        port = strchr (host, ':');
+
+    if (port == NULL)
+        return 0;
+    *port++ = '\0'; /* skip ':' */
+    return atoi (port);
+}
+
+
+/**
+ * Control callback
+ */
+static int Control (demux_t *demux, int i_query, va_list args)
+{
+    /*demux_sys_t *p_sys  = demux->p_sys;*/
+    (void)demux;
+
+    switch (i_query)
+    {
+        case DEMUX_GET_POSITION:
+        {
+            float *v = va_arg (args, float *);
+            *v = 0.;
+            return 0;
+        }
+
+        case DEMUX_GET_LENGTH:
+        case DEMUX_GET_TIME:
+        {
+            int64_t *v = va_arg (args, int64_t *);
+            *v = 0;
+            return 0;
+        }
+
+        case DEMUX_GET_PTS_DELAY:
+        {
+            int64_t *v = va_arg (args, int64_t *);
+            *v = var_GetInteger (demux, "rtp-caching");
+            return 0;
+        }
+    }
+
+    return VLC_EGENERIC;
+}
+
+
+/**
+ * Gets a datagram from the network
+ */
+static block_t *rtp_dgram_recv (demux_t *demux, int fd)
+{
+    block_t *block = block_Alloc (0xffff);
+
+    ssize_t len = net_Read (VLC_OBJECT (demux), fd, NULL,
+                            block->p_buffer, block->i_buffer, false);
+    if (len == -1)
+    {
+        block_Release (block);
+        return NULL;
+    }
+    return block_Realloc (block, 0, len);
+}
+
+
+/*
+ * Generic packet handlers
+ */
+static void *stream_init (demux_t *demux, const char *name)
+{
+    return stream_DemuxNew (demux, name, demux->out);
+}
+
+static void stream_destroy (demux_t *demux, void *data)
+{
+    if (data)
+        stream_DemuxDelete ((stream_t *)data);
+    (void)demux;
+}
+
+/* Send a packet to a chained demuxer */
+static void stream_decode (demux_t *demux, void *data, block_t *block)
+{
+    if (data)
+        stream_DemuxSend ((stream_t *)data, block);
+    (void)demux;
+}
+
+/* PT=33
+ * MP2: MPEG TS (RFC2250, §2)
+ */
+static void *ts_init (demux_t *demux)
+{
+    return stream_init (demux, "ts");
+}
+
+
+/**
+ * Processing callback
+ */
+static int Demux (demux_t *demux)
+{
+    demux_sys_t *p_sys = demux->p_sys;
+    block_t     *block;
+
+    block = rtp_dgram_recv (demux, p_sys->fd);
+    if (block)
+    {
+        /* Not using SDP, we need to guess the payload format used */
+        if (p_sys->autodetect && block->i_buffer >= 2)
+        {
+            rtp_pt_t pt = { .init = NULL, };
+
+            switch (pt.number = (block->p_buffer[1] & 0x7f))
+            {
+              case 14:
+                msg_Dbg (demux, "detected MPEG Audio over RTP");
+                pt.frequency = 44100;
+                break;
+
+              case 32:
+                msg_Dbg (demux, "detected MPEG Video over RTP");
+                pt.frequency = 90000;
+                break;
+
+              case 33:
+                msg_Dbg (demux, "detected MPEG2 TS over RTP");
+                pt.init = ts_init;
+                pt.destroy = stream_destroy;
+                pt.decode = stream_decode;
+                pt.frequency = 90000;
+                break;
+            }
+            rtp_add_type (demux, p_sys->session, &pt);
+            p_sys->autodetect = false;
+        }
+
+        rtp_receive (demux, p_sys->session, block);
+    }
+
+    return 1;
+}
+
+
+/* Send a packet to decoder */
+#if 0
+static void pt_decode (demux_t *obj, block_t *block, rtp_pt_t *self)
+{
+    p_block->i_pts = p_block->i_dts = date_... (...);
+    es_out_Control (obj->out, ES_OUT_SET_PCR, p_block->i_pts);
+    es_out_Send (obj->out, (es_out_id_t *)*p_id, block);
+    return 0;
+}
+#endif
+
+
+#if 0
+/*
+ * Static payload types handler
+ */
+
+/* PT=14
+ * MPA: MPEG Audio (RFC2250, §3.4)
+ */
+static int pt_mpa (demux_t *obj, block_t *block, rtp_pt_t *self)
+{
+    if (block->i_buffer < 4)
+        return VLC_EGENERIC;
+
+    block->i_buffer -= 4; // 32 bits RTP/MPA header
+    block->p_buffer += 4;
+
+    return pt_demux (obj, block, self, "mpga");
+}
+
+
+/* PT=32
+ * MPV: MPEG Video (RFC2250, §3.5)
+ */
+static int pt_mpv (demux_t *obj, block_t *block, rtp_pt_t *self)
+{
+    if (block->i_buffer < 4)
+        return VLC_EGENERIC;
+
+    block->i_buffer -= 4; // 32 bits RTP/MPV header
+    block->p_buffer += 4;
+
+    if (block->p_buffer[-3] & 0x4)
+    {
+        /* MPEG2 Video extension header */
+        /* TODO: shouldn't we skip this too ? */
+    }
+
+    return pt_demux (obj, block, self, "mpgv");
+}
+
+
+#endif
+
+/*
+ * Dynamic payload type handlers
+ * Hmm, none implemented yet.
+ */

modules/demux/rtp.h

+/**
+ * @file rtp.h
+ * @brief RTP demux module shared declarations
+ */
+/*****************************************************************************
+ * Copyright © 2008 Rémi Denis-Courmont
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2.0
+ * of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ ****************************************************************************/
+
+/* RTP payload format */
+typedef struct rtp_pt_t rtp_pt_t;
+struct rtp_pt_t
+{
+    void   *(*init) (demux_t *);
+    void    (*destroy) (demux_t *, void *);
+    void    (*decode) (demux_t *, void *, block_t *);
+    uint32_t  frequency; /* RTP clock rate (Hz) */
+    uint8_t   number;
+};
+
+/* RTP session */
+typedef struct rtp_session_t rtp_session_t;
+rtp_session_t *rtp_session_create (demux_t *);
+void rtp_session_destroy (demux_t *, rtp_session_t *);
+void rtp_receive (demux_t *, rtp_session_t *, block_t *);
+int rtp_add_type (demux_t *demux, rtp_session_t *ses, const rtp_pt_t *pt);
+
+
+/* Global data */
+struct demux_sys_t
+{
+    rtp_session_t *session;
+    int           fd;
+
+    unsigned      timeout;
+    uint8_t       max_src;
+    uint16_t      max_dropout;
+    uint16_t      max_misorder;
+    bool          autodetect;
+};
+

modules/demux/rtpsession.c

+/**
+ * @file session.c
+ * @brief RTP session handling
+ */
+/*****************************************************************************
+ * Copyright © 2008 Rémi Denis-Courmont
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2.0
+ * of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ ****************************************************************************/
+
+#ifdef HAVE_CONFIG_H
+# include <config.h>
+#endif
+
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+
+#include <vlc/vlc.h>
+#include <vlc_demux.h>
+
+#include "rtp.h"
+
+typedef struct rtp_source_t rtp_source_t;
+
+/** State for a RTP session: */
+struct rtp_session_t
+{
+    rtp_source_t **srcv;
+    unsigned       srcc;
+    uint8_t        ptc;
+    rtp_pt_t      *ptv;
+};
+
+static rtp_source_t *
+rtp_source_create (demux_t *, const rtp_session_t *, uint32_t, uint16_t);
+static void
+rtp_source_destroy (demux_t *, const rtp_session_t *, rtp_source_t *);
+
+static void rtp_decode (demux_t *, const rtp_session_t *, rtp_source_t *);
+
+/**
+ * Creates a new RTP session.
+ */
+rtp_session_t *
+rtp_session_create (demux_t *demux)
+{
+    rtp_session_t *session = malloc (sizeof (*session));
+    if (session == NULL)
+        return NULL;
+
+    session->srcv = NULL;
+    session->srcc = 0;
+    session->ptc = 0;
+    session->ptv = NULL;
+
+    (void)demux;
+    return session;
+}
+
+
+/**
+ * Destroys an RTP session.
+ */
+void rtp_session_destroy (demux_t *demux, rtp_session_t *session)
+{
+    for (unsigned i = 0; i < session->srcc; i++)
+        rtp_source_destroy (demux, session, session->srcv[i]);
+
+    free (session->srcv);
+    free (session->ptv);
+    free (session);
+    (void)demux;
+}
+
+static void *no_init (demux_t *demux)
+{
+    return demux;
+}
+
+static void no_destroy (demux_t *demux, void *opaque)
+{
+    (void)demux; (void)opaque;
+}
+
+static void no_decode (demux_t *demux, void *opaque, block_t *block)
+{
+    (void)demux; (void)opaque;
+    block_Release (block);
+}
+
+/**
+ * Adds a payload type to an RTP session.
+ */
+int rtp_add_type (demux_t *demux, rtp_session_t *ses, const rtp_pt_t *pt)
+{
+    if (ses->srcc > 0)
+    {
+        msg_Err (demux, "cannot change RTP payload formats during session");
+        return EINVAL;
+    }
+
+    rtp_pt_t *ppt = realloc (ses->ptv, (ses->ptc + 1) * sizeof (rtp_pt_t));
+    if (ppt == NULL)
+        return ENOMEM;
+
+    ses->ptv = ppt;
+    ppt += ses->ptc++;
+
+    ppt->init = pt->init ? pt->init : no_init;
+    ppt->destroy = pt->destroy ? pt->destroy : no_destroy;
+    ppt->decode = pt->decode ? pt->decode : no_decode;
+    ppt->frequency = pt->frequency;
+    ppt->number = pt->number;
+
+    assert (ppt->frequency > 0); /* SIGFPE! */
+    (void)demux;
+    return 0;
+}
+
+/** State for an RTP source */
+struct rtp_source_t
+{
+    mtime_t  expiry;  /* inactivation date */
+    uint32_t ssrc;
+    uint16_t bad_seq; /* tentatively next expected sequence for resync */
+    uint16_t max_seq; /* next expected sequence */
+
+    block_t *blocks; /* re-ordered blocks queue */
+    void    *opaque[0]; /* Per-source prviate payload data */
+};
+
+/**
+ * Initializes a new RTP source within an RTP session.
+ */
+static rtp_source_t *
+rtp_source_create (demux_t *demux, const rtp_session_t *session,
+                   uint32_t ssrc, uint16_t init_seq)
+{
+    rtp_source_t *source;
+
+    source = malloc (sizeof (*source) + (sizeof (void *) * session->ptc));
+    if (source == NULL)
+        return NULL;
+
+    source->ssrc = ssrc;
+    source->max_seq = source->bad_seq = init_seq;
+    source->blocks = NULL;
+
+    /* Initializes all payload */
+    for (unsigned i = 0; i < session->ptc; i++)
+        source->opaque[i] = session->ptv[i].init (demux);
+
+    msg_Dbg (demux, "added RTP source (%08x)", ssrc);
+    return source;
+}
+
+
+/**
+ * Destroys an RTP source and its associated streams.
+ */
+static void
+rtp_source_destroy (demux_t *demux, const rtp_session_t *session,
+                    rtp_source_t *source)
+{
+    msg_Dbg (demux, "removing RTP source (%08x)", source->ssrc);
+
+    for (unsigned i = 0; i < session->ptc; i++)
+        session->ptv[i].destroy (demux, source->opaque[i]);
+    block_ChainRelease (source->blocks);
+    free (source);
+}
+
+
+static inline uint16_t rtp_seq (const block_t *block)
+{
+    assert (block->i_buffer >= 4);
+    return GetWBE (block->p_buffer + 2);
+}
+
+/**
+ * Receives an RTP packet and queues it.
+ * @param demux VLC demux object
+ * @param session RTP session receiving the packet
+ * @param block RTP packet including the RTP header
+ */
+void
+rtp_receive (demux_t *demux, rtp_session_t *session, block_t *block)
+{
+    demux_sys_t *p_sys = demux->p_sys;
+
+    /* RTP header sanity checks (see RFC 3550) */
+    if (block->i_buffer < 12)
+        goto drop;
+    if ((block->p_buffer[0] >> 6 ) != 2) /* RTP version number */
+        goto drop;
+
+    /* Remove padding if present */
+    if (block->p_buffer[0] & 0x20)
+    {
+        uint8_t padding = block->p_buffer[block->i_buffer - 1];
+        if ((padding == 0) || (block->i_buffer < (12u + padding)))
+            goto drop; /* illegal value */
+
+        block->i_buffer -= padding;
+    }
+
+    mtime_t        now = mdate ();
+    rtp_source_t  *src  = NULL;
+    const uint16_t seq  = GetWBE (block->p_buffer + 2);
+    const uint32_t ssrc = GetDWBE (block->p_buffer + 8);
+
+    /* In most case, we know this source already */
+    for (unsigned i = 0, max = session->srcc; i < max; i++)
+    {
+        rtp_source_t *tmp = session->srcv[i];
+        if (tmp->ssrc == ssrc)
+        {
+            src = tmp;
+            break;
+        }
+
+        /* RTP source garbage collection */
+        if (tmp->expiry < now)
+        {
+            rtp_source_destroy (demux, session, tmp);
+            if (--session->srcc > 0)
+                session->srcv[i] = session->srcv[session->srcc - 1];
+        }
+    }
+
+    if (src == NULL)
+    {
+        /* New source */
+        if (session->srcc >= p_sys->max_src)
+        {
+            msg_Warn (demux, "too many RTP sessions");
+            goto drop;
+        }
+
+        rtp_source_t **tab;
+        tab = realloc (session->srcv, (session->srcc + 1) * sizeof (*tab));
+        if (tab == NULL)
+            goto drop;
+        session->srcv = tab;
+
+        src = rtp_source_create (demux, session, ssrc, seq);
+        if (src == NULL)
+            goto drop;
+
+        tab[session->srcc++] = src;
+    }
+
+    /* Check sequence number */
+    /* NOTE: the sequence number is per-source,
+     * but is independent from the payload type. */
+    uint16_t delta_seq = seq - (src->max_seq + 1);
+    if ((delta_seq < 0x8000) ? (delta_seq > p_sys->max_dropout)
+                             : ((65535 - delta_seq) > p_sys->max_misorder))
+    {
+        msg_Dbg (demux, "sequence discontinuity (got: %u, expected: %u)",
+                 seq, (src->max_seq + 1) & 0xffff);
+        if (seq == ((src->bad_seq + 1) & 0xffff))
+        {
+            src->max_seq = src->bad_seq = seq;
+            msg_Warn (demux, "sequence resynchronized");
+            block_ChainRelease (src->blocks);
+            src->blocks = NULL;
+        }
+        else
+        {
+            src->bad_seq = seq;
+            goto drop;
+        }
+    }
+
+    /* Queues the block in sequence order,
+     * hence there is a single queue for all payload types. */
+    block_t **pp = &src->blocks;
+    for (block_t *prev = *pp; prev != NULL; prev = *pp)
+    {
+        if ((int16_t)(seq - rtp_seq (*pp)) >= 0)
+            break;
+        pp = &prev->p_next;
+    }
+    block->p_next = *pp;
+    *pp = block;
+
+    rtp_decode (demux, session, src);
+    return;
+
+drop:
+    block_Release (block);
+}
+
+
+static void
+rtp_decode (demux_t *demux, const rtp_session_t *session, rtp_source_t *src)
+{
+    block_t *block = src->blocks;
+
+    /* Buffer underflow? */
+    if (!block || !block->p_next || !block->p_next->p_next)
+        return;
+    /* TODO: use time rather than packet counts for buffer measurement */
+    src->blocks = block->p_next;
+    block->p_next = NULL;
+
+    /* Match the payload type */
+    const rtp_pt_t *pt = NULL;
+    void *pt_data = NULL;
+    const uint8_t   ptype = block->p_buffer[1] & 0x7F;
+
+    for (unsigned i = 0; i < session->ptc; i++)
+    {
+        if (session->ptv[i].number == ptype)
+        {
+            pt = &session->ptv[i];
+            pt_data = src->opaque[i];
+            break;
+        }
+    }
+
+    if (pt == NULL)
+    {
+        msg_Dbg (demux, "ignoring unknown payload (%"PRIu8")", ptype);
+        goto drop;
+    }
+
+    /* Computes the PTS from the RTP timestamp and payload RTP frequency.
+     * DTS is unknown. Also, while the clock frequency depends on the payload
+     * format, a single source MUST only use payloads of a chosen frequency.
+     * Otherwise it would be impossible to compute consistent timestamps. */
+    /* FIXME: handle timestamp wrap properly */
+    /* TODO: sync multiple sources sanely... */
+    const uint32_t timestamp = GetDWBE (block->p_buffer + 4);
+    block->i_pts = UINT64_C(1) * CLOCK_FREQ * timestamp / pt->frequency;
+    //msg_Dbg (demux, "pts = %"PRIu64, block->i_pts);
+
+    /* CSRC count */
+    size_t skip = 12u + (block->p_buffer[0] & 0x0F) * 4;
+
+    /* Extension header (ignored for now) */
+    if (block->p_buffer[0] & 0x10)
+    {
+        skip += 4;
+        if (block->i_buffer < skip)
+            goto drop;
+
+        skip += 4 * GetWBE (block->p_buffer + skip - 2);
+    }
+
+    if (block->i_buffer < skip)
+        goto drop;
+
+    block->p_buffer += skip;
+    block->i_buffer -= skip;
+
+    pt->decode (demux, pt_data, block);
+    return;
+
+drop:
+    block_Release (block);
+}