diff options
Diffstat (limited to 'tinyDAV/src/t140/tdav_session_t140.c')
| -rw-r--r-- | tinyDAV/src/t140/tdav_session_t140.c | 1165 |
1 files changed, 1165 insertions, 0 deletions
diff --git a/tinyDAV/src/t140/tdav_session_t140.c b/tinyDAV/src/t140/tdav_session_t140.c new file mode 100644 index 0000000..cd4b17c --- /dev/null +++ b/tinyDAV/src/t140/tdav_session_t140.c @@ -0,0 +1,1165 @@ +/* +* Copyright (C) 2012 Doubango Telecom <http://www.doubango.org> +* +* This file is part of Open Source Doubango Framework. +* +* DOUBANGO 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 3 of the License, or +* (at your option) any later version. +* +* DOUBANGO 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 General Public License +* along with DOUBANGO. +* +*/ + +/**@file tdav_session_t140.c + * @brief Session for T140 protocol (RFC 4103/5194/2198) + */ +#include "tinydav/t140/tdav_session_t140.h" +#include "tinydav/t140/tdav_consumer_t140.h" +#include "tinydav/t140/tdav_producer_t140.h" +#include "tinydav/codecs/fec/tdav_codec_red.h" + +#include "tinymedia/tmedia_consumer.h" +#include "tinymedia/tmedia_producer.h" +#include "tinymedia/tmedia_defaults.h" + +#include "tinyrtp/trtp_manager.h" +#include "tinyrtp/rtp/trtp_rtp_packet.h" + +#include "tsk_debug.h" + +#include <limits.h> + +#define T140_BLOCKS_MAX_COUNT 20 /* number of blocks to buffer for packet loss detection and reordering */ +#define T140_BLOCK_MAX_TRANSMIT_COUNT 4 /* maximum number of times to retransmit a bloc when RED is ON */ + +#define T140_ZERO_WIDTH_NO_BREAK_SPACE 0xEFBBBF /* send at the begining of the T.140 session */ +#define T140_BACKSPACE 0x08 +#define T140_ESC 0x1B +#define T140_CR 0x0D +#define T140_LF 0x0A +#define T140_CR_LF 0x0D0A +#define T140_BELL 0x07 +#define T140_SOS 0x98 +#define T140_STRING_TERM 0x9C +#define T140_GRAPHIC_START 0x9B +#define T140_GRAPHIC_END 0x6D +#define T140_LOSS_CHAR_CHAR 0xFFFD +#define T140_LOSS_UTF8 0xEFBFBD + +#define T140_WAIT_FOR_MISSING_PKT_RED_OFF 500 /*Time to wait for missing pkts when RED is OFF (in ms) */ +#define T140_WAIT_FOR_MISSING_PKT_RED_ON 3000 /*Time to wait for missing pkts when RED is ON (in ms) */ +#define T140_WAIT_FOR_BUFFERING 300 /* Time to wait for buffering (T.140 blocks forming) */ +#define T140_WAIT_FOR_IDLE 10000 /* Time to wait before entering in IDLE state */ + +static const char zero_width_no_break_space[3] = +{ + (T140_ZERO_WIDTH_NO_BREAK_SPACE >> 16) & 0xFF, + (T140_ZERO_WIDTH_NO_BREAK_SPACE >> 8) & 0xFF, + T140_ZERO_WIDTH_NO_BREAK_SPACE & 0xFF +}; + +typedef struct t140_block_s +{ + TSK_DECLARE_OBJECT; + + uint8_t pt; + uint16_t seq_num; + uint32_t timestamp; + uint32_t transmit_count; // number of times this block have been transmitted + // whether the block have been forwarded to the end user for UI display or sent through the RTP channel + tsk_bool_t delivered; + // whether the block contains data skipped because too late + tsk_bool_t too_late; + struct{ + uint8_t* ptr; + tsk_size_t size; + }data; +} +t140_block_t; +#define T140_BLOCK(self) ((t140_block_t*)(self)) +extern const tsk_object_def_t *t140_block_def_t; + +static int _tdav_session_t140_producer_enc_cb(const void* callback_data, const void* buffer, tsk_size_t size); +static int _tdav_session_t140_save_outgoing_data(tdav_session_t140_t* self, const void* data_ptr, tsk_size_t data_size); +static tsk_size_t _tdav_session_t140_encap_red(tdav_session_t140_t* self, void** out_data_ptr, tsk_size_t* out_data_size, tsk_bool_t *last_data); +static int _tdav_session_t140_rtp_cb(const void* callback_data, const struct trtp_rtp_packet_s* packet); +static int _tdav_session_t140_recv_red(tdav_session_t140_t* self, const struct trtp_rtp_packet_s* packet); +static int _tdav_session_t140_recv_raw(tdav_session_t140_t* self, const void* data_ptr, tsk_size_t data_size); +static int _tdav_session_t140_decode_and_deliver_blocks(tdav_session_t140_t* self, tsk_bool_t called_from_timer); +static int _tdav_session_t140_consume_data(tdav_session_t140_t* self, enum tmedia_t140_data_type_e data_type, const void *data_ptr, tsk_size_t data_size); +static int _tdav_session_t140_timer_cb(const void* arg, tsk_timer_id_t timer_id); +static int _tdav_session_t140_pred_block_with_transmit_count_more_than_or_equal(const tsk_list_item_t *item, const void *transmit_count); + +static tsk_bool_t _tdav_session_t140_blocks_has_gap(tdav_session_t140_blocks_L_t* blocks); +static tsk_bool_t _tdav_session_t140_blocks_has_seqnum(tdav_session_t140_blocks_L_t* blocks, uint16_t seq_num); +static int _tdav_session_t140_blocks_add(tdav_session_t140_blocks_L_t* blocks, int64_t* blocks_count, t140_block_t** block, int64_t blocks_count_max); +static t140_block_t* _tdav_session_t140_block_create(uint8_t pt, uint16_t seq_num, uint32_t timestamp, tsk_bool_t delivered, const void* data_ptr, tsk_size_t data_size); + +#define T140_RESIZE_BUFFER(_in_buff_ptr, _in_buff_size, _new_size) \ +if((_in_buff_size) < (_new_size)){ \ + if(!((_in_buff_ptr) = tsk_realloc((_in_buff_ptr), (_new_size)))){ \ + TSK_DEBUG_ERROR("Failed to allocate new buffer"); \ + (_in_buff_size) = 0; \ + } \ + (_in_buff_size) = _new_size; \ +} + +static int tdav_session_t140_send_data(tmedia_session_t* self, enum tmedia_t140_data_type_e data_type, const void* data_ptr, unsigned data_size); +#define tdav_session_t140_send_data_2(self, data_type) tdav_session_t140_send_data((self), (data_type), tsk_null, 0) + +static int tdav_session_t140_set(tmedia_session_t* self, const tmedia_param_t* param) +{ + int ret = 0; + tdav_session_t140_t* t140; + + if(!self){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + if(tdav_session_av_set(TDAV_SESSION_AV(self), param) == tsk_true){ + return 0; + } + + t140 = (tdav_session_t140_t*)self; + + if(param->plugin_type == tmedia_ppt_consumer){ + TSK_DEBUG_ERROR("Not expected consumer(%s)", param->key); + } + else if(param->plugin_type == tmedia_ppt_producer){ + TSK_DEBUG_ERROR("Not expected producer_set(%s)", param->key); + } + else{ + TSK_DEBUG_ERROR("Not expected producer_set(%s)", param->key); + } + + return ret; +} + +static int tdav_session_t140_get(tmedia_session_t* self, tmedia_param_t* param) +{ + if(!self || !param){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + if(tdav_session_av_get(TDAV_SESSION_AV(self), param) == tsk_true){ + return 0; + } + + TSK_DEBUG_ERROR("(%s) not expected param key for plugin_type=%d", param->key, param->plugin_type); + return -2; +} + +static int tdav_session_t140_prepare(tmedia_session_t* self) +{ + tdav_session_av_t* base = (tdav_session_av_t*)(self); + int ret; + + if((ret = tdav_session_av_prepare(base))){ + TSK_DEBUG_ERROR("tdav_session_av_prepare(t140) failed"); + return ret; + } + + if(base->rtp_manager){ + ret = trtp_manager_set_rtp_callback(base->rtp_manager, _tdav_session_t140_rtp_cb, base); + } + + return ret; +} + +static int tdav_session_t140_start(tmedia_session_t* self) +{ + int ret; + tdav_session_t140_t* t140; + const tmedia_codec_t* best_codec; + tdav_session_av_t* base; + + if(!self){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + t140 = (tdav_session_t140_t*)self; + base = (tdav_session_av_t*)self; + + if(!(best_codec = tdav_session_av_get_best_neg_codec(base))){ + TSK_DEBUG_ERROR("No codec matched"); + return -2; + } + + TSK_OBJECT_SAFE_FREE(t140->encoder.codec); + t140->encoder.codec = tsk_object_ref((tsk_object_t*)best_codec); + + if((ret = tdav_session_av_start(base, best_codec))){ + TSK_DEBUG_ERROR("tdav_session_av_start(t140) failed"); + return ret; + } + + if((ret = tsk_timer_manager_start(t140->h_timer)) != 0){ + TSK_DEBUG_ERROR("Failed to start the timer"); + return ret; + } + + if(base->rtp_manager){ + int pt; + t140->encoder.payload_type = t140->encoder.codec->neg_format ? atoi(t140->encoder.codec->neg_format) : atoi(t140->encoder.codec->format); + pt = base->red.codec ? + ( base->red.codec->neg_format ? atoi(base->red.codec->neg_format) : atoi(base->red.codec->format) ) : + ( t140->encoder.payload_type ); + trtp_manager_set_payload_type(base->rtp_manager, pt); + } + + t140->decoder.last_seq_num = -1; + + t140->started = (ret == 0); + + if(t140->started){ + t140->encoder.timer_idle.timeout = T140_WAIT_FOR_IDLE; + t140->encoder.timer_buffering.timeout = T140_WAIT_FOR_BUFFERING; + t140->encoder.timer_idle.id = tsk_timer_manager_schedule(t140->h_timer, t140->encoder.timer_idle.timeout, _tdav_session_t140_timer_cb, t140); + TSK_DEBUG_INFO("T.140 - Start IDLE timer"); + } + + return ret; +} + +static int tdav_session_t140_pause(tmedia_session_t* self) +{ + return 0; +} + +static int tdav_session_t140_stop(tmedia_session_t* self) +{ + int ret; + ret = tsk_timer_manager_stop(TDAV_SESSION_T140(self)->h_timer); + ret = tdav_session_av_stop(TDAV_SESSION_AV(self)); + TSK_OBJECT_SAFE_FREE(TDAV_SESSION_T140(self)->encoder.codec); + TSK_OBJECT_SAFE_FREE(TDAV_SESSION_T140(self)->decoder.codec); + TDAV_SESSION_T140(self)->started = tsk_false; + return ret; +} + +static const tsdp_header_M_t* tdav_session_t140_get_lo(tmedia_session_t* self) +{ + tsk_bool_t updated = tsk_false; + const tsdp_header_M_t* ret; + tdav_session_av_t* base = TDAV_SESSION_AV(self); + + + if(!(ret = tdav_session_av_get_lo(base, &updated))){ + TSK_DEBUG_ERROR("tdav_session_av_get_lo(t140) failed"); + return tsk_null; + } + + if(updated){ + tsk_safeobj_lock(base); + TSK_OBJECT_SAFE_FREE(TDAV_SESSION_T140(self)->encoder.codec); + tsk_safeobj_unlock(base); + } + + return ret; +} + +static int tdav_session_t140_set_ro(tmedia_session_t* self, const tsdp_header_M_t* ro) +{ + int ret; + tsk_bool_t updated = tsk_false; + tdav_session_av_t* base = TDAV_SESSION_AV(self); + + if((ret = tdav_session_av_set_ro(base, ro, &updated))){ + TSK_DEBUG_ERROR("tdav_session_av_set_ro(t140) failed"); + return ret; + } + + if(updated){ + tsk_safeobj_lock(base); + TSK_OBJECT_SAFE_FREE(TDAV_SESSION_T140(self)->encoder.codec); + tsk_safeobj_unlock(base); + } + + return ret; +} + +static int tdav_session_t140_set_ondata_cbfn(tmedia_session_t* self, const void* context, tmedia_session_t140_ondata_cb_f func) +{ + tdav_session_t140_t* t140 = (tdav_session_t140_t*)self; + tdav_session_av_t* base = (tdav_session_av_t*)self; + if(!self){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + t140->cb_ondata.context = context; + t140->cb_ondata.func = func; + + if(base->consumer){ + return tdav_consumer_t140_set_ondata_cbfn(TDAV_CONSUMER_T140(base->consumer), context, func); + } + return -2; +} + +static int tdav_session_t140_send_data(tmedia_session_t* self, enum tmedia_t140_data_type_e data_type, const void* data_ptr, unsigned data_size) +{ + tdav_session_av_t* base = (tdav_session_av_t*)self; + if(!self){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + if(base->producer){ + return tdav_producer_send_data(TDAV_PRODUCER_T140(base->producer), data_type, data_ptr, data_size); + } + return -2; +} + +// RTP/RTCP callback (From the network to the consumer) +static int _tdav_session_t140_rtp_cb(const void* callback_data, const struct trtp_rtp_packet_s* packet) +{ + tdav_session_t140_t* t140 = (tdav_session_t140_t*)callback_data; + tdav_session_av_t* base = (tdav_session_av_t*)callback_data; + + if(!t140 || !packet || !packet->header){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + if(!t140->started || !base->consumer){ + TSK_DEBUG_ERROR("Not ready yet"); + return -1; + } + + if(packet->header->payload_type == base->red.payload_type){ + //static void* __red_buffer_ptr = tsk_null; // Never used + //static tsk_size_t __red_buffer_size = 0; // Never used + if(!base->red.codec){ + TSK_DEBUG_ERROR("No RED codec could be found"); + return -2; + } + // Decode RED data + /* base->red.codec->plugin->decode( + base->red.codec, + (packet->payload.data ? packet->payload.data : packet->payload.data_const), packet->payload.size, + &__red_buffer_ptr, &__red_buffer_size, + packet->header + ); + return 0; + */ + _tdav_session_t140_recv_red(t140, packet); + } + else{ + tsk_size_t out_size; + // Find the codec to use to decode the RTP payload + if(!t140->decoder.codec || t140->decoder.payload_type != packet->header->payload_type){ + tsk_istr_t format; + TSK_OBJECT_SAFE_FREE(t140->decoder.codec); + tsk_itoa(packet->header->payload_type, &format); + if(!(t140->decoder.codec = tmedia_codec_find_by_format(TMEDIA_SESSION(t140)->neg_codecs, format)) || !t140->decoder.codec->plugin || !t140->decoder.codec->plugin->decode){ + TSK_DEBUG_ERROR("%s is not a valid payload for this session", format); + return -2; + } + t140->decoder.payload_type = packet->header->payload_type; + } + + // Open codec if not already done + if(!TMEDIA_CODEC(t140->decoder.codec)->opened){ + int ret; + tsk_safeobj_lock(base); + if((ret = tmedia_codec_open(t140->decoder.codec))){ + tsk_safeobj_unlock(base); + TSK_OBJECT_SAFE_FREE(t140->decoder.codec); + TSK_DEBUG_ERROR("Failed to open [%s] codec", t140->decoder.codec->plugin->desc); + return ret; + } + tsk_safeobj_unlock(base); + } + // Decode data + out_size = t140->decoder.codec->plugin->decode(t140->decoder.codec, packet->payload.data, packet->payload.size, &t140->decoder.buffer, &t140->decoder.buffer_size, packet->header); + if(out_size){ + _tdav_session_t140_recv_raw(t140, t140->decoder.buffer, out_size); + } + } + + return 0; +} + +// Producer callback (From the producer to the network). Will encode() data before sending +static int _tdav_session_t140_producer_enc_cb(const void* callback_data, const void* buffer, tsk_size_t size) +{ + int ret = 0; + + tdav_session_t140_t* t140 = (tdav_session_t140_t*)callback_data; + tdav_session_av_t* base = (tdav_session_av_t*)callback_data; + + if(!t140){ + TSK_DEBUG_ERROR("Null session"); + return 0; + } + + if(t140->started && base->rtp_manager && base->rtp_manager->is_started && t140->encoder.codec){ + /* encode */ + //tsk_size_t out_size = 0; + + if(!base->rtp_manager->is_started){ + TSK_DEBUG_ERROR("Not started"); + return 0; + } + + // Open codec if not already done + if(!t140->encoder.codec->opened){ + tsk_safeobj_lock(base); + if((ret = tmedia_codec_open(t140->encoder.codec))){ + tsk_safeobj_unlock(base); + TSK_DEBUG_ERROR("Failed to open [%s] codec", t140->encoder.codec->plugin->desc); + return -4; + } + tsk_safeobj_unlock(base); + } + + ret = _tdav_session_t140_save_outgoing_data(t140, buffer, size); + + // start buffering timer if not already done + if(!TSK_TIMER_ID_IS_VALID(t140->encoder.timer_buffering.id)){ + TSK_DEBUG_INFO("T.140 - Schedule buffering timer"); + t140->encoder.timer_buffering.id = tsk_timer_manager_schedule(t140->h_timer, t140->encoder.timer_buffering.timeout, _tdav_session_t140_timer_cb, t140); + } + + //if((t140->encoder.codec = tsk_object_ref(t140->encoder.codec))){ /* Thread safeness (SIP reINVITE or UPDATE could update the encoder) */ + //out_size = t140->encoder.codec->plugin->encode(t140->encoder.codec, buffer, size, &t140->encoder.buffer, &t140->encoder.buffer_size); + //if(out_size){ + // trtp_manager_send_rtp(base->rtp_manager, t140->encoder.buffer, out_size, TMEDIA_CODEC_PCM_FRAME_SIZE(t140->encoder.codec), tsk_false/*Marker*/, tsk_true/*lastPacket*/); + //} + //tsk_object_unref(t140->encoder.codec); + //} + } + return ret; +} + +// save data before sending (when timer fires) +static int _tdav_session_t140_save_outgoing_data(tdav_session_t140_t* self, const void* data_ptr, tsk_size_t data_size) +{ + tdav_session_av_t* base = (tdav_session_av_t*)self; + + if(!self || !data_ptr || !data_size){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + tsk_safeobj_lock(base); + + T140_RESIZE_BUFFER(self->encoder.buffer, self->encoder.buffer_size, (self->encoder.buffer_idx + data_size)); + if(self->encoder.buffer && self->encoder.buffer_size){ + memcpy(&((uint8_t*)self->encoder.buffer)[self->encoder.buffer_idx], data_ptr, data_size); + self->encoder.buffer_idx += data_size; + } + + tsk_safeobj_unlock(base); + + return 0; +} + +// encapsulated data for sending +static tsk_size_t _tdav_session_t140_encap_red(tdav_session_t140_t* self, void** out_data_ptr, tsk_size_t* out_data_size, tsk_bool_t *last_data) +{ + tdav_session_av_t* base = (tdav_session_av_t*)self; + t140_block_t* block, *block_curr = tsk_null; + t140_block_t* red_blocks[T140_BLOCK_MAX_TRANSMIT_COUNT] = { tsk_null }; + tsk_size_t red_blocks_idx, idx, red_blocks_size, red_blocks_hdrs_size; + tsk_list_item_t* item; + uint16_t rtp_seq_num; + uint32_t* pout_data_ptr32; + uint8_t* pout_data_ptr8; + static const uint32_t max_transmit_count = T140_BLOCK_MAX_TRANSMIT_COUNT; + + if(!self || !base->rtp_manager || !out_data_ptr || !out_data_size || !last_data){ + TSK_DEBUG_ERROR("Invalid parameter"); + return 0; + } + + *last_data = tsk_false; + + tsk_safeobj_lock(base); + tsk_list_lock(self->encoder.blocks); + + // cancel IDLE timer + if(TSK_TIMER_ID_IS_VALID(self->encoder.timer_idle.id)){ + tsk_timer_manager_cancel(self->h_timer, self->encoder.timer_idle.id); + self->encoder.timer_idle.id = TSK_INVALID_TIMER_ID; + TSK_DEBUG_INFO("T.140 - Cancel IDLE timer"); + } + + rtp_seq_num = base->rtp_manager->rtp.seq_num; + + if(self->encoder.buffer && self->encoder.buffer_idx){ + block_curr = _tdav_session_t140_block_create(self->encoder.payload_type, rtp_seq_num, 0, tsk_false, self->encoder.buffer, self->encoder.buffer_idx); + self->encoder.buffer_idx = 0; + } + + // build RED blocks + red_blocks_idx = 0; + tsk_list_foreach(item, self->encoder.blocks){ + if(!(block = item->data) || (block->transmit_count + 1) >= T140_BLOCK_MAX_TRANSMIT_COUNT){ + continue; + } + + if(block->seq_num <= rtp_seq_num && (idx = (rtp_seq_num - block->seq_num)) < T140_BLOCK_MAX_TRANSMIT_COUNT){ + ++block->transmit_count; // increment transmission count + red_blocks[(T140_BLOCK_MAX_TRANSMIT_COUNT - idx - 1)] = block; // backwards counting + ++red_blocks_idx; + } + + if((red_blocks_idx + 1) >= T140_BLOCK_MAX_TRANSMIT_COUNT){ + break; + } + } + + // compute RED blocks size + red_blocks_hdrs_size = ((T140_BLOCK_MAX_TRANSMIT_COUNT - 1) << 2) + 1; + red_blocks_size = red_blocks_hdrs_size; + for(idx = 0; idx < T140_BLOCK_MAX_TRANSMIT_COUNT; ++idx){ + if(red_blocks[idx] && red_blocks[idx]->data.ptr && red_blocks[idx]->data.size){ + red_blocks_size += red_blocks[idx]->data.size; + } + } + if(block_curr && block_curr->data.ptr && block_curr->data.size){ + red_blocks_size += block_curr->data.size; + } + + // resize output buffer + T140_RESIZE_BUFFER(*out_data_ptr, *out_data_size, red_blocks_size); + if(!*out_data_ptr) { + red_blocks_size = 0; + goto bail; + } + + // compute RED headers + pout_data_ptr32 = (uint32_t*)*out_data_ptr; + for(idx = 0; idx < T140_BLOCK_MAX_TRANSMIT_COUNT; ++idx){ + if((idx + 1) == T140_BLOCK_MAX_TRANSMIT_COUNT){ // last header ? + *((uint8_t*)&pout_data_ptr32[idx]) = (self->encoder.payload_type & 0x7F); // F=0 | PT + } + else{ + pout_data_ptr32[idx] = tnet_htonl( + ( (0x80 | (self->encoder.payload_type & 0x7F)) << 24 ) | // F=1 | PT + ( 0x0000 << 8 ) | // imestamp offset + ( red_blocks[idx] ? (red_blocks[idx]->data.size & 0xFF) : 0x00 ) // block length + ); + } + } + + // compute RED data + pout_data_ptr8 = &((uint8_t*)*out_data_ptr)[red_blocks_hdrs_size]; + for(idx = 0; idx < T140_BLOCK_MAX_TRANSMIT_COUNT; ++idx){ + if(red_blocks[idx] && red_blocks[idx]->data.ptr && red_blocks[idx]->data.size){ + memcpy(pout_data_ptr8, red_blocks[idx]->data.ptr, red_blocks[idx]->data.size); + pout_data_ptr8 += red_blocks[idx]->data.size; + } + } + if(block_curr && block_curr->data.ptr && block_curr->data.size){ + memcpy(pout_data_ptr8, block_curr->data.ptr, block_curr->data.size); + pout_data_ptr8 += block_curr->data.size; + } + + if((*last_data = (red_blocks_hdrs_size == red_blocks_size))){ + // no data available -> clear items, send empty block, schedule idle timer, cancel buffering timer + if(TSK_TIMER_ID_IS_VALID(self->encoder.timer_buffering.id)){ + tsk_timer_manager_cancel(self->h_timer, self->encoder.timer_buffering.id); + self->encoder.timer_buffering.id = TSK_INVALID_TIMER_ID; + TSK_DEBUG_INFO("T.140 - Cancel buffering timer"); + } + self->encoder.timer_idle.id = tsk_timer_manager_schedule(self->h_timer, self->encoder.timer_idle.timeout, _tdav_session_t140_timer_cb, self); + TSK_DEBUG_INFO("T.140 - Schedule IDLE timer"); + tsk_list_clear_items(self->encoder.blocks); + } + else{ + // clean up all blocks retransmitted more than max times + while(tsk_list_remove_item_by_pred(self->encoder.blocks, _tdav_session_t140_pred_block_with_transmit_count_more_than_or_equal, &max_transmit_count)) ; + } + +bail: + if(block_curr){ + _tdav_session_t140_blocks_add(self->encoder.blocks, &self->encoder.blocks_count, &block_curr, INT_MAX); + TSK_OBJECT_SAFE_FREE(block_curr); + } + tsk_list_unlock(self->encoder.blocks); + tsk_safeobj_unlock(base); + + return red_blocks_size; +} + +// decode data from RTP channel +static int _tdav_session_t140_recv_red(tdav_session_t140_t* self, const struct trtp_rtp_packet_s* packet) +{ + const uint8_t* pdata; + const uint8_t* red_hdr; + tsk_size_t red_hdrs_count, i, in_size; + tsk_bool_t last, block_add; + uint8_t F, pt; + int32_t pkt_loss_start = -1, pkt_loss_count, seq_num; + uint16_t timestamp_offset, block_length; + + if(!self || !packet || !packet->header || !(packet->payload.data || packet->payload.data_const) || (packet->payload.size < TDAV_CODEC_RED_MIN_PKT_SIZE)){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + pdata = red_hdr = packet->payload.data ? packet->payload.data : packet->payload.data_const; + in_size = packet->payload.size; + + if((F = (pdata[0] & 0x80)) == 0){ + i = 1; + red_hdrs_count = 1; + } + else{ + for(i = 0, red_hdrs_count = 0; i < in_size; i+= 4, ++red_hdrs_count){ + if((F = (pdata[i] & 0x80)) == 0){ ++i; ++red_hdrs_count; break; } + } + } + + if(i == in_size){ + TSK_DEBUG_INFO("T.140 - Empty block"); + return 0; + } + + pdata += i; + in_size -= i; + if(self->decoder.last_seq_num != -1 && (self->decoder.last_seq_num + 1) != packet->header->seq_num){ + pkt_loss_start = (self->decoder.last_seq_num + 1); + pkt_loss_count = (packet->header->seq_num - pkt_loss_start); + TSK_DEBUG_INFO("T.140 - Packet loss[curr=%d, loss_start=%d, loss_count=%d]", packet->header->seq_num, pkt_loss_start, pkt_loss_count); + } + + for(i = 0, block_add = tsk_false; i < red_hdrs_count; ++i, block_add = tsk_false){ + last = (i == (red_hdrs_count - 1)); + F = (red_hdr[0] & 0x80); + pt = (red_hdr[0] & 0x7F); + if(last || !F){ + /* + 0 1 2 3 4 5 6 7 + +-+-+-+-+-+-+-+-+ + |0| Block PT | + +-+-+-+-+-+-+-+-+ + */ + block_length = (uint16_t)in_size; + seq_num = packet->header->seq_num; + timestamp_offset = 0; + block_add = tsk_true; + } + else{ + /* + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + |1| block PT=7 | timestamp offset | block length | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + */ + seq_num = (int32_t)(packet->header->seq_num - (red_hdrs_count - 1 - i)); // inferred by counting backwards + block_add = (pkt_loss_start != -1 && (seq_num <= pkt_loss_start && pkt_loss_start >= seq_num)); + timestamp_offset = ((red_hdr[1] << 8) | red_hdr[2]) >> 2; + block_length = ((red_hdr[2] & 0x03) << 8) | red_hdr[3]; + /* block_add &= (block_length > 0); */ // take empty block to avoid scheduling pkt loss timer + if(pkt_loss_start != -1){ + TSK_DEBUG_INFO("retransmit seq_num=%d, takken=%d", seq_num, block_add); + } + if(block_length > in_size){ + TSK_DEBUG_ERROR("Invalid 'block length'"); + break; + } + red_hdr += 4; + } + + if(block_add && !_tdav_session_t140_blocks_has_seqnum(self->decoder.blocks, seq_num)){ + t140_block_t* block = _tdav_session_t140_block_create( + pt, + seq_num, + (packet->header->timestamp + timestamp_offset), + tsk_false, + pdata, + block_length); + _tdav_session_t140_blocks_add(self->decoder.blocks, &self->decoder.blocks_count, &block, T140_BLOCKS_MAX_COUNT); + TSK_OBJECT_SAFE_FREE(block); + } + + + pdata += block_length; + in_size -= block_length; + } + + self->decoder.last_seq_num = packet->header->seq_num; + + return _tdav_session_t140_decode_and_deliver_blocks(self, tsk_false); +} + +static int _tdav_session_t140_recv_raw(tdav_session_t140_t* self, const void* data_ptr, tsk_size_t data_size) +{ + tsk_size_t i, j, cmd_extra_len, utf8_start, utf8_end; + tsk_bool_t is_cmd; + enum tmedia_t140_data_type_e cmd_type; + const uint8_t* _data_ptr = (const uint8_t*)data_ptr; + + if(!self || !data_ptr || !data_size){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + for(i = 0, j = 0, is_cmd = tsk_false, cmd_extra_len = 0; i < data_size; ++i, is_cmd = tsk_false, cmd_extra_len = 0){ + switch(_data_ptr[i]){ + case T140_BACKSPACE: + { + TSK_DEBUG_INFO("T.140 - BACKSPACE"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_backspace; + break; + } + case T140_ESC: + { + TSK_DEBUG_INFO("T.140 - ESC"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_esc; + break; + } + case T140_LF: + { + TSK_DEBUG_INFO("T.140 - LF"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_lf; + break; + } + case T140_CR: + { + if((i + 1) < data_size && _data_ptr[i + 1] == T140_LF){ + // CR_LF + TSK_DEBUG_INFO("T.140 - CRLF"); + cmd_extra_len = 1; + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_cr_lf; + } + else{ + // CR + TSK_DEBUG_INFO("T.140 - CR"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_cr; + } + break; + } + case T140_BELL: + { + TSK_DEBUG_INFO("T.140 - BELL"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_bell; + break; + } + case T140_SOS: + { + TSK_DEBUG_INFO("T.140 - SOS"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_sos; + break; + } + case T140_STRING_TERM: + { + TSK_DEBUG_INFO("T.140 - ST"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_string_term; + break; + } + case (T140_LOSS_CHAR_CHAR >> 8): + { + if((i + 1) < data_size && _data_ptr[i + 1] == (T140_LOSS_CHAR_CHAR & 0xFF)){ + TSK_DEBUG_INFO("T.140 - LOSS_CHAR_CHAR"); + cmd_extra_len = 1; + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_loss_char_char; + } + break; + } + case T140_GRAPHIC_START: + { + TSK_DEBUG_INFO("T.140 - GRAPHIC_START"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_graphic_start; + break; + } + case T140_GRAPHIC_END: + { + TSK_DEBUG_INFO("T.140 - GRAPHIC_END"); + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_graphic_end; + break; + } + // case (T140_ZERO_WIDTH_NO_BREAK_SPACE >> 16): + // case (T140_LOSS_UTF8 >> 16): + case 0xEF: + { + if((i + 2) < data_size && _data_ptr[i + 1] == ((T140_ZERO_WIDTH_NO_BREAK_SPACE >> 8) & 0xFF) && _data_ptr[i + 2] == (T140_ZERO_WIDTH_NO_BREAK_SPACE & 0xFF)){ + TSK_DEBUG_INFO("T.140 - ZERO_WIDTH_NO_BREAK_SPACE"); + cmd_extra_len = 2; + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_zero_width_no_break_space; + } + else if((i + 2) < data_size && _data_ptr[i + 1] == ((T140_LOSS_UTF8 >> 8) & 0xFF) && _data_ptr[i + 2] == (T140_LOSS_UTF8 & 0xFF)){ + TSK_DEBUG_INFO("T.140 - LOSS_UTF8"); + cmd_extra_len = 2; + is_cmd = tsk_true; + cmd_type = tmedia_t140_data_type_loss_utf8; + } + break; + } + }//switch + + if(is_cmd || ((i + 1) >= data_size)){ + utf8_start = j; + utf8_end = utf8_start + ( (j < i || (j == i && !is_cmd)) ? (i - j + 1) : 0 ); + if(utf8_start < utf8_end){ // there is utf8 data to deliver + _tdav_session_t140_consume_data(self, tmedia_t140_data_type_utf8, &_data_ptr[utf8_start], (utf8_end - utf8_start)); + } + if(is_cmd){ // there is cmd data to deliver + i += cmd_extra_len; + _tdav_session_t140_consume_data(self, cmd_type, tsk_null, 0); + } + j = i; + } + }// for + + return 0; +} + +static int _tdav_session_t140_decode_and_deliver_blocks(tdav_session_t140_t* self, tsk_bool_t called_from_timer) +{ + tdav_session_av_t* base = (tdav_session_av_t*)self; + if(!self){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + tsk_safeobj_lock(base); + tsk_list_lock(self->decoder.blocks); + + if(!called_from_timer && _tdav_session_t140_blocks_has_gap(self->decoder.blocks)){ + self->decoder.timer_pkt_loss.timeout = (base->red.codec ? T140_WAIT_FOR_MISSING_PKT_RED_ON : T140_WAIT_FOR_MISSING_PKT_RED_OFF); + self->decoder.timer_pkt_loss.id = tsk_timer_manager_schedule(self->h_timer, self->decoder.timer_pkt_loss.timeout, _tdav_session_t140_timer_cb, self); + } + else{ + const tsk_list_item_t* item; + const t140_block_t* block; + + tsk_list_foreach(item, self->decoder.blocks){ + if(!(block = (const t140_block_t*)item->data) || block->delivered || block->too_late){ + continue; + } + if(block->data.ptr && block->data.size){ + _tdav_session_t140_recv_raw(self, block->data.ptr, block->data.size); + } + ((t140_block_t*)block)->delivered = tsk_true; + } + } + + tsk_list_unlock(self->decoder.blocks); + tsk_safeobj_unlock(base); + + return 0; +} + +static int _tdav_session_t140_consume_data(tdav_session_t140_t* self, enum tmedia_t140_data_type_e data_type, const void *data_ptr, tsk_size_t data_size) +{ + tdav_session_av_t* base = (tdav_session_av_t*)self; + if(!self || !base->consumer){ + TSK_DEBUG_ERROR("Invalid parameter"); + return -1; + } + + T140_RESIZE_BUFFER(self->decoder.consumer_buffer, self->decoder.consumer_buffer_size, (data_size + 4)); + if(self->decoder.consumer_buffer){ + *((int32_t*)self->decoder.consumer_buffer) = data_type; + if(data_ptr && data_size){ + memcpy(&((uint8_t*)self->decoder.consumer_buffer)[4], data_ptr, data_size); + } + tmedia_consumer_consume(base->consumer, self->decoder.consumer_buffer, (data_size + 4), tsk_null); + } + return 0; +} + +static int _tdav_session_t140_pred_block_with_transmit_count_more_than_or_equal(const tsk_list_item_t *item, const void *transmit_count) +{ + if(item && item->data){ + t140_block_t *block = item->data; + return (block->transmit_count >= *((uint32_t*)transmit_count)) ? 0 : -1; + } + return -1; +} + +static int _tdav_session_t140_timer_cb(const void* arg, tsk_timer_id_t timer_id) +{ + tdav_session_t140_t* t140 = (tdav_session_t140_t*)arg; + tdav_session_av_t* base = (tdav_session_av_t*)arg; + static const tsk_bool_t called_from_timer = tsk_true; + int ret = 0; + + tsk_safeobj_lock(base); + + if(timer_id == t140->decoder.timer_pkt_loss.id){ + ret = _tdav_session_t140_decode_and_deliver_blocks(t140, called_from_timer); + } + else if(timer_id == t140->encoder.timer_idle.id){ + ret = tdav_session_t140_send_data_2(TMEDIA_SESSION(t140), tmedia_t140_data_type_zero_width_no_break_space); + t140->encoder.timer_idle.id = tsk_timer_manager_schedule(t140->h_timer, t140->encoder.timer_idle.timeout, _tdav_session_t140_timer_cb, t140); + } + else if(timer_id == t140->encoder.timer_buffering.id){ + void* buffer_ptr; + tsk_size_t buffer_size; + tsk_bool_t was_idle = TSK_TIMER_ID_IS_VALID(t140->encoder.timer_idle.id); + tsk_bool_t last_data = tsk_false; + if(base->red.codec){ + // encapsulated red data and schedule timers if required + buffer_size = _tdav_session_t140_encap_red(t140, &t140->encoder.red_buffer, &t140->encoder.red_buffer_size, &last_data); + buffer_ptr = t140->encoder.red_buffer; + } + else{ + // send buffered data + buffer_ptr = t140->encoder.buffer; + buffer_size = t140->encoder.buffer_idx; + last_data = tsk_true; + t140->encoder.buffer_idx = 0; + } + if(buffer_ptr && buffer_size){ + trtp_manager_send_rtp(base->rtp_manager, buffer_ptr, buffer_size, 300, was_idle, last_data); + } + t140->encoder.timer_buffering.id = last_data + ? TSK_INVALID_TIMER_ID + : tsk_timer_manager_schedule(t140->h_timer, t140->encoder.timer_buffering.timeout, _tdav_session_t140_timer_cb, t140); + } + + tsk_safeobj_unlock(base); + + return ret; +} + + +//================================================================================================= +// Session Audio Plugin object definition +// +/* constructor */ +static tsk_object_t* tdav_session_t140_ctor(tsk_object_t * self, va_list * app) +{ + tdav_session_t140_t *t140 = self; + if(t140){ + int ret; + tdav_session_av_t *base = TDAV_SESSION_AV(self); + + /* init() base */ + if((ret = tdav_session_av_init(base, tmedia_t140)) != 0){ + TSK_DEBUG_ERROR("tdav_session_av_init(t140) failed"); + return tsk_null; + } + + /* init() self */ + t140->encoder.timer_buffering.id = t140->decoder.timer_pkt_loss.id = TSK_INVALID_TIMER_ID; + if(base->producer){ + tmedia_producer_set_enc_callback(base->producer, _tdav_session_t140_producer_enc_cb, t140); + } + if(base->consumer){ + } + if(!(t140->encoder.blocks = tsk_list_create())){ + TSK_DEBUG_ERROR("Failed to create encoder blocks list"); + return tsk_null; + } + if(!(t140->decoder.blocks = tsk_list_create())){ + TSK_DEBUG_ERROR("Failed to create decoder blocks list"); + return tsk_null; + } + if(!(t140->h_timer = tsk_timer_manager_create())){ + TSK_DEBUG_ERROR("Failed to create timer manager"); + return tsk_null; + } + } + return self; +} +/* destructor */ +static tsk_object_t* tdav_session_t140_dtor(tsk_object_t * self) +{ + tdav_session_t140_t *t140 = self; + if(t140){ + tdav_session_t140_stop((tmedia_session_t*)t140); + TSK_OBJECT_SAFE_FREE(t140->h_timer); + // Do it in this order (deinit self first) + TSK_OBJECT_SAFE_FREE(t140->encoder.codec); + TSK_FREE(t140->encoder.buffer); + TSK_FREE(t140->encoder.red_buffer); + TSK_OBJECT_SAFE_FREE(t140->encoder.blocks); + TSK_OBJECT_SAFE_FREE(t140->decoder.codec); + TSK_FREE(t140->decoder.buffer); + TSK_FREE(t140->decoder.consumer_buffer); + TSK_OBJECT_SAFE_FREE(t140->decoder.blocks); + + /* deinit base */ + tdav_session_av_deinit(TDAV_SESSION_AV(self)); + } + + return self; +} +/* object definition */ +static const tsk_object_def_t tdav_session_t140_def_s = +{ + sizeof(tdav_session_t140_t), + tdav_session_t140_ctor, + tdav_session_t140_dtor, + tmedia_session_cmp, +}; +/* plugin definition*/ +static const tmedia_session_plugin_def_t tdav_session_t140_plugin_def_s = +{ + &tdav_session_t140_def_s, + + tmedia_t140, + "text", + + tdav_session_t140_set, + tdav_session_t140_get, + tdav_session_t140_prepare, + tdav_session_t140_start, + tdav_session_t140_pause, + tdav_session_t140_stop, + + /* Audio part */ + {0}, + + tdav_session_t140_get_lo, + tdav_session_t140_set_ro, + + /* T.140 part */ + { + tdav_session_t140_set_ondata_cbfn, + tdav_session_t140_send_data + } +}; +const tmedia_session_plugin_def_t *tdav_session_t140_plugin_def_t = &tdav_session_t140_plugin_def_s; + +//================================================================================================= +// T140Block object definition +// +static tsk_object_t* t140_block_ctor(tsk_object_t * self, va_list * app) +{ + t140_block_t *block = self; + if(block){ + } + return self; +} +static tsk_object_t* t140_block_dtor(tsk_object_t * self) +{ + t140_block_t *block = self; + if(block){ + TSK_FREE(block->data.ptr); + } + + return self; +} +static int t140_block_cmp(const tsk_object_t *_b1, const tsk_object_t *_b2) +{ + const t140_block_t *b1 = _b1; + const t140_block_t *b2 = _b2; + + if(b1 && b2){ + return (int)(b1->seq_num - b2->seq_num); + } + else if(!b1 && !b2) return 0; + else return -1; +} +static const tsk_object_def_t t140_block_def_s = +{ + sizeof(t140_block_t), + t140_block_ctor, + t140_block_dtor, + t140_block_cmp, +}; +const tsk_object_def_t *t140_block_def_t = &t140_block_def_s; + + +static tsk_bool_t _tdav_session_t140_blocks_has_gap(tdav_session_t140_blocks_L_t* blocks) +{ + if(blocks && !TSK_LIST_IS_EMPTY(blocks)){ + const tsk_list_item_t* item; + int32_t last_seq_num = -1; + tsk_bool_t has_gap = tsk_false; + tsk_list_lock(blocks); + tsk_list_foreach(item, blocks){ + if(last_seq_num > 0 && ((last_seq_num + 1) != T140_BLOCK(item->data)->seq_num)){ + has_gap = tsk_true; + } + last_seq_num = T140_BLOCK(item->data)->seq_num; + if(has_gap){ + break; + } + } + tsk_list_unlock(blocks); + return has_gap; + } + return tsk_false; +} + +static tsk_bool_t _tdav_session_t140_blocks_has_seqnum(tdav_session_t140_blocks_L_t* blocks, uint16_t seq_num) +{ + if(blocks){ + const tsk_list_item_t* item; + tsk_bool_t has_seqnum = tsk_false; + tsk_list_lock(blocks); + tsk_list_foreach(item, blocks){ + if(seq_num == T140_BLOCK(item->data)->seq_num){ + has_seqnum = tsk_true; + break; + } + } + tsk_list_unlock(blocks); + return has_seqnum; + } + return tsk_false; +} + +static int _tdav_session_t140_blocks_add(tdav_session_t140_blocks_L_t* blocks, int64_t* blocks_count, t140_block_t** block, int64_t blocks_count_max) +{ + if(blocks && blocks_count && block){ + tsk_list_lock(blocks); + if(tsk_list_push_ascending_data(blocks, (void**)block) == 0){ + if((*blocks_count = (*blocks_count) + 1) > blocks_count_max){ + tsk_list_item_t* first_item = tsk_list_pop_first_item(blocks); + TSK_OBJECT_SAFE_FREE(first_item); + } + } + tsk_list_unlock(blocks); + } + return 0; +} + +static t140_block_t* _tdav_session_t140_block_create(uint8_t pt, uint16_t seq_num, uint32_t timestamp, tsk_bool_t delivered, const void* data_ptr, tsk_size_t data_size) +{ + void* _data_ptr = tsk_null; + t140_block_t* block = tsk_null; + + if(data_ptr && data_size){ + if(!(_data_ptr = tsk_malloc(data_size))){ + TSK_DEBUG_ERROR("Failed to alloc data with size = %u", data_size); + goto bail; + } + memcpy(_data_ptr, data_ptr, data_size); + } + if(!(block = tsk_object_new(t140_block_def_t))){ + TSK_DEBUG_ERROR("Failed to create new block object"); + goto bail; + } + if(_data_ptr){ + block->data.ptr = _data_ptr, _data_ptr = tsk_null; + block->data.size = data_size; + } + block->pt = pt; + block->seq_num = seq_num; + block->timestamp = timestamp; + +bail: + TSK_FREE(_data_ptr); + return block; +}
\ No newline at end of file |
