Skip to content

feature: sip publish events support #3605

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
digiboridev wants to merge 9 commits into
base: master
Choose a base branch
from
Open

feature: sip publish events support #3605

Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
f28e428
publish events support
digiboridev Oct 28, 2025
8749587
fix: remove ttl double validation
digiboridev Nov 13, 2025
ce6414a
fix: publish call_id generating memory leak
digiboridev Nov 13, 2025
76b69af
fix: intent
digiboridev Nov 13, 2025
0527e96
fix: add sip etag responce check
digiboridev Nov 13, 2025
a427c3c
fix: publish response var names, add missed break
digiboridev Nov 13, 2025
7cfaa15
feat: publish doc
digiboridev Nov 13, 2025
ed835cd
refactor: change publishers storing logic to one per event_type
digiboridev Nov 13, 2025
c9eaa71
fix: publish call id create bug
digiboridev Nov 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
270 changes: 269 additions & 1 deletion src/plugins/janus_sip.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -895,6 +895,16 @@ static struct janus_json_parameter sipmessage_parameters[] = {
{"headers", JSON_OBJECT, 0},
{"call_id", JANUS_JSON_STRING, 0}
};
static struct janus_json_parameter publish_parameters[] = {
{"to", JSON_STRING, 0},
{"event", JSON_STRING, JANUS_JSON_PARAM_REQUIRED},
{"content_type", JSON_STRING, 0},
{"content", JSON_STRING, 0},
{"publish_ttl", JANUS_JSON_INTEGER, 0},
{"etag", JSON_STRING, 0},
{"headers", JSON_OBJECT, 0},
{"call_id", JANUS_JSON_STRING, 0}
};
static struct janus_json_parameter keyframe_parameters[] = {
{"user", JANUS_JSON_BOOL, 0},
{"peer", JANUS_JSON_BOOL, 0}
Expand All @@ -915,6 +925,8 @@ static char *user_agent;
static int register_ttl = JANUS_DEFAULT_REGISTER_TTL;
#define JANUS_DEFAULT_SUBSCRIBE_TTL 3600
static int subscribe_ttl = JANUS_DEFAULT_SUBSCRIBE_TTL;
#define JANUS_DEFAULT_PUBLISH_TTL 3600
static int publish_ttl = JANUS_DEFAULT_PUBLISH_TTL;
static uint16_t rtp_range_min = 10000;
static uint16_t rtp_range_max = 60000;
static int dscp_audio_rtp = 0;
Expand Down Expand Up @@ -1022,6 +1034,7 @@ struct ssip_s {
nua_handle_t *s_nh_r, *s_nh_i, *s_nh_m;
char *contact_header; /* Only needed for Sofia SIP >= 1.13 */
GHashTable *subscriptions;
GHashTable *publishers;
janus_mutex smutex;
struct janus_sip_session *session;
};
Expand Down Expand Up @@ -3616,6 +3629,168 @@ static void *janus_sip_handler(void *data) {
SIPTAG_EXPIRES_STR("0"), TAG_END());
result = json_object();
json_object_set_new(result, "event", json_string("unsubscribing"));
} else if(!strcasecmp(request_text, "publish")) {
/* Send a SIP PUBLISH request for an event package */
JANUS_VALIDATE_JSON_OBJECT(root, publish_parameters,
error_code, error_cause, TRUE,
JANUS_SIP_ERROR_MISSING_ELEMENT, JANUS_SIP_ERROR_INVALID_ELEMENT);
if(error_code != 0)
goto error;
if(session->account.registration_status != janus_sip_registration_status_registered &&
session->account.registration_status != janus_sip_registration_status_disabled) {
JANUS_LOG(LOG_ERR, "Wrong state (not registered)\n");
error_code = JANUS_SIP_ERROR_WRONG_STATE;
g_snprintf(error_cause, 512, "Wrong state (not registered)");
goto error;
}
const char *to = json_string_value(json_object_get(root, "to"));
if(to == NULL)
to = session->account.identity;
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the To you put there yourself or someone else? Should it be required instead of optional?

Copy link
Contributor Author

@digiboridev digiboridev Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes i think is ok to be optional, for presence event for example you can omit "to" for send it to your identity.

const char *event_type = json_string_value(json_object_get(root, "event"));
/* TTL */
int ttl = publish_ttl;
json_t *pub_ttl = json_object_get(root, "publish_ttl");
if(pub_ttl && json_is_integer(pub_ttl))
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's no need to check the type when we have validation. You already have the

{"publish_ttl", JANUS_JSON_INTEGER, 0},

to take care of that. I see you do the same thing for other properties you defined, so please fix it for those too.

ttl = json_integer_value(pub_ttl);
if(ttl <= 0)
ttl = JANUS_DEFAULT_PUBLISH_TTL;
char ttl_text[20];
g_snprintf(ttl_text, sizeof(ttl_text), "%d", ttl);

/* Optional ETag for refresh/modify (If-Match) */
const char *etag = json_string_value(json_object_get(root, "etag"));

/* Take call-id from request, if it exists */
const char *callid = NULL;
json_t *request_callid = json_object_get(root, "call_id");
if(request_callid)
callid = json_string_value(request_callid);

/* Create a hash key for the publishers table */
const char *hashkey = event_type;
if(callid)
hashkey = callid;

/* If call-id does not exist in request, create a random one */
if(callid == NULL) {
JANUS_LOG(LOG_WARN, "Invalid call_id provided, generating a random one\n");
callid = g_malloc0(24);
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a memory leak. You're not freeing this memory when you're done. Adding a g_free would be broken anyway, since if you get it from json_string_value then it should not be freed that way. Probably better to use a generic string, e.g., char new_callid[24], call janus_sip_random_string on that, and have callid be a pointer to that.

janus_sip_random_string(24, callid);
}

/* Prepare or reuse per-event handle (publishers) */
nua_handle_t *nh = NULL;
janus_mutex_lock(&session->stack->smutex);
if(session->stack->publishers != NULL)
nh = g_hash_table_lookup(session->stack->publishers, (char *)hashkey);
if(nh == NULL) {
/* Create a handle in the appropriate NUA */
nua_t *use_nua = NULL;
ssip_t *use_stack = session->stack;
if(session->helper && session->master && session->master->stack)
use_stack = session->master->stack;
if(use_stack->s_nua == NULL) {
janus_mutex_unlock(&session->stack->smutex);
JANUS_LOG(LOG_ERR, "NUA destroyed while publishing?\n");
error_code = JANUS_SIP_ERROR_LIBSOFIA_ERROR;
g_snprintf(error_cause, 512, "Invalid NUA");
goto error;
}
use_nua = use_stack->s_nua;
nh = nua_handle(use_nua, session, TAG_END());
if(session->stack->publishers == NULL) {
/* Create table for mapping publishers too */
session->stack->publishers = g_hash_table_new_full(g_str_hash, g_str_equal,
(GDestroyNotify)g_free, (GDestroyNotify)nua_handle_destroy);
}
g_hash_table_insert(session->stack->publishers, g_strdup(hashkey), nh);
}
janus_mutex_unlock(&session->stack->smutex);

char custom_headers[2048];
janus_sip_parse_custom_headers(root, (char *)&custom_headers, sizeof(custom_headers));
char *contact_header = janus_sip_session_contact_header_retrieve(session);
char *proxy = session->helper && session->master ?
session->master->account.outbound_proxy : session->account.outbound_proxy;
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code style: broken indentation (unneeded extra tab).

const char *content_type = NULL;
json_t *content_type_text = json_object_get(root, "content_type");
if(content_type_text && json_is_string(content_type_text))
content_type = json_string_value(content_type_text);
const char *msg_content = NULL;
json_t *msg_content_text = json_object_get(root, "content");
if(msg_content_text && json_is_string(msg_content_text))
msg_content = json_string_value(msg_content_text);

/* Send PUBLISH */
nua_publish(nh,
SIPTAG_TO_STR(to),
SIPTAG_EVENT_STR(event_type),
SIPTAG_CALL_ID_STR(callid),
TAG_IF(contact_header != NULL, SIPTAG_CONTACT_STR(contact_header)),
SIPTAG_EXPIRES_STR(ttl_text),
TAG_IF(proxy != NULL, NUTAG_PROXY(proxy)),
TAG_IF(strlen(custom_headers) > 0, SIPTAG_HEADER_STR(custom_headers)),
TAG_IF(content_type != NULL && msg_content != NULL, SIPTAG_CONTENT_TYPE_STR(content_type)),
TAG_IF(content_type != NULL && msg_content != NULL, SIPTAG_PAYLOAD_STR(msg_content)),
TAG_IF(etag != NULL, SIPTAG_IF_MATCH_STR(etag)),
TAG_END());
result = json_object();
json_object_set_new(result, "event", json_string("publishing"));
if(callid)
json_object_set_new(result, "call_id", json_string(callid));
} else if(!strcasecmp(request_text, "unpublish")) {
/* Unpublish from some SIP events */
JANUS_VALIDATE_JSON_OBJECT(root, publish_parameters,
error_code, error_cause, TRUE,
JANUS_SIP_ERROR_MISSING_ELEMENT, JANUS_SIP_ERROR_INVALID_ELEMENT);
if(error_code != 0)
goto error;
if(session->account.registration_status != janus_sip_registration_status_registered &&
session->account.registration_status != janus_sip_registration_status_disabled) {
JANUS_LOG(LOG_ERR, "Wrong state (not registered)\n");
error_code = JANUS_SIP_ERROR_WRONG_STATE;
g_snprintf(error_cause, 512, "Wrong state (not registered)");
goto error;
}
const char *to = json_string_value(json_object_get(root, "to"));
if(to == NULL)
to = session->account.identity;
const char *event_type = json_string_value(json_object_get(root, "event"));

/* Take call-id from request, if it exists */
const char *callid = NULL;
json_t *request_callid = json_object_get(root, "call_id");
if(request_callid)
callid = json_string_value(request_callid);

/* Create a hash key for the publishers table */
const char *hashkey = event_type;
if(callid)
hashkey = callid;

/* Get the handle we used for this publishing */
janus_mutex_lock(&session->stack->smutex);
nua_handle_t *nh = NULL;
if(session->stack->publishers != NULL)
nh = g_hash_table_lookup(session->stack->publishers, (char *)hashkey);
janus_mutex_unlock(&session->stack->smutex);
if(nh == NULL) {
JANUS_LOG(LOG_ERR, "Wrong state (no publishers to this call id or event type)\n");
error_code = JANUS_SIP_ERROR_WRONG_STATE;
g_snprintf(error_cause, 512, "Wrong state (no publishers to this call id or event type)");
goto error;
}

/* Send the PUBLISH with Expires set to 0 */
nua_publish(nh,
SIPTAG_TO_STR(to),
SIPTAG_EVENT_STR(event_type),
SIPTAG_EXPIRES_STR("0"), TAG_END()
);
result = json_object();
json_object_set_new(result, "event", json_string("unpublishing"));
if(callid)
json_object_set_new(result, "call_id", json_string(callid));
} else if(!strcasecmp(request_text, "call")) {
/* Call another peer */
if(session->stack == NULL) {
Expand Down Expand Up @@ -5992,6 +6167,95 @@ void janus_sip_sofia_callback(nua_event_t event, int status, char const *phrase,
g_free(messageid);
}
break;
case nua_r_publish: {
JANUS_LOG(LOG_VERB, "[%s][%s]: %d %s\n", session->account.username, nua_event_name(event), status, phrase ? phrase : "??");
if(status == 200 || status == 202) {
/* Success */
json_t *eventj = json_object();
json_object_set_new(eventj, "sip", json_string("event"));
if(sip && sip->sip_call_id)
json_object_set_new(eventj, "call_id", json_string(sip->sip_call_id->i_id));
json_t *resultj = json_object();
json_object_set_new(resultj, "event", json_string("publish_succeeded"));
json_object_set_new(resultj, "code", json_integer(status));
if(session->incoming_header_prefixes) {
json_t *headers = janus_sip_get_incoming_headers(sip, session);
json_object_set_new(resultj, "headers", headers);
}
json_object_set_new(resultj, "etag", json_string(sip->sip_etag->g_string));
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is missing a check on whether sip->sip_etag and sip->sip_etag->g_string exist, as you do for sip_call_id above. It will crash if it doesn't.

if (sip && sip->sip_expires)
json_object_set_new(resultj, "expires", json_integer(sip->sip_expires->ex_delta));
json_object_set_new(resultj, "reason", json_string(phrase ? phrase : ""));
json_object_set_new(eventj, "result", resultj);
int ret = gateway->push_event(session->handle, &janus_sip_plugin, session->transaction, eventj, NULL);
JANUS_LOG(LOG_VERB, " >> Pushing event to peer: %d (%s)\n", ret, janus_get_api_error(ret));
json_decref(eventj);
} else if(status == 401 || status == 407) {
const char *scheme = NULL;
const char *realm = NULL;
if(status == 401) {
sip_www_authenticate_t const* www_auth = sip->sip_www_authenticate;
if(www_auth == NULL)
break;
scheme = www_auth->au_scheme;
realm = msg_params_find(www_auth->au_params, "realm=");
} else {
sip_proxy_authenticate_t const* proxy_auth = sip->sip_proxy_authenticate;
if(proxy_auth == NULL)
break;
scheme = proxy_auth->au_scheme;
realm = msg_params_find(proxy_auth->au_params, "realm=");
}
char authuser[100], secret[100];
memset(authuser, 0, sizeof(authuser));
memset(secret, 0, sizeof(secret));
if(session->helper) {
if(session->master == NULL) {
JANUS_LOG(LOG_WARN, "No master session for this helper, authentication will fail...\n");
} else {
session = session->master;
}
}
if(session->account.authuser && strchr(session->account.authuser, ':'))
g_snprintf(authuser, sizeof(authuser), "\"%s\"", session->account.authuser);
else
g_snprintf(authuser, sizeof(authuser), "%s", session->account.authuser);
if(session->account.secret && strchr(session->account.secret, ':'))
g_snprintf(secret, sizeof(secret), "\"%s\"", session->account.secret);
else
g_snprintf(secret, sizeof(secret), "%s", session->account.secret);
char auth[256];
memset(auth, 0, sizeof(auth));
g_snprintf(auth, sizeof(auth), "%s%s:%s:%s:%s%s",
session->account.secret_type == janus_sip_secret_type_hashed ? "HA1+" : "",
scheme,
realm,
authuser,
session->account.secret_type == janus_sip_secret_type_hashed ? "HA1+" : "",
secret);
JANUS_LOG(LOG_VERB, "\t%s\n", auth);
nua_authenticate(nh, NUTAG_AUTH(auth), TAG_END());
} else if(status >= 400) {
JANUS_LOG(LOG_WARN, "[%s] PUBLISH failed: %d %s\n", session->account.username, status, phrase ? phrase : "");
Copy link
Member

@lminiero lminiero Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this a tempoary log line? We don't have indicators of failure for any of the other requests.

Copy link
Contributor Author

@digiboridev digiboridev Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I don't understand what you mean.

Copy link
Member

@lminiero lminiero Nov 13, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I meant that we don't print LOG_WARN errors when other requests fail, so I don't see why we should for PUBLISH.

Copy link
Contributor Author

@digiboridev digiboridev Nov 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but SUBSCRIBE has, on line 6806

json_t *eventj = json_object();
json_object_set_new(eventj, "sip", json_string("event"));
if(sip && sip->sip_call_id)
json_object_set_new(eventj, "call_id", json_string(sip->sip_call_id->i_id));
json_t *resultj = json_object();
json_object_set_new(resultj, "event", json_string("publish_failed"));
json_object_set_new(resultj, "code", json_integer(status));
json_object_set_new(resultj, "reason", json_string(phrase ? phrase : ""));
if(session->incoming_header_prefixes) {
json_t *headers = janus_sip_get_incoming_headers(sip, session);
json_object_set_new(resultj, "headers", headers);
}
json_object_set_new(eventj, "result", resultj);
int ret = gateway->push_event(session->handle, &janus_sip_plugin, session->transaction, eventj, NULL);
JANUS_LOG(LOG_VERB, " >> Pushing event to peer: %d (%s)\n", ret, janus_get_api_error(ret));
json_decref(eventj);
}
break;
}
case nua_r_refer: {
JANUS_LOG(LOG_VERB, "[%s][%s]: %d %s\n", session->account.username, nua_event_name(event), status, phrase ? phrase : "??");
/* We got a response to our REFER */
Expand Down Expand Up @@ -7618,6 +7882,7 @@ gpointer janus_sip_sofia_thread(gpointer user_data) {
session->stack->s_nh_m = NULL;
session->stack->s_root = su_root_create(session->stack);
session->stack->subscriptions = NULL;
session->stack->publishers = NULL;
janus_mutex_init(&session->stack->smutex);
JANUS_LOG(LOG_VERB, "Setting up sofia stack (sip:%s@%s)\n", session->account.username, local_ip);
char sip_url[128];
Expand All @@ -7637,7 +7902,7 @@ gpointer janus_sip_sofia_thread(gpointer user_data) {
session->stack->s_nua = nua_create(session->stack->s_root,
janus_sip_sofia_callback,
session,
SIPTAG_ALLOW_STR("INVITE, ACK, BYE, CANCEL, OPTIONS, REFER, MESSAGE, INFO, NOTIFY"),
SIPTAG_ALLOW_STR("INVITE, ACK, BYE, CANCEL, OPTIONS, REFER, MESSAGE, INFO, NOTIFY, PUBLISH"),
NUTAG_M_USERNAME(session->account.username),
NUTAG_URL(sip_url),
TAG_IF(session->account.sips, NUTAG_SIPS_URL(sips_url)),
Expand Down Expand Up @@ -7675,6 +7940,9 @@ gpointer janus_sip_sofia_thread(gpointer user_data) {
if(session->stack->subscriptions != NULL)
g_hash_table_unref(session->stack->subscriptions);
session->stack->subscriptions = NULL;
if(session->stack->publishers != NULL)
g_hash_table_unref(session->stack->publishers);
session->stack->publishers = NULL;
janus_mutex_unlock(&session->stack->smutex);
nua_destroy(s_nua);
su_root_destroy(session->stack->s_root);
Expand Down