Author: anthm Date: Tue Jul 18 21:22:56 2006 New Revision: 1964 Modified: freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.c freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.h Log: work in progress Modified: freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.c ============================================================================== --- freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.c (original) +++ freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.c Tue Jul 18 21:22:56 2006 @@ -79,18 +79,19 @@ }; -enum tech_flags { +typedef enum { TFLAG_OUTBOUND = (1 << 0), TFLAG_INBOUND = (1 << 1), TFLAG_IO = (1 << 2), TFLAG_BYE = (1 << 3), TFLAG_ANSWERED = (1 << 4), TFLAG_USING_CODEC = (1 << 5), + TFLAG_RTP = (1 << 6), TFLAG_READING = (1 << 10), TFLAG_WRITING = (1 << 11), TFLAG_TIMER = (1 << 20) -}; +} tech_flag_t; #define set_param(ptr, val) if ( ptr ) { free( ptr ); ptr = NULL; } if ( val ) { ptr = strdup( val ); } @@ -347,8 +348,8 @@ /* remove call from hash */ switch_core_hash_delete( profile->call_hash, tech_pvt->call_id ); - switch_set_flag( tech_pvt, TFLAG_BYE ); - switch_clear_flag( tech_pvt, TFLAG_IO ); + switch_set_flag_locked( tech_pvt, TFLAG_BYE ); + switch_clear_flag_locked( tech_pvt, TFLAG_IO ); stop_rtp( tech_pvt ); @@ -443,7 +444,7 @@ return SWITCH_STATUS_GENERR; } } - switch_set_flag( tech_pvt, TFLAG_ANSWERED ); + switch_set_flag_locked( tech_pvt, TFLAG_ANSWERED ); return SWITCH_STATUS_SUCCESS; } @@ -462,8 +463,8 @@ tech_pvt = switch_core_session_get_private( session ); assert( tech_pvt != NULL ); - switch_clear_flag( tech_pvt, TFLAG_IO ); - switch_set_flag( tech_pvt, TFLAG_BYE ); + switch_clear_flag_locked( tech_pvt, TFLAG_IO ); + switch_set_flag_locked( tech_pvt, TFLAG_BYE ); if( tech_pvt->rtp_session ) { switch_rtp_kill_socket( tech_pvt->rtp_session ); @@ -523,24 +524,20 @@ switch_core_session_add_stream( *new_session, NULL ); /* create new call */ - tech_pvt = (struct private_object *) switch_core_session_alloc(*new_session, sizeof(struct private_object)); - if (!tech_pvt) { + if (!(tech_pvt = (struct private_object *) switch_core_session_alloc(*new_session, sizeof(struct private_object)))) { switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to create new private object for call\n" ); goto err; } - pj_memset( tech_pvt, 0, sizeof(struct private_object) ); - + switch_mutex_init(&tech_pvt->flag_mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(*new_session)); + tech_pvt->profile = profile; + tech_pvt->session = *new_session; switch_core_session_set_private( *new_session, tech_pvt ); - channel = switch_core_session_get_channel( *new_session ); /* create channel name, copy profile and parse destination number (?) */ snprintf( tmp, sizeof(tmp), "pjsip/%s-%04x", outbound_profile->destination_number, rand() & 0xffff ); switch_channel_set_name( channel, tmp ); - tech_pvt->session = *new_session; - tech_pvt->profile = profile; - /* get list of all available codecs */ tech_pvt->num_codecs = switch_loadable_module_get_codecs( switch_core_session_get_pool(tech_pvt->session), tech_pvt->codecs, sizeof(tech_pvt->codecs) / sizeof(tech_pvt->codecs[0]) ); @@ -560,103 +557,163 @@ return SWITCH_STATUS_GENERR; } + static switch_status_t pjsip_read_frame(switch_core_session_t *session, switch_frame_t **frame, int timeout, - switch_io_flag_t flags, int stream_id) + switch_io_flag_t flags, int stream_id) { - struct private_object *tech_pvt; - switch_channel_t *channel; + struct private_object *tech_pvt = NULL; + size_t bytes = 0, samples = 0, frames = 0, ms = 0; + switch_channel_t *channel = NULL; + int payload = 0; + switch_time_t now, started = switch_time_now(), last_act = switch_time_now(); + unsigned int elapsed; + uint32_t hard_timeout = 60000 * 3; - assert( session != NULL ); - assert( frame != NULL ); + channel = switch_core_session_get_channel(session); + assert(channel != NULL); - tech_pvt = switch_core_session_get_private( session ); - assert( tech_pvt != NULL ); + tech_pvt = switch_core_session_get_private(session); + assert(tech_pvt != NULL); - channel = switch_core_session_get_channel( session ); - assert( channel != NULL ); + tech_pvt->read_frame.datalen = 0; + switch_set_flag_locked(tech_pvt, TFLAG_READING); - switch_set_flag( tech_pvt, TFLAG_READING ); - tech_pvt->read_frame.datalen = 0; + samples = tech_pvt->read_codec.implementation->samples_per_frame; + ms = tech_pvt->read_codec.implementation->microseconds_per_frame; + + if (tech_pvt->last_read) { + elapsed = (unsigned int)((switch_time_now() - tech_pvt->last_read) / 1000); + if (elapsed > 60000) { + return SWITCH_STATUS_TIMEOUT; + } + } - if( switch_test_flag( tech_pvt, TFLAG_IO ) ) { + if (switch_test_flag(tech_pvt, TFLAG_IO)) { switch_status_t status; - assert( tech_pvt->rtp_session != NULL ); + assert(tech_pvt->rtp_session != NULL); + tech_pvt->read_frame.datalen = 0; - while( !switch_test_flag( tech_pvt, TFLAG_BYE ) && - switch_test_flag( tech_pvt, TFLAG_IO ) && - !tech_pvt->read_frame.datalen ) - { - status = switch_rtp_zerocopy_read_frame( tech_pvt->rtp_session, &tech_pvt->read_frame ); - if( status != SWITCH_STATUS_SUCCESS && status != SWITCH_STATUS_BREAK ) { + + while (!switch_test_flag(tech_pvt, TFLAG_BYE) && switch_test_flag(tech_pvt, TFLAG_IO) && tech_pvt->read_frame.datalen == 0) { + now = switch_time_now(); + tech_pvt->read_frame.flags = 0; + + status = switch_rtp_zerocopy_read_frame(tech_pvt->rtp_session, &tech_pvt->read_frame); + if (status != SWITCH_STATUS_SUCCESS && status != SWITCH_STATUS_BREAK) { return SWITCH_STATUS_FALSE; } + + payload = tech_pvt->read_frame.payload; - if( tech_pvt->read_frame.datalen > 0 ) { - int samples, bytes, frames; + elapsed = (unsigned int)((switch_time_now() - started) / 1000); + if (timeout > -1) { + if (elapsed >= (unsigned int)timeout) { + return SWITCH_STATUS_BREAK; + } + } + + elapsed = (unsigned int)((switch_time_now() - last_act) / 1000); + + if (elapsed >= hard_timeout) { + return SWITCH_STATUS_BREAK; + } + + if (tech_pvt->read_codec.implementation->encoded_bytes_per_frame) { + bytes = tech_pvt->read_codec.implementation->encoded_bytes_per_frame; + frames = tech_pvt->read_frame.datalen / bytes; + } else { + frames = 1; + } + + samples = frames * tech_pvt->read_codec.implementation->samples_per_frame; + + + if (switch_rtp_has_dtmf(tech_pvt->rtp_session)) { + char dtmf[128]; + switch_rtp_dequeue_dtmf(tech_pvt->rtp_session, dtmf, sizeof(dtmf)); + switch_channel_queue_dtmf(channel, dtmf); + } + + + if (tech_pvt->read_frame.datalen > 0) { + tech_pvt->last_read = switch_time_now(); if (tech_pvt->read_codec.implementation->encoded_bytes_per_frame) { - bytes = tech_pvt->read_codec.implementation->encoded_bytes_per_frame; - frames = tech_pvt->read_frame.datalen / bytes; + bytes = tech_pvt->read_codec.implementation->encoded_bytes_per_frame; + frames = (tech_pvt->read_frame.datalen / bytes); } else { frames = 1; } - samples = frames * tech_pvt->read_codec.implementation->samples_per_frame; - - tech_pvt->read_frame.samples = (int)samples; + samples = frames * tech_pvt->read_codec.implementation->samples_per_frame; + ms = frames * tech_pvt->read_codec.implementation->microseconds_per_frame; + tech_pvt->read_frame.samples = (int) samples; break; } - switch_yield( 1000 ); + switch_yield(1000); } + } - switch_clear_flag( tech_pvt, TFLAG_READING ); + switch_clear_flag_locked(tech_pvt, TFLAG_READING); + + if (switch_test_flag(tech_pvt, TFLAG_BYE)) { + switch_channel_hangup(channel, SWITCH_CAUSE_NORMAL_CLEARING); + return SWITCH_STATUS_FALSE; + } + *frame = &tech_pvt->read_frame; return SWITCH_STATUS_SUCCESS; -} +} + static switch_status_t pjsip_write_frame(switch_core_session_t *session, switch_frame_t *frame, int timeout, - switch_io_flag_t flags, int stream_id) + switch_io_flag_t flags, int stream_id) { struct private_object *tech_pvt; - switch_channel_t *channel; + switch_channel_t *channel = NULL; + switch_status_t status = SWITCH_STATUS_SUCCESS; + int bytes = 0, samples = 0, frames = 0; - assert( session != NULL ); - assert( frame != NULL ); + channel = switch_core_session_get_channel(session); + assert(channel != NULL); - tech_pvt = switch_core_session_get_private( session ); - assert( tech_pvt != NULL ); + tech_pvt = switch_core_session_get_private(session); + assert(tech_pvt != NULL); - channel = switch_core_session_get_channel( session ); - assert( channel != NULL ); + if (!switch_test_flag(tech_pvt, TFLAG_IO)) { + return SWITCH_STATUS_SUCCESS; + } - switch_set_flag( tech_pvt, TFLAG_WRITING ); + if (switch_test_flag(tech_pvt, TFLAG_BYE)) { + switch_channel_hangup(channel, SWITCH_CAUSE_NORMAL_CLEARING); + return SWITCH_STATUS_FALSE; + } - if( switch_test_flag( tech_pvt, TFLAG_IO ) ) { - int samples, frames, bytes; + switch_set_flag_locked(tech_pvt, TFLAG_WRITING); - assert( tech_pvt->rtp_session != NULL ); - if (tech_pvt->read_codec.implementation->encoded_bytes_per_frame) { - bytes = tech_pvt->read_codec.implementation->encoded_bytes_per_frame; - frames = (int)frame->datalen / bytes; - } else { - frames = 1 ; - } - samples = frames * tech_pvt->read_codec.implementation->samples_per_frame; - switch_rtp_write_frame( tech_pvt->rtp_session, frame, samples ); + + if (tech_pvt->read_codec.implementation->encoded_bytes_per_frame) { + bytes = tech_pvt->read_codec.implementation->encoded_bytes_per_frame; + frames = ((int) frame->datalen / bytes); + } else { + frames = 1; } - switch_clear_flag( tech_pvt, TFLAG_WRITING ); + samples = frames * tech_pvt->read_codec.implementation->samples_per_frame; + switch_rtp_write_frame( tech_pvt->rtp_session, frame, samples ); - return SWITCH_STATUS_SUCCESS; -} + switch_clear_flag_locked(tech_pvt, TFLAG_WRITING); + return status; +} + /* END: freeswitch callbacks */ @@ -705,15 +762,21 @@ /* allocate new session */ switch_core_session_add_stream( session, NULL ); - if ((tech_pvt = (struct private_object *) switch_core_session_alloc( session, sizeof(struct private_object) ))) { - memset( tech_pvt, 0, sizeof(struct private_object) ); - switch_core_session_set_private( session, tech_pvt ); - tech_pvt->session = session; - } else { + if (!(tech_pvt = (struct private_object *) switch_core_session_alloc( session, sizeof(struct private_object) ))) { switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to create new channel for inbound call\n" ); switch_core_session_destroy( &session ); return; } + + switch_mutex_init(&tech_pvt->flag_mutex, SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session)); + + /* need the invite data in pjsip_on_init */ + tech_pvt->sip_ua = pjsip_ua_instance(); + tech_pvt->profile = profile; + + switch_core_session_set_private( session, tech_pvt ); + tech_pvt->session = session; + /* give the new session a name */ pj_strdup_with_null( pjsip_pool, &tmp, &((pjsip_sip_uri *)rxdata->msg_info.from->uri)->host ); snprintf( name, sizeof(name), "pjsip/%s/%s-%04x", profile->name, tmp.ptr, rand() & 0xffff ); @@ -721,9 +784,7 @@ switch_channel_clear_flag( channel, CF_OUTBOUND ); - /* need the invite data in pjsip_on_init */ - tech_pvt->sip_ua = pjsip_ua_instance(); - tech_pvt->profile = profile; + /* create UAS dialog */ status = pjsip_dlg_create_uas( tech_pvt->sip_ua, rxdata, NULL, &dialog ); @@ -739,8 +800,8 @@ /* load codecs */ tech_pvt->num_codecs = switch_loadable_module_get_codecs( switch_core_session_get_pool(tech_pvt->session), tech_pvt->codecs, - sizeof(tech_pvt->codecs) / sizeof(tech_pvt->codecs[0]) ); - + sizeof(tech_pvt->codecs) / sizeof(tech_pvt->codecs[0]) ); + /* Create our initial sdp */ create_sdp( dialog->pool, tech_pvt, &sdp, NULL ); @@ -782,7 +843,7 @@ /* set call id */ pjsip_strncpy( tech_pvt->call_id, &dialog->call_id->id, sizeof( tech_pvt->call_id ) ); - switch_set_flag( tech_pvt, TFLAG_INBOUND ); + switch_set_flag_locked( tech_pvt, TFLAG_INBOUND ); /* Send 100 Trying */ status = pjsip_inv_initial_answer( tech_pvt->sip_invite, rxdata, 100, NULL, NULL, &txdata ); @@ -1017,7 +1078,7 @@ switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "SDP re-negotiation received %s\n", tech_pvt->call_id ); stop_rtp( tech_pvt ); - switch_clear_flag( tech_pvt, TFLAG_IO ); + switch_clear_flag_locked( tech_pvt, TFLAG_IO ); /* destroy codecs */ if ( switch_test_flag( tech_pvt, TFLAG_USING_CODEC ) ) { @@ -1054,8 +1115,8 @@ if( status != PJ_SUCCESS ) { /* No valid codec found */ switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "SDP negotation failed, codec unknown\n" ); - switch_set_flag( tech_pvt, TFLAG_BYE ); - switch_clear_flag( tech_pvt, TFLAG_IO ); + switch_set_flag_locked( tech_pvt, TFLAG_BYE ); + switch_clear_flag_locked( tech_pvt, TFLAG_IO ); switch_channel_hangup( channel, SWITCH_CAUSE_DESTINATION_OUT_OF_ORDER ); return; } @@ -1112,7 +1173,7 @@ start_rtp( tech_pvt ); /* activate io */ - switch_set_flag( tech_pvt, TFLAG_IO ); + switch_set_flag_locked( tech_pvt, TFLAG_IO ); } static switch_status_t pjsip_receive_message(switch_core_session_t *session, switch_core_session_message_t *msg) @@ -1129,17 +1190,17 @@ switch( msg->message_id ) { case SWITCH_MESSAGE_INDICATE_BRIDGE: - switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "MSG Indicate Bridge\n" ); - if( tech_pvt->rtp_session && switch_test_flag( tech_pvt, TFLAG_TIMER ) ) { - switch_rtp_clear_flag( tech_pvt->rtp_session, SWITCH_RTP_FLAG_USE_TIMER ); - } - break; + if (tech_pvt->rtp_session && switch_test_flag(tech_pvt, TFLAG_TIMER)) { + switch_rtp_clear_flag(tech_pvt->rtp_session, SWITCH_RTP_FLAG_USE_TIMER); + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "De-activate timed RTP!\n"); + } + break; case SWITCH_MESSAGE_INDICATE_UNBRIDGE: - switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "MSG Indicate Unbridge\n" ); - if( tech_pvt->rtp_session && switch_test_flag( tech_pvt, TFLAG_TIMER ) ) { - switch_rtp_set_flag( tech_pvt->rtp_session, SWITCH_RTP_FLAG_USE_TIMER ); - } - break; + if (tech_pvt->rtp_session && switch_test_flag(tech_pvt, TFLAG_TIMER)) { + switch_rtp_set_flag(tech_pvt->rtp_session, SWITCH_RTP_FLAG_USE_TIMER); + switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Re-activate timed RTP!\n"); + } + break; case SWITCH_MESSAGE_INDICATE_PROGRESS: { pj_status_t status; @@ -1323,27 +1384,28 @@ rate, ms ); + flags = SWITCH_RTP_FLAG_RAW_WRITE | SWITCH_RTP_FLAG_USE_TIMER; tech_pvt->rtp_session = switch_rtp_new( tech_pvt->local_sdp_audio_addr, - tech_pvt->local_sdp_audio_port, - tech_pvt->remote_sdp_audio_addr, - tech_pvt->remote_sdp_audio_port, - tech_pvt->read_codec.implementation->ianacode, - 0, - tech_pvt->read_codec.implementation->microseconds_per_frame / 1000, - flags, - NULL, - &err, switch_core_session_get_pool( tech_pvt->session ) ); - + tech_pvt->local_sdp_audio_port, + tech_pvt->remote_sdp_audio_addr, + tech_pvt->remote_sdp_audio_port, + tech_pvt->read_codec.implementation->ianacode, + 0, + tech_pvt->read_codec.implementation->microseconds_per_frame, + flags, + NULL, + &err, switch_core_session_get_pool( tech_pvt->session ) ); + if( !tech_pvt->rtp_session ) { switch_channel_hangup( channel, SWITCH_CAUSE_DESTINATION_OUT_OF_ORDER ); - switch_set_flag( tech_pvt, TFLAG_BYE ); + switch_set_flag_locked( tech_pvt, TFLAG_BYE ); return SWITCH_STATUS_FALSE; } switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_DEBUG, "Starting RTP done\n" ); - switch_set_flag( tech_pvt, TFLAG_IO ); + switch_set_flag_locked( tech_pvt, TFLAG_IO); return SWITCH_STATUS_SUCCESS; } @@ -1472,6 +1534,8 @@ switch_log_printf( SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Failed to initialize pjlib\n" ); return SWITCH_STATUS_TERM; } + + pj_log_set_level(0); /* next: pjlib_util */ status = pjlib_util_init(); Modified: freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.h ============================================================================== --- freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.h (original) +++ freeswitch/branches/stkn/src/mod/endpoints/mod_pjsip/mod_pjsip.h Tue Jul 18 21:22:56 2006 @@ -35,6 +35,8 @@ int num_codecs; switch_frame_t read_frame; + switch_mutex_t *flag_mutex; + switch_time_t last_read; }; /*