demux: ts: extract multiplexed metadata access units

modules/demux/mpeg/ts_metadata.c

@@ -47,16 +47,125 @@ typedef struct
 {
     es_out_t *out;
     ts_stream_t *p_stream;
-
+    block_t *p_head;
+    block_t **pp_tail;
+    uint8_t i_sequence_number;
 } Metadata_stream_processor_context_t;
 
+static void Metadata_stream_processor_Reset( ts_stream_processor_t *h )
+{
+    Metadata_stream_processor_context_t *ctx = (Metadata_stream_processor_context_t *) h->priv;
+    block_ChainRelease(ctx->p_head);
+    ctx->p_head = NULL;
+    ctx->pp_tail = &ctx->p_head;
+}
+
 static void Metadata_stream_processor_Delete( ts_stream_processor_t *h )
 {
     Metadata_stream_processor_context_t *ctx = (Metadata_stream_processor_context_t *) h->priv;
+    block_ChainRelease(ctx->p_head);
     free( ctx );
     free( h );
 }
 
+static block_t * Metadata_stream_processor_AggregateMAU( ts_stream_processor_t *h,
+                                                         uint8_t i_sequence, block_t *p_block,
+                                                         size_t i_cellsize )
+{
+    Metadata_stream_processor_context_t *ctx = (Metadata_stream_processor_context_t *) h->priv;
+
+    bool b_corrupt = ctx->p_head && i_sequence != ((ctx->i_sequence_number + 1) & 0xFF);
+
+    if( unlikely(p_block->i_buffer > i_cellsize) )
+    {
+        block_t *cell = block_Duplicate( p_block );
+        if( cell )
+        {
+            cell->i_buffer = i_cellsize;
+            block_ChainLastAppend( &ctx->pp_tail, cell );
+        }
+        p_block->i_buffer -= i_cellsize;
+        p_block->i_buffer += i_cellsize;
+    }
+    else
+    {
+        assert( p_block->i_buffer == i_cellsize );
+        block_ChainLastAppend( &ctx->pp_tail, p_block );
+        p_block = NULL;
+    }
+
+    ctx->i_sequence_number = i_sequence;
+    if( b_corrupt )
+        Metadata_stream_processor_Reset( h );
+
+    return p_block;
+}
+
+static block_t * Metadata_stream_processor_OutputMAU( ts_stream_processor_t *h )
+{
+    Metadata_stream_processor_context_t *ctx = (Metadata_stream_processor_context_t *) h->priv;
+
+    block_t *p_chain = ctx->p_head;
+    if( !p_chain )
+        return NULL;
+    ctx->p_head = NULL;
+    ctx->pp_tail = &ctx->p_head;
+    return block_ChainGather( p_chain );
+}
+
+static block_t * Metadata_stream_processor_PushMAU( ts_stream_processor_t *h,
+                                                    const ts_es_t *p_es, block_t *p_block )
+{
+    block_t *p_return_chain = NULL;
+    block_t **pp_return = &p_return_chain;
+
+    while( p_block->i_buffer >= 5 )
+    {
+        const uint8_t i_service_id = p_block->p_buffer[0];
+        const uint8_t i_sequence = p_block->p_buffer[1];
+        const uint8_t i_fragment_indication = p_block->p_buffer[2] >> 6;
+        const uint16_t i_length = GetWBE(&p_block->p_buffer[3]);
+
+        p_block->i_buffer -= 5;
+        p_block->p_buffer += 5;
+
+        if( p_block->i_buffer < i_length )
+            break;
+
+        if( i_service_id == p_es->metadata.i_service_id )
+        {
+            if( i_fragment_indication == 0x03 ) /* FULL AU */
+            {
+                Metadata_stream_processor_Reset( h ); /* flush anything that not went to last frag */
+                p_block = Metadata_stream_processor_AggregateMAU( h, i_sequence, p_block, i_length );
+            }
+            else
+            {
+                if( i_fragment_indication == 0x02 ) /* First */
+                    Metadata_stream_processor_Reset( h ); /* flush anything that not went to last frag */
+                p_block = Metadata_stream_processor_AggregateMAU( h, i_sequence, p_block, i_length );
+                if( i_fragment_indication == 0x01 ) /* Last */
+                {
+                    block_t *out = Metadata_stream_processor_OutputMAU( h );
+                    if( out )
+                        block_ChainLastAppend( &pp_return, out );
+                }
+            }
+        }
+
+        if( !p_block )
+            break;
+
+        p_block->i_buffer -= i_length;
+        p_block->i_buffer += i_length;
+    };
+
+    if( p_block )
+        block_Release( p_block );
+
+    return p_return_chain;
+}
+
 static block_t * Metadata_stream_processor_Push( ts_stream_processor_t *h, uint8_t i_stream_id, block_t *p_block )
 {
     Metadata_stream_processor_context_t *ctx = (Metadata_stream_processor_context_t *) h->priv;
@@ -91,11 +200,14 @@ ts_stream_processor_t *Metadata_stream_processor_New( ts_stream_t *p_stream, es_
     }
     ctx->out = out;
     ctx->p_stream = p_stream;
+    ctx->i_sequence_number = 0;
+    ctx->p_head = NULL;
+    ctx->pp_tail = &ctx->p_head;
 
     h->priv = ctx;
     h->pf_delete = Metadata_stream_processor_Delete;
     h->pf_push = Metadata_stream_processor_Push;
-    h->pf_reset = NULL;
+    h->pf_reset = Metadata_stream_processor_Reset;
 
     return h;
 }