sipwise
/
rtpengine
mirror of https://github.com/sipwise/rtpengine.git
1
0
Fork 0
Code Issues Releases Wiki Activity

MT#55283 streamline Redis snapshotting code

Browse Source 
Change-Id: Ia88e3c0a5f9a9d5eaea7ab3fa291341a04b0b73f
pull/2157/merge
Richard Fuchs 5 days ago
parent 1b2afd52f1
commit dcdafb5069
1 changed files with 388 additions and 381 deletions
Show Stats Download Patch File Download Diff File

769
daemon/redis.c
Unescape View File

@ -2693,430 +2693,438 @@ static void json_update_detected_endpoints(const ng_parser_t *parser, parser_arg
? endpoint_print_buf(&ps->detected_endpoints[i]) : "");
}
/**
* encodes the few (k,v) pairs for one call under one json structure
*/
static void redis_encode_ml_basic(struct call_monologue *ml, const ng_parser_t *parser, parser_arg root) {
char tmp[128];
snprintf(tmp, sizeof(tmp), "tag-%u", ml->unique_id);
parser_arg inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE("created", "%" PRId64, ml->created_us);
JSON_SET_SIMPLE("deleted", "%" PRId64, ml->deleted_us);
JSON_SET_SIMPLE("block_dtmf", "%i", ml->block_dtmf);
JSON_SET_SIMPLE("ml_flags", "%" PRIu64, atomic64_get_na(&ml->ml_flags));
JSON_SET_SIMPLE_CSTR("desired_family", ml->desired_family ? ml->desired_family->rfc_name : "");
if (ml->logical_intf)
JSON_SET_SIMPLE_STR("logical_intf", &ml->logical_intf->name);
if (ml->tag.s)
JSON_SET_SIMPLE_STR("tag", &ml->tag);
if (ml->call_id.s)
JSON_SET_SIMPLE_STR("call_id", &ml->tag);
if (ml->viabranch.s)
JSON_SET_SIMPLE_STR("via-branch", &ml->viabranch);
if (ml->label.s)
JSON_SET_SIMPLE_STR("label", &ml->label);
if (ml->metadata.s)
JSON_SET_SIMPLE_STR("metadata", &ml->metadata);
JSON_SET_SIMPLE_STR("sdp_session_name", &ml->sdp_session_name);
JSON_SET_SIMPLE_STR("sdp_session_timing", &ml->sdp_session_timing);
if (ml->sdp_orig_in.parsed) {
JSON_SET_SIMPLE_STR("sdp_orig_username", &ml->sdp_orig_in.username);
JSON_SET_SIMPLE_STR("sdp_orig_session_id", &ml->sdp_orig_in.session_id);
JSON_SET_SIMPLE("sdp_orig_version_num", "%llu", ml->sdp_orig_in.version_num);
JSON_SET_SIMPLE("sdp_orig_parsed", "%u", ml->sdp_orig_in.parsed);
JSON_SET_SIMPLE_STR("sdp_orig_address_network_type", &ml->sdp_orig_in.address.network_type);
JSON_SET_SIMPLE_STR("sdp_orig_address_address_type", &ml->sdp_orig_in.address.address_type);
JSON_SET_SIMPLE_STR("sdp_orig_address_address", &ml->sdp_orig_in.address.address);
}
if (ml->sdp_orig_out.parsed) {
JSON_SET_SIMPLE_STR("last_sdp_orig_username", &ml->sdp_orig_out.username);
JSON_SET_SIMPLE_STR("last_sdp_orig_session_id", &ml->sdp_orig_out.session_id);
JSON_SET_SIMPLE("last_sdp_orig_version_num", "%llu", ml->sdp_orig_out.version_num);
JSON_SET_SIMPLE("last_sdp_orig_parsed", "%u", ml->sdp_orig_out.parsed);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_network_type", &ml->sdp_orig_out.address.network_type);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_address_type", &ml->sdp_orig_out.address.address_type);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_address", &ml->sdp_orig_out.address.address);
}
if (ml->sdp_session_bandwidth.as >= 0)
JSON_SET_SIMPLE("sdp_session_as", "%ld", ml->sdp_session_bandwidth.as);
if (ml->sdp_session_bandwidth.ct >= 0)
JSON_SET_SIMPLE("sdp_session_ct", "%ld", ml->sdp_session_bandwidth.ct);
if (ml->sdp_session_bandwidth.rr >= 0)
JSON_SET_SIMPLE("sdp_session_rr", "%ld", ml->sdp_session_bandwidth.rr);
if (ml->sdp_session_bandwidth.rs >= 0)
JSON_SET_SIMPLE("sdp_session_rs", "%ld", ml->sdp_session_bandwidth.rs);
if (ml->sdp_session_bandwidth.tias >= 0)
JSON_SET_SIMPLE("sdp_session_tias", "%ld", ml->sdp_session_bandwidth.tias);
if (ml->last_out_sdp && ml->last_out_sdp->len)
JSON_SET_SIMPLE_LEN("last_out_sdp", ml->last_out_sdp->len,
ml->last_out_sdp->str);
snprintf(tmp, sizeof(tmp), "medias-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (unsigned int j = 0; j < ml->medias->len; j++) {
struct call_media *media = ml->medias->pdata[j];
JSON_ADD_LIST_STRING("%u", media ? media->unique_id : -1);
}
// scope = write only these monologues' state, plus the state only a rollback
// reads. NULL writes the whole call, which is what the Redis record wants.
static bool ml_in_scope(struct call_monologue * const *scope, const struct call_monologue *ml) {
return !scope || ml == scope[0] || ml == scope[1];
}
static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free,
struct call_monologue * const *scope)
{
static parser_arg redis_encode_media_basic(struct call_media *media, const ng_parser_t *parser, parser_arg root) {
char tmp[128];
snprintf(tmp, sizeof(tmp), "payload_types-%u", media->unique_id);
parser_arg inner = parser->dict_add_list_dup(root, tmp);
redis_encode_codec_store(parser, inner, &media->codecs);
// SSRC table dump
// XXX needs fixing
LOCK(&media->ssrc_hash_in.lock);
snprintf(tmp, sizeof(tmp), "ssrc_table-%u", media->unique_id);
parser_arg list = parser->dict_add_list_dup(root, tmp);
for (GList *m = media->ssrc_hash_in.nq.head; m; m = m->next) {
struct ssrc_entry_call *se = m->data;
inner = parser->list_add_dict(list);
JSON_SET_SIMPLE("ssrc", "%" PRIu32, se->h.ssrc);
// XXX use function for in/out
JSON_SET_SIMPLE("in_srtp_index", "%u", atomic_get_na(&se->stats->ext_seq));
JSON_SET_SIMPLE("in_srtcp_index", "%u", atomic_get_na(&se->stats->rtcp_seq));
JSON_SET_SIMPLE("in_payload_type", "%i", se->tracker.most[0]);
//JSON_SET_SIMPLE("out_srtp_index", "%u", atomic_get_na(&se->output_ctx.stats->ext_seq));
//JSON_SET_SIMPLE("out_srtcp_index", "%u", atomic_get_na(&se->output_ctx.stats->rtcp_seq));
//JSON_SET_SIMPLE("out_payload_type", "%i", se->output_ctx.tracker.most[0]);
// XXX add rest of info
}
snprintf(tmp, sizeof(tmp), "media-%u", media->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE("tag","%u", media->monologue->unique_id);
JSON_SET_SIMPLE("index","%u", media->index);
JSON_SET_SIMPLE_STR("type", &media->type);
if (media->format_str.s)
JSON_SET_SIMPLE_STR("format_str", &media->format_str);
if (media->media_id.s)
JSON_SET_SIMPLE_STR("media_id", &media->media_id);
if (media->label.s)
JSON_SET_SIMPLE_STR("label", &media->label);
JSON_SET_SIMPLE_CSTR("protocol", media->protocol ? media->protocol->name : "");
JSON_SET_SIMPLE_CSTR("desired_family", media->desired_family ? media->desired_family->rfc_name : "");
if (media->logical_intf)
JSON_SET_SIMPLE_STR("logical_intf", &media->logical_intf->name);
JSON_SET_SIMPLE("ptime","%i", media->ptime);
JSON_SET_SIMPLE("maxptime","%i", media->maxptime);
JSON_SET_SIMPLE("media_flags", "%" PRIu64, atomic64_get_na(&media->media_flags));
if (media->sdp_media_bandwidth.as >= 0)
JSON_SET_SIMPLE("bandwidth_as","%ld", media->sdp_media_bandwidth.as);
if (media->sdp_media_bandwidth.rr >= 0)
JSON_SET_SIMPLE("bandwidth_rr","%ld", media->sdp_media_bandwidth.rr);
if (media->sdp_media_bandwidth.rs >= 0)
JSON_SET_SIMPLE("bandwidth_rs","%ld", media->sdp_media_bandwidth.rs);
if (media->sdp_media_bandwidth.ct >= 0)
JSON_SET_SIMPLE("bandwidth_ct","%ld", media->sdp_media_bandwidth.ct);
if (media->sdp_media_bandwidth.tias >= 0)
JSON_SET_SIMPLE("bandwidth_tias","%ld", media->sdp_media_bandwidth.tias);
redis_encode_sdes_params(parser, inner, "sdes_in", &media->sdes_in);
redis_encode_sdes_params(parser, inner, "sdes_out", &media->sdes_out);
redis_encode_dtls_fingerprint(parser, inner, &media->fingerprint);
return inner;
}
static parser_arg redis_encode_stream_basic(struct packet_stream *ps, const ng_parser_t *parser, parser_arg root) {
char tmp[128];
snprintf(tmp, sizeof(tmp), "stream_sfds-%u", ps->unique_id);
parser_arg inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->sfds.head; k; k = k->next) {
stream_fd *sfd = k->data;
JSON_ADD_LIST_STRING("%u", sfd->unique_id);
}
snprintf(tmp, sizeof(tmp), "stream-%u", ps->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE("media","%u",ps->media->unique_id);
JSON_SET_SIMPLE("sfd","%u",ps->selected_sfd ? ps->selected_sfd->unique_id : -1);
JSON_SET_SIMPLE("rtcp_sibling","%u",ps->rtcp_sibling ? ps->rtcp_sibling->unique_id : -1);
JSON_SET_SIMPLE("ps_flags", "%" PRIu64, atomic64_get_na(&ps->ps_flags));
JSON_SET_SIMPLE("component","%u",ps->component);
JSON_SET_SIMPLE_CSTR("endpoint",endpoint_print_buf(&ps->endpoint));
JSON_SET_SIMPLE_CSTR("advertised_endpoint",endpoint_print_buf(&ps->advertised_endpoint));
JSON_SET_SIMPLE("stats-packets","%" PRIu64, atomic64_get_na(&ps->stats_in->packets));
JSON_SET_SIMPLE("stats-bytes","%" PRIu64, atomic64_get_na(&ps->stats_in->bytes));
JSON_SET_SIMPLE("stats-errors","%" PRIu64, atomic64_get_na(&ps->stats_in->errors));
json_update_crypto_params(parser, inner, "", &ps->crypto.params);
return inner;
}
static str redis_encode_json_ml(ng_parser_ctx_t *ctx, call_t *c, void **to_free,
struct call_monologue *ml)
{
char tmp[128];
const ng_parser_t *parser = ctx->parser;
parser_arg root = parser->dict(ctx);
{
parser_arg inner = {0};
if (!scope) {
inner = parser->dict_add_dict(root, "json");
JSON_SET_SIMPLE("created","%" PRId64, c->created);
JSON_SET_SIMPLE("destroyed","%" PRId64, c->destroyed);
JSON_SET_SIMPLE("last_signal","%" PRId64, c->last_signal_us);
JSON_SET_SIMPLE("tos","%u", (int) c->tos);
JSON_SET_SIMPLE("deleted","%" PRId64, c->deleted_us);
JSON_SET_SIMPLE("num_sfds","%u", t_queue_get_length(&c->stream_fds));
JSON_SET_SIMPLE("num_streams","%u", t_queue_get_length(&c->streams));
JSON_SET_SIMPLE("num_medias","%u", t_queue_get_length(&c->medias));
JSON_SET_SIMPLE("num_tags","%u", t_queue_get_length(&c->monologues));
JSON_SET_SIMPLE("num_maps","%u", t_queue_get_length(&c->endpoint_maps));
JSON_SET_SIMPLE("ml_deleted","%" PRId64, c->ml_deleted_us);
JSON_SET_SIMPLE("redis_hosted_db","%u", c->redis_hosted_db);
JSON_SET_SIMPLE_STR("recording_metadata", &c->metadata);
JSON_SET_SIMPLE("block_dtmf","%i", c->block_dtmf);
JSON_SET_SIMPLE("call_flags", "%" PRIu64, atomic64_get_na(&c->call_flags));
if (c->created_from.len)
JSON_SET_SIMPLE_STR("created_from", &c->created_from);
if (c->recording_meta_prefix.len)
JSON_SET_SIMPLE_STR("recording_meta_prefix", &c->recording_meta_prefix);
if (c->recording_file.len)
JSON_SET_SIMPLE_STR("recording_file", &c->recording_file);
if (c->recording_path.len)
JSON_SET_SIMPLE_STR("recording_path", &c->recording_path);
if (c->recording_pattern.len)
JSON_SET_SIMPLE_STR("recording_pattern", &c->recording_pattern);
if (c->recording_random_tag.len)
JSON_SET_SIMPLE_STR("recording_random_tag", &c->recording_random_tag);
parser_arg inner;
redis_encode_ml_basic(ml, parser, root);
for (unsigned int i = 0; i < ml->medias->len; i++) {
__auto_type media = ml->medias->pdata[i];
if (!media)
continue;
inner = redis_encode_media_basic(media, parser, root);
if (media->tls_id.s)
JSON_SET_SIMPLE_STR("tls_id", &media->tls_id);
if (media->fp_hash_func)
JSON_SET_SIMPLE_CSTR("preferred_hash_func",
media->fp_hash_func->name);
if (media->endpoint_map)
JSON_SET_SIMPLE("endpoint_map", "%u",
media->endpoint_map->unique_id);
unsigned int num_cands = 0;
for (__auto_type m = media->ice_candidates.head; m; m = m->next)
num_cands++;
JSON_SET_SIMPLE("num_ice_candidates", "%u", num_cands);
JSON_SET_SIMPLE("had_ice", "%i", media->ice_agent ? 1 : 0);
if (media->ice_agent) {
JSON_SET_SIMPLE_STR("ice_ufrag_local", &media->ice_agent->ufrag[0]);
JSON_SET_SIMPLE_STR("ice_ufrag_remote", &media->ice_agent->ufrag[1]);
JSON_SET_SIMPLE_STR("ice_pwd_local", &media->ice_agent->pwd[0]);
JSON_SET_SIMPLE_STR("ice_pwd_remote", &media->ice_agent->pwd[1]);
}
for (__auto_type l = scope ? NULL : c->monologues.head; l; l = l->next) {
const struct call_monologue *ml = l->data;
if (!ml->checkpoint)
continue;
snprintf(tmp, sizeof(tmp), "checkpoint-%u", ml->unique_id);
unsigned int ci = 0;
for (__auto_type m = media->ice_candidates.head; m; m = m->next, ci++) {
const struct ice_candidate *cand = m->data;
snprintf(tmp, sizeof(tmp), "ice_candidate-%u-%u", media->unique_id, ci);
inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE("pending", "%i", ml->checkpoint->pending ? 1 : 0);
if (ml->checkpoint->snapshot.len) {
/* nested as a string; heap buffer rather than a VLA, as escape() can
* need up to 3x the input */
char *enc = g_malloc_n(ml->checkpoint->snapshot.len + 1, 3);
str encs = parser->escape(enc, ml->checkpoint->snapshot.s,
ml->checkpoint->snapshot.len);
parser->dict_add_str_dup(inner, "snapshot", &encs);
g_free(enc);
}
JSON_SET_SIMPLE_STR("foundation", &cand->foundation);
JSON_SET_SIMPLE("component", "%lu", (unsigned long) cand->component_id);
JSON_SET_SIMPLE_CSTR("transport",
cand->transport ? cand->transport->name : "");
JSON_SET_SIMPLE("priority", "%lu", (unsigned long) cand->priority);
JSON_SET_SIMPLE("type", "%u", cand->type);
JSON_SET_SIMPLE_STR("ufrag", &cand->ufrag);
JSON_SET_SIMPLE_CSTR("endpoint",
cand->endpoint.address.family
? endpoint_print_buf(&cand->endpoint) : "");
JSON_SET_SIMPLE_CSTR("related",
cand->related.address.family
? endpoint_print_buf(&cand->related) : "");
}
for (__auto_type l = scope ? NULL : c->stream_fds.head; l; l = l->next) {
stream_fd *sfd = l->data;
snprintf(tmp, sizeof(tmp), "offered_payload_types-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
redis_encode_codec_store(parser, inner, &media->offered_codecs);
snprintf(tmp, sizeof(tmp), "sfd-%u", sfd->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
IQUEUE_FOREACH(&media->streams, ps) {
if (!ps->media)
continue;
{
JSON_SET_SIMPLE_CSTR("pref_family", sfd->local_intf->logical->preferred_family->rfc_name);
JSON_SET_SIMPLE("localport","%u", sfd->socket.local.port);
JSON_SET_SIMPLE("fd", "%i", sfd->socket.fd);
JSON_SET_SIMPLE_STR("logical_intf", &sfd->local_intf->logical->name);
JSON_SET_SIMPLE("local_intf_uid","%u", sfd->local_intf->unique_id);
JSON_SET_SIMPLE("stream","%u", sfd->stream->unique_id);
LOCK(&ps->lock);
json_update_crypto_params(parser, inner, "", &sfd->crypto.params);
}
inner = redis_encode_stream_basic(ps, parser, root);
JSON_SET_SIMPLE_CSTR("learned_endpoint",
ps->learned_endpoint.address.family
? endpoint_print_buf(&ps->learned_endpoint) : "");
JSON_SET_SIMPLE_CSTR("last_local_endpoint",
ps->last_local_endpoint.address.family
? endpoint_print_buf(&ps->last_local_endpoint) : "");
JSON_SET_SIMPLE("ep_detect_signal", "%" PRId64, ps->ep_detect_signal);
JSON_SET_SIMPLE("el_flags", "%u", ps->el_flags);
json_update_detected_endpoints(parser, inner, ps);
}
}
} // --- for
return parser->collapse(ctx, root, to_free);
}
IQUEUE_FOREACH(&c->streams, ps) {
if (!ps->media || !ml_in_scope(scope, ps->media->monologue))
continue;
LOCK(&ps->lock);
static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free)
{
char tmp[128];
const ng_parser_t *parser = ctx->parser;
snprintf(tmp, sizeof(tmp), "stream-%u", ps->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
parser_arg root = parser->dict(ctx);
{
JSON_SET_SIMPLE("media","%u",ps->media->unique_id);
JSON_SET_SIMPLE("sfd","%u",ps->selected_sfd ? ps->selected_sfd->unique_id : -1);
JSON_SET_SIMPLE("rtcp_sibling","%u",ps->rtcp_sibling ? ps->rtcp_sibling->unique_id : -1);
JSON_SET_SIMPLE("ps_flags", "%" PRIu64, atomic64_get_na(&ps->ps_flags));
JSON_SET_SIMPLE("component","%u",ps->component);
JSON_SET_SIMPLE_CSTR("endpoint",endpoint_print_buf(&ps->endpoint));
JSON_SET_SIMPLE_CSTR("advertised_endpoint",endpoint_print_buf(&ps->advertised_endpoint));
if (scope) {
JSON_SET_SIMPLE_CSTR("learned_endpoint",
ps->learned_endpoint.address.family
? endpoint_print_buf(&ps->learned_endpoint) : "");
JSON_SET_SIMPLE_CSTR("last_local_endpoint",
ps->last_local_endpoint.address.family
? endpoint_print_buf(&ps->last_local_endpoint) : "");
JSON_SET_SIMPLE("ep_detect_signal", "%" PRId64, ps->ep_detect_signal);
JSON_SET_SIMPLE("el_flags", "%u", ps->el_flags);
json_update_detected_endpoints(parser, inner, ps);
}
JSON_SET_SIMPLE("stats-packets","%" PRIu64, atomic64_get_na(&ps->stats_in->packets));
JSON_SET_SIMPLE("stats-bytes","%" PRIu64, atomic64_get_na(&ps->stats_in->bytes));
JSON_SET_SIMPLE("stats-errors","%" PRIu64, atomic64_get_na(&ps->stats_in->errors));
json_update_crypto_params(parser, inner, "", &ps->crypto.params);
}
parser_arg inner;
inner = parser->dict_add_dict(root, "json");
JSON_SET_SIMPLE("created","%" PRId64, c->created);
JSON_SET_SIMPLE("destroyed","%" PRId64, c->destroyed);
JSON_SET_SIMPLE("last_signal","%" PRId64, c->last_signal_us);
JSON_SET_SIMPLE("tos","%u", (int) c->tos);
JSON_SET_SIMPLE("deleted","%" PRId64, c->deleted_us);
JSON_SET_SIMPLE("num_sfds","%u", t_queue_get_length(&c->stream_fds));
JSON_SET_SIMPLE("num_streams","%u", t_queue_get_length(&c->streams));
JSON_SET_SIMPLE("num_medias","%u", t_queue_get_length(&c->medias));
JSON_SET_SIMPLE("num_tags","%u", t_queue_get_length(&c->monologues));
JSON_SET_SIMPLE("num_maps","%u", t_queue_get_length(&c->endpoint_maps));
JSON_SET_SIMPLE("ml_deleted","%" PRId64, c->ml_deleted_us);
JSON_SET_SIMPLE("redis_hosted_db","%u", c->redis_hosted_db);
JSON_SET_SIMPLE_STR("recording_metadata", &c->metadata);
JSON_SET_SIMPLE("block_dtmf","%i", c->block_dtmf);
JSON_SET_SIMPLE("call_flags", "%" PRIu64, atomic64_get_na(&c->call_flags));
if (c->created_from.len)
JSON_SET_SIMPLE_STR("created_from", &c->created_from);
if (c->recording_meta_prefix.len)
JSON_SET_SIMPLE_STR("recording_meta_prefix", &c->recording_meta_prefix);
if (c->recording_file.len)
JSON_SET_SIMPLE_STR("recording_file", &c->recording_file);
if (c->recording_path.len)
JSON_SET_SIMPLE_STR("recording_path", &c->recording_path);
if (c->recording_pattern.len)
JSON_SET_SIMPLE_STR("recording_pattern", &c->recording_pattern);
if (c->recording_random_tag.len)
JSON_SET_SIMPLE_STR("recording_random_tag", &c->recording_random_tag);
snprintf(tmp, sizeof(tmp), "stream_sfds-%u", ps->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->sfds.head; k; k = k->next) {
stream_fd *sfd = k->data;
JSON_ADD_LIST_STRING("%u", sfd->unique_id);
}
for (__auto_type l = c->monologues.head; l; l = l->next) {
const struct call_monologue *ml = l->data;
if (!ml->checkpoint)
continue;
snprintf(tmp, sizeof(tmp), "checkpoint-%u", ml->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE("pending", "%i", ml->checkpoint->pending ? 1 : 0);
if (ml->checkpoint->snapshot.len) {
/* nested as a string; heap buffer rather than a VLA, as escape() can
* need up to 3x the input */
char *enc = g_malloc_n(ml->checkpoint->snapshot.len + 1, 3);
str encs = parser->escape(enc, ml->checkpoint->snapshot.s,
ml->checkpoint->snapshot.len);
parser->dict_add_str_dup(inner, "snapshot", &encs);
g_free(enc);
}
}
if (!scope) {
snprintf(tmp, sizeof(tmp), "rtp_sinks-%u", ps->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->rtp_sinks.head; k; k = k->next) {
struct sink_handler *sh = k->data;
struct packet_stream *sink = sh->sink;
JSON_ADD_LIST_STRING("%u", sink->unique_id);
}
}
for (__auto_type l = c->stream_fds.head; l; l = l->next) {
stream_fd *sfd = l->data;
if (!scope) {
snprintf(tmp, sizeof(tmp), "rtcp_sinks-%u", ps->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->rtcp_sinks.head; k; k = k->next) {
struct sink_handler *sh = k->data;
struct packet_stream *sink = sh->sink;
JSON_ADD_LIST_STRING("%u", sink->unique_id);
}
}
} // --- for streams.head
snprintf(tmp, sizeof(tmp), "sfd-%u", sfd->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
for (__auto_type l = c->monologues.head; l; l = l->next) {
struct call_monologue *ml = l->data;
JSON_SET_SIMPLE_CSTR("pref_family", sfd->local_intf->logical->preferred_family->rfc_name);
JSON_SET_SIMPLE("localport","%u", sfd->socket.local.port);
JSON_SET_SIMPLE("fd", "%i", sfd->socket.fd);
JSON_SET_SIMPLE_STR("logical_intf", &sfd->local_intf->logical->name);
JSON_SET_SIMPLE("local_intf_uid","%u", sfd->local_intf->unique_id);
JSON_SET_SIMPLE("stream","%u", sfd->stream->unique_id);
if (!ml_in_scope(scope, ml))
continue;
json_update_crypto_params(parser, inner, "", &sfd->crypto.params);
snprintf(tmp, sizeof(tmp), "tag-%u", ml->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
} // --- for
{
JSON_SET_SIMPLE("created", "%" PRId64, ml->created_us);
JSON_SET_SIMPLE("deleted", "%" PRId64, ml->deleted_us);
JSON_SET_SIMPLE("block_dtmf", "%i", ml->block_dtmf);
JSON_SET_SIMPLE("ml_flags", "%" PRIu64, atomic64_get_na(&ml->ml_flags));
JSON_SET_SIMPLE_CSTR("desired_family", ml->desired_family ? ml->desired_family->rfc_name : "");
if (ml->logical_intf)
JSON_SET_SIMPLE_STR("logical_intf", &ml->logical_intf->name);
if (ml->tag.s)
JSON_SET_SIMPLE_STR("tag", &ml->tag);
if (ml->call_id.s)
JSON_SET_SIMPLE_STR("call_id", &ml->tag);
if (ml->viabranch.s)
JSON_SET_SIMPLE_STR("via-branch", &ml->viabranch);
if (ml->label.s)
JSON_SET_SIMPLE_STR("label", &ml->label);
if (ml->metadata.s)
JSON_SET_SIMPLE_STR("metadata", &ml->metadata);
JSON_SET_SIMPLE_STR("sdp_session_name", &ml->sdp_session_name);
JSON_SET_SIMPLE_STR("sdp_session_timing", &ml->sdp_session_timing);
if (ml->sdp_orig_in.parsed) {
JSON_SET_SIMPLE_STR("sdp_orig_username", &ml->sdp_orig_in.username);
JSON_SET_SIMPLE_STR("sdp_orig_session_id", &ml->sdp_orig_in.session_id);
JSON_SET_SIMPLE("sdp_orig_version_num", "%llu", ml->sdp_orig_in.version_num);
JSON_SET_SIMPLE("sdp_orig_parsed", "%u", ml->sdp_orig_in.parsed);
JSON_SET_SIMPLE_STR("sdp_orig_address_network_type", &ml->sdp_orig_in.address.network_type);
JSON_SET_SIMPLE_STR("sdp_orig_address_address_type", &ml->sdp_orig_in.address.address_type);
JSON_SET_SIMPLE_STR("sdp_orig_address_address", &ml->sdp_orig_in.address.address);
}
if (ml->sdp_orig_out.parsed) {
JSON_SET_SIMPLE_STR("last_sdp_orig_username", &ml->sdp_orig_out.username);
JSON_SET_SIMPLE_STR("last_sdp_orig_session_id", &ml->sdp_orig_out.session_id);
JSON_SET_SIMPLE("last_sdp_orig_version_num", "%llu", ml->sdp_orig_out.version_num);
JSON_SET_SIMPLE("last_sdp_orig_parsed", "%u", ml->sdp_orig_out.parsed);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_network_type", &ml->sdp_orig_out.address.network_type);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_address_type", &ml->sdp_orig_out.address.address_type);
JSON_SET_SIMPLE_STR("last_sdp_orig_address_address", &ml->sdp_orig_out.address.address);
}
if (ml->sdp_session_bandwidth.as >= 0)
JSON_SET_SIMPLE("sdp_session_as", "%ld", ml->sdp_session_bandwidth.as);
if (ml->sdp_session_bandwidth.ct >= 0)
JSON_SET_SIMPLE("sdp_session_ct", "%ld", ml->sdp_session_bandwidth.ct);
if (ml->sdp_session_bandwidth.rr >= 0)
JSON_SET_SIMPLE("sdp_session_rr", "%ld", ml->sdp_session_bandwidth.rr);
if (ml->sdp_session_bandwidth.rs >= 0)
JSON_SET_SIMPLE("sdp_session_rs", "%ld", ml->sdp_session_bandwidth.rs);
if (ml->sdp_session_bandwidth.tias >= 0)
JSON_SET_SIMPLE("sdp_session_tias", "%ld", ml->sdp_session_bandwidth.tias);
if (ml->last_out_sdp && ml->last_out_sdp->len)
JSON_SET_SIMPLE_LEN("last_out_sdp", ml->last_out_sdp->len,
ml->last_out_sdp->str);
}
IQUEUE_FOREACH(&c->streams, ps) {
if (!ps->media)
continue;
GList *k = g_hash_table_get_values(ml->associated_tags);
if (!scope) {
snprintf(tmp, sizeof(tmp), "associated_tags-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (GList *m = k; m; m = m->next) {
struct call_monologue *ml2 = m->data;
JSON_ADD_LIST_STRING("%u", ml2->unique_id);
}
}
LOCK(&ps->lock);
g_list_free(k);
redis_encode_stream_basic(ps, parser, root);
if (!scope) {
snprintf(tmp, sizeof(tmp), "tag_aliases-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type alias = ml->tag_aliases.head; alias; alias = alias->next)
JSON_ADD_LIST_STRING(STR_FORMAT, STR_FMT(alias->data));
}
snprintf(tmp, sizeof(tmp), "rtp_sinks-%u", ps->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->rtp_sinks.head; k; k = k->next) {
struct sink_handler *sh = k->data;
struct packet_stream *sink = sh->sink;
JSON_ADD_LIST_STRING("%u", sink->unique_id);
}
snprintf(tmp, sizeof(tmp), "medias-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (unsigned int j = 0; j < ml->medias->len; j++) {
struct call_media *media = ml->medias->pdata[j];
JSON_ADD_LIST_STRING("%u", media ? media->unique_id : -1);
}
} // --- for monologues.head
snprintf(tmp, sizeof(tmp), "rtcp_sinks-%u", ps->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type k = ps->rtcp_sinks.head; k; k = k->next) {
struct sink_handler *sh = k->data;
struct packet_stream *sink = sh->sink;
JSON_ADD_LIST_STRING("%u", sink->unique_id);
}
} // --- for streams.head
for (__auto_type l = c->medias.head; l; l = l->next) {
struct call_media *media = l->data;
for (__auto_type l = c->monologues.head; l; l = l->next) {
struct call_monologue *ml = l->data;
if (!media || !ml_in_scope(scope, media->monologue))
continue;
redis_encode_ml_basic(ml, parser, root);
if (!scope) {
/* store media subscriptions */
snprintf(tmp, sizeof(tmp), "media-subscriptions-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->media_subscriptions, ms) {
JSON_ADD_LIST_STRING("%u/%u/%u/%u/%u",
ms->media->unique_id,
ms->attrs.offer_answer,
ms->attrs.rtcp_only,
ms->attrs.egress,
ms->attrs.inject);
}
}
GList *k = g_hash_table_get_values(ml->associated_tags);
snprintf(tmp, sizeof(tmp), "associated_tags-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (GList *m = k; m; m = m->next) {
struct call_monologue *ml2 = m->data;
JSON_ADD_LIST_STRING("%u", ml2->unique_id);
}
g_list_free(k);
snprintf(tmp, sizeof(tmp), "media-%u", media->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
snprintf(tmp, sizeof(tmp), "tag_aliases-%u", ml->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type alias = ml->tag_aliases.head; alias; alias = alias->next)
JSON_ADD_LIST_STRING(STR_FORMAT, STR_FMT(alias->data));
} // --- for monologues.head
{
JSON_SET_SIMPLE("tag","%u", media->monologue->unique_id);
JSON_SET_SIMPLE("index","%u", media->index);
JSON_SET_SIMPLE_STR("type", &media->type);
if (media->format_str.s)
JSON_SET_SIMPLE_STR("format_str", &media->format_str);
if (media->media_id.s)
JSON_SET_SIMPLE_STR("media_id", &media->media_id);
if (media->label.s)
JSON_SET_SIMPLE_STR("label", &media->label);
JSON_SET_SIMPLE_CSTR("protocol", media->protocol ? media->protocol->name : "");
JSON_SET_SIMPLE_CSTR("desired_family", media->desired_family ? media->desired_family->rfc_name : "");
if (media->logical_intf)
JSON_SET_SIMPLE_STR("logical_intf", &media->logical_intf->name);
JSON_SET_SIMPLE("ptime","%i", media->ptime);
JSON_SET_SIMPLE("maxptime","%i", media->maxptime);
JSON_SET_SIMPLE("media_flags", "%" PRIu64, atomic64_get_na(&media->media_flags));
if (media->sdp_media_bandwidth.as >= 0)
JSON_SET_SIMPLE("bandwidth_as","%ld", media->sdp_media_bandwidth.as);
if (media->sdp_media_bandwidth.rr >= 0)
JSON_SET_SIMPLE("bandwidth_rr","%ld", media->sdp_media_bandwidth.rr);
if (media->sdp_media_bandwidth.rs >= 0)
JSON_SET_SIMPLE("bandwidth_rs","%ld", media->sdp_media_bandwidth.rs);
if (media->sdp_media_bandwidth.ct >= 0)
JSON_SET_SIMPLE("bandwidth_ct","%ld", media->sdp_media_bandwidth.ct);
if (media->sdp_media_bandwidth.tias >= 0)
JSON_SET_SIMPLE("bandwidth_tias","%ld", media->sdp_media_bandwidth.tias);
if (scope) {
if (media->tls_id.s)
JSON_SET_SIMPLE_STR("tls_id", &media->tls_id);
if (media->fp_hash_func)
JSON_SET_SIMPLE_CSTR("preferred_hash_func",
media->fp_hash_func->name);
if (media->endpoint_map)
JSON_SET_SIMPLE("endpoint_map", "%u",
media->endpoint_map->unique_id);
unsigned int num_cands = 0;
for (__auto_type m = media->ice_candidates.head; m; m = m->next)
num_cands++;
JSON_SET_SIMPLE("num_ice_candidates", "%u", num_cands);
JSON_SET_SIMPLE("had_ice", "%i", media->ice_agent ? 1 : 0);
if (media->ice_agent) {
JSON_SET_SIMPLE_STR("ice_ufrag_local", &media->ice_agent->ufrag[0]);
JSON_SET_SIMPLE_STR("ice_ufrag_remote", &media->ice_agent->ufrag[1]);
JSON_SET_SIMPLE_STR("ice_pwd_local", &media->ice_agent->pwd[0]);
JSON_SET_SIMPLE_STR("ice_pwd_remote", &media->ice_agent->pwd[1]);
}
}
redis_encode_sdes_params(parser, inner, "sdes_in", &media->sdes_in);
redis_encode_sdes_params(parser, inner, "sdes_out", &media->sdes_out);
redis_encode_dtls_fingerprint(parser, inner, &media->fingerprint);
}
for (__auto_type l = c->medias.head; l; l = l->next) {
struct call_media *media = l->data;
if (scope) {
unsigned int ci = 0;
for (__auto_type m = media->ice_candidates.head; m; m = m->next, ci++) {
const struct ice_candidate *cand = m->data;
snprintf(tmp, sizeof(tmp), "ice_candidate-%u-%u", media->unique_id, ci);
inner = parser->dict_add_dict_dup(root, tmp);
JSON_SET_SIMPLE_STR("foundation", &cand->foundation);
JSON_SET_SIMPLE("component", "%lu", (unsigned long) cand->component_id);
JSON_SET_SIMPLE_CSTR("transport",
cand->transport ? cand->transport->name : "");
JSON_SET_SIMPLE("priority", "%lu", (unsigned long) cand->priority);
JSON_SET_SIMPLE("type", "%u", cand->type);
JSON_SET_SIMPLE_STR("ufrag", &cand->ufrag);
JSON_SET_SIMPLE_CSTR("endpoint",
cand->endpoint.address.family
? endpoint_print_buf(&cand->endpoint) : "");
JSON_SET_SIMPLE_CSTR("related",
cand->related.address.family
? endpoint_print_buf(&cand->related) : "");
}
}
if (!media)
continue;
if (!scope) {
snprintf(tmp, sizeof(tmp), "streams-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->streams, ps)
JSON_ADD_LIST_STRING("%u", ps->unique_id);
}
/* store media subscriptions */
snprintf(tmp, sizeof(tmp), "media-subscriptions-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->media_subscriptions, ms) {
JSON_ADD_LIST_STRING("%u/%u/%u/%u/%u",
ms->media->unique_id,
ms->attrs.offer_answer,
ms->attrs.rtcp_only,
ms->attrs.egress,
ms->attrs.inject);
}
if (!scope) {
snprintf(tmp, sizeof(tmp), "maps-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->endpoint_maps, ep)
JSON_ADD_LIST_STRING("%u", ep->unique_id);
}
redis_encode_media_basic(media, parser, root);
snprintf(tmp, sizeof(tmp), "payload_types-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
redis_encode_codec_store(parser, inner, &media->codecs);
snprintf(tmp, sizeof(tmp), "streams-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->streams, ps)
JSON_ADD_LIST_STRING("%u", ps->unique_id);
if (scope) {
snprintf(tmp, sizeof(tmp), "offered_payload_types-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
redis_encode_codec_store(parser, inner, &media->offered_codecs);
}
snprintf(tmp, sizeof(tmp), "maps-%u", media->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
IQUEUE_FOREACH(&media->endpoint_maps, ep)
JSON_ADD_LIST_STRING("%u", ep->unique_id);
// SSRC table dump
// XXX needs fixing
LOCK(&media->ssrc_hash_in.lock);
snprintf(tmp, sizeof(tmp), "ssrc_table-%u", media->unique_id);
parser_arg list = parser->dict_add_list_dup(root, tmp);
for (GList *m = media->ssrc_hash_in.nq.head; m; m = m->next) {
struct ssrc_entry_call *se = m->data;
inner = parser->list_add_dict(list);
JSON_SET_SIMPLE("ssrc", "%" PRIu32, se->h.ssrc);
// XXX use function for in/out
JSON_SET_SIMPLE("in_srtp_index", "%u", atomic_get_na(&se->stats->ext_seq));
JSON_SET_SIMPLE("in_srtcp_index", "%u", atomic_get_na(&se->stats->rtcp_seq));
JSON_SET_SIMPLE("in_payload_type", "%i", se->tracker.most[0]);
//JSON_SET_SIMPLE("out_srtp_index", "%u", atomic_get_na(&se->output_ctx.stats->ext_seq));
//JSON_SET_SIMPLE("out_srtcp_index", "%u", atomic_get_na(&se->output_ctx.stats->rtcp_seq));
//JSON_SET_SIMPLE("out_payload_type", "%i", se->output_ctx.tracker.most[0]);
// XXX add rest of info
}
} // --- for medias.head
if (!scope) {
IQUEUE_FOREACH(&c->endpoint_maps, ep) {
snprintf(tmp, sizeof(tmp), "map-%u", ep->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
{
JSON_SET_SIMPLE("wildcard","%i", ep->wildcard);
JSON_SET_SIMPLE("num_ports","%u", ep->num_ports);
JSON_SET_SIMPLE_CSTR("intf_preferred_family", ep->logical_intf->preferred_family->rfc_name);
JSON_SET_SIMPLE_STR("logical_intf", &ep->logical_intf->name);
JSON_SET_SIMPLE_CSTR("endpoint", endpoint_print_buf(&ep->endpoint));
}
snprintf(tmp, sizeof(tmp), "map_sfds-%u", ep->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type m = ep->intf_sfds.head; m; m = m->next) {
struct sfd_intf_list *il = m->data;
JSON_ADD_LIST_STRING("loc-%u", il->local_intf->unique_id);
for (__auto_type n = il->list.head; n; n = n->next) {
stream_fd *sfd = n->data;
JSON_ADD_LIST_STRING("%u", sfd->unique_id);
}
}
}
} // --- for c->endpoint_maps.head
} // --- for medias.head
IQUEUE_FOREACH(&c->endpoint_maps, ep) {
snprintf(tmp, sizeof(tmp), "map-%u", ep->unique_id);
inner = parser->dict_add_dict_dup(root, tmp);
{
JSON_SET_SIMPLE("wildcard","%i", ep->wildcard);
JSON_SET_SIMPLE("num_ports","%u", ep->num_ports);
JSON_SET_SIMPLE_CSTR("intf_preferred_family", ep->logical_intf->preferred_family->rfc_name);
JSON_SET_SIMPLE_STR("logical_intf", &ep->logical_intf->name);
JSON_SET_SIMPLE_CSTR("endpoint", endpoint_print_buf(&ep->endpoint));
}
snprintf(tmp, sizeof(tmp), "map_sfds-%u", ep->unique_id);
inner = parser->dict_add_list_dup(root, tmp);
for (__auto_type m = ep->intf_sfds.head; m; m = m->next) {
struct sfd_intf_list *il = m->data;
JSON_ADD_LIST_STRING("loc-%u", il->local_intf->unique_id);
for (__auto_type n = il->list.head; n; n = n->next) {
stream_fd *sfd = n->data;
JSON_ADD_LIST_STRING("%u", sfd->unique_id);
}
}
}
return parser->collapse(ctx, root, to_free);
@ -3124,14 +3132,13 @@ static str redis_encode_json(ng_parser_ctx_t *ctx, call_t *c, void **to_free,
str redis_snapshot_encode(call_t *c, struct call_monologue *ml) {
struct call_monologue *scope[2] = { ml, ml };
ng_parser_ctx_t ctx;
arena_t bbuf;
// never leaves the daemon, so the format is ours to pick
ng_parser_native.init(&ctx, &bbuf);
void *to_free = NULL;
str encoded = redis_encode_json(&ctx, c, &to_free, scope);
str encoded = redis_encode_json_ml(&ctx, c, &to_free, ml);
str out = STR_NULL;
if (encoded.len)
out = str_dup_str(&encoded);
@ -3560,7 +3567,7 @@ void redis_update_onekey(call_t *c, struct redis *r) {
redis_format_parsers[rtpe_config.redis_format]->init(&ctx, &bbuf);
void *to_free = NULL;
str result = redis_encode_json(&ctx, c, &to_free, NULL);
str result = redis_encode_json(&ctx, c, &to_free);
if (!result.len)
goto err;

Write Preview
Loading…
Cancel
Save