@ -36,7 +36,6 @@
# include "asterisk/bridge.h" # include "asterisk/bridge.h"
# include "asterisk/translate.h" # include "asterisk/translate.h"
# include "asterisk/stasis.h" # include "asterisk/stasis.h"
# include "asterisk/stasis_cache_pattern.h"
# include "asterisk/stasis_channels.h" # include "asterisk/stasis_channels.h"
# include "asterisk/dial.h" # include "asterisk/dial.h"
# include "asterisk/linkedlists.h" # include "asterisk/linkedlists.h"
@ -117,60 +116,83 @@
# define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7 # define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7
static struct stasis_ cp_all * channel_cache _all; static struct stasis_ topic * channel_topic _all;
static struct stasis_cache * channel_cache_by_nam e; static struct ao2_container * channel_cach e;
static struct stasis_caching_topic * channel_by_name_topic ; static struct ao2_container * channel_cache_by_name ;
struct stasis_cp_all * ast_channel_cache_all ( void ) struct ao2_container * ast_channel_cache_all ( void )
{ {
return channel_cache_all ;
return ao2_bump ( channel_cache ) ;
}
struct stasis_cache * ast_channel_cache ( void )
{
return stasis_cp_all_cache ( channel_cache_all ) ;
} }
struct stasis_topic * ast_channel_topic_all ( void ) struct stasis_topic * ast_channel_topic_all ( void )
{ {
return stasis_cp_all_topic ( channel_cache_all ) ;
return channel_topic_all ;
} }
struct stasis_topic * ast_channel_topic_all_cached ( void ) struct ao2_container * ast_channel_cache_by_name ( void )
{ {
return stasis_cp_all_topic_cached( channel_cache_all ) ;
return ao2_bump( channel_cache_by_name ) ;
} }
struct stasis_cache * ast_channel_cache_by_name ( void ) /*!
* \ internal
* \ brief Hash function for \ ref ast_channel_snapshot objects
*/
static int channel_snapshot_hash_cb ( const void * obj , const int flags )
{ {
return channel_cache_by_name ;
const struct ast_channel_snapshot * object = obj ;
} const char * key ;
static const char * channel_snapshot_get_id ( struct stasis_message * message ) switch ( flags & OBJ_SEARCH_MASK ) {
{ case OBJ_SEARCH_KEY :
struct ast_channel_snapshot * snapshot ;
key = obj ;
if ( ast_channel_snapshot_type ( ) ! = stasis_message_type ( message ) ) {
break ;
return NULL ;
case OBJ_SEARCH_OBJECT :
key = object - > name ;
break ;
default :
ast_assert ( 0 ) ;
return 0 ;
}
}
snapshot = stasis_message_data ( message ) ;
return ast_str_case_hash ( key ) ;
return snapshot - > uniqueid ;
} }
static const char * channel_snapshot_get_name ( struct stasis_message * message ) /*!
* \ internal
* \ brief Comparison function for \ ref ast_channel_snapshot objects
*/
static int channel_snapshot_cmp_cb ( void * obj , void * arg , int flags )
{ {
struct ast_channel_snapshot * snapshot ;
const struct ast_channel_snapshot * object_left = obj ;
if ( ast_channel_snapshot_type ( ) ! = stasis_message_type ( message ) ) {
const struct ast_channel_snapshot * object_right = arg ;
return NULL ;
const char * right_key = arg ;
int cmp ;
switch ( flags & OBJ_SEARCH_MASK ) {
case OBJ_SEARCH_OBJECT :
right_key = object_right - > name ;
case OBJ_SEARCH_KEY :
cmp = strcasecmp ( object_left - > name , right_key ) ;
break ;
case OBJ_SEARCH_PARTIAL_KEY :
cmp = strncasecmp ( object_left - > name , right_key , strlen ( right_key ) ) ;
break ;
default :
cmp = 0 ;
break ;
}
}
snapshot = stasis_message_data ( message ) ;
if ( cmp ) {
return snapshot - > name ;
return 0 ;
}
return CMP_MATCH ;
} }
/*!
/*!
* \ internal
* \ internal
* \ brief Hash function for \ ref ast_channel_snapshot objects
* \ brief Hash function ( using uniqueid ) for \ ref ast_channel_snapshot objects
*/
*/
static int channel_snapshot_hash_cb ( const void * obj , const int flags ) static int channel_snapshot_ uniqueid_ hash_cb( const void * obj , const int flags )
{ {
const struct ast_channel_snapshot * object = obj ;
const struct ast_channel_snapshot * object = obj ;
const char * key ;
const char * key ;
@ -180,7 +202,7 @@ static int channel_snapshot_hash_cb(const void *obj, const int flags)
key = obj ;
key = obj ;
break ;
break ;
case OBJ_SEARCH_OBJECT :
case OBJ_SEARCH_OBJECT :
key = object - > name ;
key = object - > uniqueid ;
break ;
break ;
default :
default :
ast_assert ( 0 ) ;
ast_assert ( 0 ) ;
@ -191,9 +213,9 @@ static int channel_snapshot_hash_cb(const void *obj, const int flags)
/*!
/*!
* \ internal
* \ internal
* \ brief Comparison function for \ ref ast_channel_snapshot objects
* \ brief Comparison function ( using uniqueid ) for \ ref ast_channel_snapshot objects
*/
*/
static int channel_snapshot_ cmp_cb( void * obj , void * arg , int flags ) static int channel_snapshot_ uniqueid_ cmp_cb( void * obj , void * arg , int flags )
{ {
const struct ast_channel_snapshot * object_left = obj ;
const struct ast_channel_snapshot * object_left = obj ;
const struct ast_channel_snapshot * object_right = arg ;
const struct ast_channel_snapshot * object_right = arg ;
@ -202,12 +224,12 @@ static int channel_snapshot_cmp_cb(void *obj, void *arg, int flags)
switch ( flags & OBJ_SEARCH_MASK ) {
switch ( flags & OBJ_SEARCH_MASK ) {
case OBJ_SEARCH_OBJECT :
case OBJ_SEARCH_OBJECT :
right_key = object_right - > name ;
right_key = object_right - > uniqueid ;
case OBJ_SEARCH_KEY :
case OBJ_SEARCH_KEY :
cmp = strcasecmp ( object_left - > name , right_key ) ;
cmp = strcasecmp ( object_left - > uniqueid , right_key ) ;
break ;
break ;
case OBJ_SEARCH_PARTIAL_KEY :
case OBJ_SEARCH_PARTIAL_KEY :
cmp = strncasecmp ( object_left - > name , right_key , strlen ( right_key ) ) ;
cmp = strncasecmp ( object_left - > uniqueid , right_key , strlen ( right_key ) ) ;
break ;
break ;
default :
default :
cmp = 0 ;
cmp = 0 ;
@ -309,6 +331,34 @@ struct ast_channel_snapshot *ast_channel_snapshot_create(struct ast_channel *cha
return snapshot ;
return snapshot ;
} }
static void channel_snapshot_update_dtor ( void * obj )
{
struct ast_channel_snapshot_update * update = obj ;
ao2_cleanup ( update - > old_snapshot ) ;
ao2_cleanup ( update - > new_snapshot ) ;
}
static struct ast_channel_snapshot_update * channel_snapshot_update_create ( struct ast_channel * chan )
{
struct ast_channel_snapshot_update * update ;
update = ao2_alloc_options ( sizeof ( * update ) , channel_snapshot_update_dtor ,
AO2_ALLOC_OPT_LOCK_NOLOCK ) ;
if ( ! update ) {
return NULL ;
}
update - > old_snapshot = ao2_bump ( ast_channel_snapshot ( chan ) ) ;
update - > new_snapshot = ast_channel_snapshot_create ( chan ) ;
if ( ! update - > new_snapshot ) {
ao2_ref ( update , - 1 ) ;
return NULL ;
}
return update ;
}
static void publish_message_for_channel_topics ( struct stasis_message * message , struct ast_channel * chan ) static void publish_message_for_channel_topics ( struct stasis_message * message , struct ast_channel * chan )
{ {
if ( chan ) {
if ( chan ) {
@ -521,7 +571,7 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
return NULL ;
return NULL ;
}
}
snapshot = chan ? a st_channel_snapshot_create ( chan ) : NULL ;
snapshot = chan ? a o2_bump( a st_channel_snapshot( chan ) ) : NULL ;
msg = create_channel_blob_message ( snapshot , type , blob ) ;
msg = create_channel_blob_message ( snapshot , type , blob ) ;
ao2_cleanup ( snapshot ) ;
ao2_cleanup ( snapshot ) ;
return msg ;
return msg ;
@ -628,38 +678,48 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
struct ast_channel_snapshot * ast_channel_snapshot_get_latest ( const char * uniqueid ) struct ast_channel_snapshot * ast_channel_snapshot_get_latest ( const char * uniqueid )
{ {
struct stasis_message * message ;
struct ast_channel_snapshot * snapshot ;
ast_assert ( ! ast_strlen_zero ( uniqueid ) ) ;
ast_assert ( ! ast_strlen_zero ( uniqueid ) ) ;
message = stasis_cache_get ( ast_channel_cache ( ) , ast_channel_snapshot_type ( ) ,
return ao2_find ( channel_cache , uniqueid , OBJ_SEARCH_KEY ) ;
uniqueid ) ;
if ( ! message ) {
return NULL ;
}
snapshot = ao2_bump ( stasis_message_data ( message ) ) ;
ao2_ref ( message , - 1 ) ;
return snapshot ;
} }
struct ast_channel_snapshot * ast_channel_snapshot_get_latest_by_name ( const char * name ) struct ast_channel_snapshot * ast_channel_snapshot_get_latest_by_name ( const char * name )
{ {
ast_assert ( ! ast_strlen_zero ( name ) ) ;
return ao2_find ( channel_cache_by_name , name , OBJ_SEARCH_KEY ) ;
}
void ast_channel_publish_final_snapshot ( struct ast_channel * chan )
{
struct ast_channel_snapshot_update * update ;
struct stasis_message * message ;
struct stasis_message * message ;
struct ast_channel_snapshot * snapshot ;
ast_assert ( ! ast_strlen_zero ( name ) ) ;
if ( ! ast_channel_snapshot_type ( ) ) {
return ;
}
message = stasis_cache_get ( ast_channel_cache_by_name ( ) , ast_channel_snapshot_type ( ) ,
update = channel_snapshot_update_create ( chan ) ;
name ) ;
if ( ! update ) {
return ;
}
message = stasis_message_create ( ast_channel_snapshot_type ( ) , update ) ;
/* In the success path message holds a reference to update so it will be valid
* for the lifetime of this function until the end .
*/
ao2_ref ( update , - 1 ) ;
if ( ! message ) {
if ( ! message ) {
return NULL ;
return ;
}
}
snapshot = ao2_bump ( stasis_message_data ( message ) ) ;
ao2_unlink ( channel_cache , update - > old_snapshot ) ;
ao2_unlink ( channel_cache_by_name , update - > old_snapshot ) ;
ast_channel_snapshot_set ( chan , NULL ) ;
stasis_publish ( ast_channel_topic ( chan ) , message ) ;
ao2_ref ( message , - 1 ) ;
ao2_ref ( message , - 1 ) ;
return snapshot ;
} }
static void channel_role_snapshot_dtor ( void * obj ) static void channel_role_snapshot_dtor ( void * obj )
@ -764,7 +824,7 @@ void ast_channel_stage_snapshot_done(struct ast_channel *chan)
void ast_channel_publish_snapshot ( struct ast_channel * chan ) void ast_channel_publish_snapshot ( struct ast_channel * chan )
{ {
struct ast_channel_snapshot * snapshot ;
struct ast_channel_snapshot _update * update ;
struct stasis_message * message ;
struct stasis_message * message ;
if ( ! ast_channel_snapshot_type ( ) ) {
if ( ! ast_channel_snapshot_type ( ) ) {
@ -775,17 +835,40 @@ void ast_channel_publish_snapshot(struct ast_channel *chan)
return ;
return ;
}
}
snapshot = ast_channel_snapshot _create( chan ) ;
update = channel_snapshot_update _create( chan ) ;
if ( ! snapshot ) {
if ( ! update ) {
return ;
return ;
}
}
message = stasis_message_create ( ast_channel_snapshot_type ( ) , snapshot ) ;
message = stasis_message_create ( ast_channel_snapshot_type ( ) , update ) ;
ao2_ref ( snapshot , - 1 ) ;
/* In the success path message holds a reference to update so it will be valid
* for the lifetime of this function until the end .
*/
ao2_ref ( update , - 1 ) ;
if ( ! message ) {
if ( ! message ) {
return ;
return ;
}
}
/* We lock these ourselves so that the update is atomic and there isn't time where a
* snapshot is not in the cache .
*/
ao2_wrlock ( channel_cache ) ;
if ( update - > old_snapshot ) {
ao2_unlink_flags ( channel_cache , update - > old_snapshot , OBJ_NOLOCK ) ;
}
ao2_link_flags ( channel_cache , update - > new_snapshot , OBJ_NOLOCK ) ;
ao2_unlock ( channel_cache ) ;
/* The same applies here. */
ao2_wrlock ( channel_cache_by_name ) ;
if ( update - > old_snapshot ) {
ao2_unlink_flags ( channel_cache_by_name , update - > old_snapshot , OBJ_NOLOCK ) ;
}
ao2_link_flags ( channel_cache_by_name , update - > new_snapshot , OBJ_NOLOCK ) ;
ao2_unlock ( channel_cache_by_name ) ;
ast_channel_snapshot_set ( chan , update - > new_snapshot ) ;
ast_assert ( ast_channel_topic ( chan ) ! = NULL ) ;
ast_assert ( ast_channel_topic ( chan ) ! = NULL ) ;
stasis_publish ( ast_channel_topic ( chan ) , message ) ;
stasis_publish ( ast_channel_topic ( chan ) , message ) ;
ao2_ref ( message , - 1 ) ;
ao2_ref ( message , - 1 ) ;
@ -841,13 +924,8 @@ void ast_channel_publish_varset(struct ast_channel *chan, const char *name, cons
ast_channel_publish_snapshot ( chan ) ;
ast_channel_publish_snapshot ( chan ) ;
}
}
if ( chan ) {
/* This function is NULL safe for global variables */
ast_channel_publish_cached_blob ( chan , ast_channel_varset_type ( ) , blob ) ;
ast_channel_publish_blob ( chan , ast_channel_varset_type ( ) , blob ) ;
} else {
/* This function is NULL safe for global variables */
ast_channel_publish_blob ( NULL , ast_channel_varset_type ( ) , blob ) ;
}
ast_json_unref ( blob ) ;
ast_json_unref ( blob ) ;
} }
@ -931,36 +1009,6 @@ static struct ast_manager_event_blob *agent_logoff_to_ami(struct stasis_message
return ev ;
return ev ;
} }
void ast_publish_channel_state ( struct ast_channel * chan )
{
struct ast_channel_snapshot * snapshot ;
struct stasis_message * message ;
if ( ! ast_channel_snapshot_type ( ) ) {
return ;
}
ast_assert ( chan ! = NULL ) ;
if ( ! chan ) {
return ;
}
snapshot = ast_channel_snapshot_create ( chan ) ;
if ( ! snapshot ) {
return ;
}
message = stasis_message_create ( ast_channel_snapshot_type ( ) , snapshot ) ;
ao2_ref ( snapshot , - 1 ) ;
if ( ! message ) {
return ;
}
ast_assert ( ast_channel_topic ( chan ) ! = NULL ) ;
stasis_publish ( ast_channel_topic ( chan ) , message ) ;
ao2_ref ( message , - 1 ) ;
}
struct ast_json * ast_channel_snapshot_to_json ( struct ast_json * ast_channel_snapshot_to_json (
const struct ast_channel_snapshot * snapshot ,
const struct ast_channel_snapshot * snapshot ,
const struct stasis_message_sanitizer * sanitize )
const struct stasis_message_sanitizer * sanitize )
@ -1332,12 +1380,12 @@ STASIS_MESSAGE_TYPE_DEFN(ast_channel_talking_stop,
static void stasis_channels_cleanup ( void ) static void stasis_channels_cleanup ( void )
{ {
stasis_caching_unsubscribe_and_join ( channel_by_name_topic ) ;
ao2_cleanup ( channel_topic_all ) ;
channel_by_name_topic = NULL ;
channel_topic_all = NULL ;
ao2_cleanup ( channel_cache ) ;
channel_cache = NULL ;
ao2_cleanup ( channel_cache_by_name ) ;
ao2_cleanup ( channel_cache_by_name ) ;
channel_cache_by_name = NULL ;
channel_cache_by_name = NULL ;
ao2_cleanup ( channel_cache_all ) ;
channel_cache_all = NULL ;
STASIS_MESSAGE_TYPE_CLEANUP ( ast_channel_snapshot_type ) ;
STASIS_MESSAGE_TYPE_CLEANUP ( ast_channel_snapshot_type ) ;
STASIS_MESSAGE_TYPE_CLEANUP ( ast_channel_dial_type ) ;
STASIS_MESSAGE_TYPE_CLEANUP ( ast_channel_dial_type ) ;
@ -1367,29 +1415,28 @@ int ast_stasis_channels_init(void)
ast_register_cleanup ( stasis_channels_cleanup ) ;
ast_register_cleanup ( stasis_channels_cleanup ) ;
channel_cache_all = stasis_cp_all_create ( " ast_channel_topic_all " ,
channel_topic_all = stasis_topic_create ( " ast_channel_topic_all " ) ;
channel_snapshot_get_id ) ;
if ( ! channel_topic_all ) {
if ( ! channel_cache_all ) {
return - 1 ;
return - 1 ;
}
}
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_agent_login_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_agent_logoff_type ) ;
channel_cache_by_name = stasis_cache_create ( channel_snapshot_get_name ) ;
channel_cache = ao2_container_alloc_hash ( AO2_ALLOC_OPT_LOCK_RWLOCK ,
if ( ! channel_cache_by_name ) {
0 , AST_NUM_CHANNEL_BUCKETS , channel_snapshot_uniqueid_hash_cb ,
NULL , channel_snapshot_uniqueid_cmp_cb ) ;
if ( ! channel_cache ) {
return - 1 ;
return - 1 ;
}
}
/* This should be initialized before the caching topic */
channel_cache_by_name = ao2_container_alloc_hash ( AO2_ALLOC_OPT_LOCK_RWLOCK ,
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_snapshot_type ) ;
0 , AST_NUM_CHANNEL_BUCKETS , channel_snapshot_hash_cb ,
NULL , channel_snapshot_cmp_cb ) ;
channel_by_name_topic = stasis_caching_topic_create (
if ( ! channel_cache_by_name ) {
stasis_cp_all_topic ( channel_cache_all ) ,
channel_cache_by_name ) ;
if ( ! channel_by_name_topic ) {
return - 1 ;
return - 1 ;
}
}
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_agent_login_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_agent_logoff_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_snapshot_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_dial_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_dial_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_varset_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_varset_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_hangup_request_type ) ;
res | = STASIS_MESSAGE_TYPE_INIT ( ast_channel_hangup_request_type ) ;
Joshua Colp
7 years ago