rtmp_amf_flv.c

vlc源码

                free( string );
            }
            msg_Dbg( p_thread, "end of object" );
        }
        else if( *i == AMF_DATATYPE_NULL)
        {
            i++;
            msg_Dbg( p_thread, "null" );
        }
        else if( *i == AMF_DATATYPE_UNDEFINED )
        {
            i++;
            msg_Dbg( p_thread, "undefined (null)" );
        }
        else
        {
            i++;
            msg_Warn( p_thread, "undefined AMF type" );
        }
    }
    
    free( rtmp_packet->body->body );
    free( rtmp_packet->body );
    free( rtmp_packet );

    return;

error:
    free( string );
    free( tmp_rtmp_packet->body->body );
    free( tmp_rtmp_packet->body );
    free( tmp_rtmp_packet );
    free( tmp_buffer );
}

/* length header calculated automatically based on last packet in the same channel */
/* timestamps passed are always absolute */
static rtmp_packet_t *
rtmp_new_packet( rtmp_control_thread_t *p_thread, uint8_t stream_index, uint32_t timestamp, uint8_t content_type, uint32_t src_dst, rtmp_body_t *body )
{
    int interchunk_headers;
    rtmp_packet_t *rtmp_packet;

    rtmp_packet = (rtmp_packet_t *) malloc( sizeof( rtmp_packet_t ) );
    if( !rtmp_packet ) return NULL;

    interchunk_headers = body->length_body / p_thread->chunk_size_send;
    if( body->length_body % p_thread->chunk_size_send == 0 )
        interchunk_headers--;

    if( src_dst != p_thread->rtmp_headers_send[stream_index].src_dst )
    {
        p_thread->rtmp_headers_send[stream_index].timestamp = timestamp;
        p_thread->rtmp_headers_send[stream_index].length_body = body->length_body;
        p_thread->rtmp_headers_send[stream_index].content_type = content_type;
        p_thread->rtmp_headers_send[stream_index].src_dst = src_dst;
        
        rtmp_packet->length_header = 12;
    }
    else if( content_type != p_thread->rtmp_headers_send[stream_index].content_type
        || body->length_body != p_thread->rtmp_headers_send[stream_index].length_body )
    {
        p_thread->rtmp_headers_send[stream_index].timestamp_relative = 
            timestamp - p_thread->rtmp_headers_send[stream_index].timestamp;
        p_thread->rtmp_headers_send[stream_index].timestamp = timestamp;
        p_thread->rtmp_headers_send[stream_index].length_body = body->length_body;
        p_thread->rtmp_headers_send[stream_index].content_type = content_type;

        rtmp_packet->length_header = 8;
    }
    else if( timestamp != p_thread->rtmp_headers_send[stream_index].timestamp )
    {
        p_thread->rtmp_headers_send[stream_index].timestamp_relative = 
            timestamp - p_thread->rtmp_headers_send[stream_index].timestamp;
        p_thread->rtmp_headers_send[stream_index].timestamp = timestamp;

        rtmp_packet->length_header = 4;
    }
    else
    {
        rtmp_packet->length_header = 1;
    }
/*TODO: puede que no haga falta guardar el timestamp relative */
    rtmp_packet->stream_index = stream_index;
    if( rtmp_packet->length_header == 12 )
    {
        rtmp_packet->timestamp = timestamp;
        rtmp_packet->timestamp_relative = 0;
    }
    else
    {
        rtmp_packet->timestamp = timestamp;
        rtmp_packet->timestamp_relative = p_thread->rtmp_headers_send[stream_index].timestamp_relative;
    }
    rtmp_packet->length_encoded = rtmp_packet->length_header + body->length_body + interchunk_headers;
    rtmp_packet->length_body = body->length_body;
    rtmp_packet->content_type = content_type;
    rtmp_packet->src_dst = src_dst;

    rtmp_packet->body = (rtmp_body_t *) malloc( sizeof( rtmp_body_t ) );
    if( !rtmp_packet->body )
    {
       free( rtmp_packet );
       return NULL;
    }

    rtmp_packet->body->length_body = body->length_body;
    rtmp_packet->body->length_buffer = body->length_body;
    rtmp_packet->body->body = (uint8_t *) malloc( rtmp_packet->body->length_buffer * sizeof( uint8_t ) );
    if( !rtmp_packet->body->body )
    {
        free( rtmp_packet->body );
        free( rtmp_packet );
        return NULL;
    }
    memcpy( rtmp_packet->body->body, body->body, rtmp_packet->body->length_body );

    return rtmp_packet;
}

static block_t *
rtmp_new_block( rtmp_control_thread_t *p_thread, uint8_t *buffer, int32_t length_buffer )
{
    block_t *p_buffer;
    /* DOWN: p_thread->p_empty_blocks->i_depth */
    while ( block_FifoCount( p_thread->p_empty_blocks ) > MAX_EMPTY_BLOCKS )
    {
        p_buffer = block_FifoGet( p_thread->p_empty_blocks );
        block_Release( p_buffer );
    }
    /* DOWN: p_thread->p_empty_blocks->i_depth */
    if( block_FifoCount( p_thread->p_empty_blocks ) == 0 )
    {
        p_buffer = block_New( p_thread, length_buffer );
    }
    else
    {
        p_buffer = block_FifoGet( p_thread->p_empty_blocks );
        p_buffer = block_Realloc( p_buffer, 0, length_buffer );
    }

    p_buffer->i_buffer = length_buffer;

    memcpy( p_buffer->p_buffer, buffer, p_buffer->i_buffer );

    return p_buffer;
}

/* call sequence for each packet rtmp_new_packet -> rtmp_encode_packet -> send */
/* no parallelism allowed because of optimization in header length */
uint8_t *
rtmp_encode_packet( rtmp_control_thread_t *p_thread, rtmp_packet_t *rtmp_packet )
{
    uint8_t *out;
    int interchunk_headers;
    uint32_t timestamp, length_body, src_dst;
    int i, j;

    out = (uint8_t *) malloc( rtmp_packet->length_encoded * sizeof( uint8_t ) );
    if( !out ) return NULL;

    interchunk_headers = rtmp_packet->body->length_body / p_thread->chunk_size_send;
    if( rtmp_packet->body->length_body % p_thread->chunk_size_send == 0 )
        interchunk_headers--;

    if( rtmp_packet->length_header == 12 )
    {
        /* Timestamp absolute */
        timestamp = hton32( rtmp_packet->timestamp );
        memcpy( out, &timestamp, sizeof( uint32_t ) );

        src_dst = hton32( rtmp_packet->src_dst );
        memcpy( out + 8, &src_dst, sizeof( uint32_t ) );
    }

    if( rtmp_packet->length_header >= 8 )
    {
        /* Length without inter chunk headers */
        length_body = hton32( rtmp_packet->body->length_body );
        memcpy( out + 3, &length_body, sizeof( uint32_t ) );

        out[7] = rtmp_packet->content_type;
    }
    if( rtmp_packet->length_header >= 4 && rtmp_packet->length_header != 12 )
    {
        /* Timestamp relative */
        timestamp = hton32( rtmp_packet->timestamp_relative );
        memcpy( out, &timestamp, sizeof( uint32_t ) );
    }

    out[0] = rtmp_encode_header_size( (vlc_object_t *) p_thread, rtmp_packet->length_header ) + rtmp_packet->stream_index;

    /* Insert inter chunk headers */
    for(i = 0, j = 0; i < rtmp_packet->body->length_body + interchunk_headers; i++, j++)
    {
        if( j % p_thread->chunk_size_send == 0 && j != 0 )
            out[rtmp_packet->length_header + i++] = RTMP_HEADER_SIZE_1 + rtmp_packet->stream_index;
        out[rtmp_packet->length_header + i] = rtmp_packet->body->body[j];
    }

    return out;
}

static rtmp_packet_t *
rtmp_encode_onBWDone( rtmp_control_thread_t *p_thread, double number )
{
    rtmp_packet_t *rtmp_packet;
    rtmp_body_t *rtmp_body;
    uint8_t *tmp_buffer;

    /* Build onBWDone */
    rtmp_body = rtmp_body_new( -1 );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_STRING, "onBWDone" );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_STRING + strlen( "onBWDone" ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NUMBER, &number );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NUMBER );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NULL, NULL );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NULL );
    free( tmp_buffer );

    rtmp_packet = rtmp_new_packet( p_thread, RTMP_DEFAULT_STREAM_INDEX_INVOKE,
        0, RTMP_CONTENT_TYPE_INVOKE, RTMP_SRC_DST_CONNECT_OBJECT, rtmp_body );
    free( rtmp_body->body );
    free( rtmp_body );

    return rtmp_packet;
}

static rtmp_packet_t *
rtmp_encode_server_bw( rtmp_control_thread_t *p_thread, uint32_t number )
{
    rtmp_packet_t *rtmp_packet;
    rtmp_body_t *rtmp_body;

    /* Build server bw */
    rtmp_body = rtmp_body_new( -1 );

    rtmp_body_append( rtmp_body, (uint8_t *) &number, sizeof( uint32_t ) );

    rtmp_packet = rtmp_new_packet( p_thread, RTMP_DEFAULT_STREAM_INDEX_CONTROL,
        0, RTMP_CONTENT_TYPE_SERVER_BW, RTMP_SRC_DST_CONNECT_OBJECT, rtmp_body );
    free( rtmp_body->body );
    free( rtmp_body );

    return rtmp_packet;
}

static rtmp_packet_t *
rtmp_encode_NetConnection_connect_result( rtmp_control_thread_t *p_thread, double number )
{
    rtmp_packet_t *rtmp_packet;
    rtmp_body_t *rtmp_body;
    uint8_t *tmp_buffer;

    /* Build NetConnection.connect result */
    rtmp_body = rtmp_body_new( -1 );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_STRING, "_result" );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_STRING + strlen( "_result" ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NUMBER, &number );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NUMBER );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NULL, NULL );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NULL );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_OBJECT, NULL );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_OBJECT );
    free( tmp_buffer );

    tmp_buffer = amf_encode_object_variable( "level",
        AMF_DATATYPE_STRING, "status" );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_OBJECT_VARIABLE + strlen( "level" ) +
        AMF_DATATYPE_SIZE_STRING + strlen( "status" ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_object_variable( "code",
        AMF_DATATYPE_STRING, "NetConnection.Connect.Success" );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_OBJECT_VARIABLE + strlen( "code" ) +
        AMF_DATATYPE_SIZE_STRING + strlen( "NetConnection.Connect.Success" ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_object_variable( "description",
        AMF_DATATYPE_STRING, "Connection succeeded." );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_OBJECT_VARIABLE + strlen( "description" ) +
        AMF_DATATYPE_SIZE_STRING + strlen( "Connection succeeded." ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element ( AMF_DATATYPE_END_OF_OBJECT, NULL );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_END_OF_OBJECT );
    free( tmp_buffer );

    rtmp_packet = rtmp_new_packet( p_thread, RTMP_DEFAULT_STREAM_INDEX_INVOKE,
        0, RTMP_CONTENT_TYPE_INVOKE, 0, rtmp_body );
    free( rtmp_body->body );
    free( rtmp_body );

    return rtmp_packet;
}

static rtmp_packet_t *
rtmp_encode_createStream_result( rtmp_control_thread_t *p_thread, double stream_client_id, double stream_server_id )
{
    rtmp_packet_t *rtmp_packet;
    rtmp_body_t *rtmp_body;
    uint8_t *tmp_buffer;

    /* Build createStream result */
    rtmp_body = rtmp_body_new( -1 );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_STRING, "_result" );
    rtmp_body_append( rtmp_body, tmp_buffer,
        AMF_DATATYPE_SIZE_STRING + strlen( "_result" ) );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NUMBER, &stream_client_id );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NUMBER );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NULL, NULL );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NULL );
    free( tmp_buffer );

    tmp_buffer = amf_encode_element( AMF_DATATYPE_NUMBER, &stream_server_id );
    rtmp_body_append( rtmp_body, tmp_buffer, AMF_DATATYPE_SIZE_NUMBER );
    free( tmp_buffer );

    rtmp_packet = rtmp_new_packet( p_thread, RTMP_DEFAULT_STREAM_INDEX_INVOKE,
        0, RTMP_CONTENT_TYPE_INVOKE, 0, rtmp_body );
    free( rtmp_body->body );
    free( rtmp_body );

    return rtmp_packet;
}

static rtmp_packet_t *
rtmp_encode_ping_reset_stream( rtmp_control_thread_t *p_thread )
{
    rtmp_packet_t