wordpress siren/aso优化服务站

1. Janus启动及线程

1.1 线程处理代码

主入口main函数位于 janus.c 文件中，下面为线程部分：

其中：需要特别注意~~janus_sessions_watchdog~~、janus_transport_requests、janus_transport_task几个处理函数

1.2 Session处理函数janus_sessions_watchdog（非本文重点）

1.3 请求处理函数janus_transport_requests

1.4 异步请求处理函数janus_transport_task

2. Janus启动及组件

2.1 组件回调处理

2.1.1 Plugin回调处理

2.1.2 Transport回调处理

2.2 EventHandlers处理代码

详细代码：

	/* Load event handlers */const char *path = NULL;DIR *dir = NULL;/* Event handlers are disabled by default, though: they need to be enabled in the configuration */item = janus_config_get(config, config_events, janus_config_type_item, "broadcast");gboolean enable_events = FALSE;if(item && item->value)enable_events = janus_is_true(item->value);if(!enable_events) {JANUS_LOG(LOG_WARN, "Event handlers support disabled\n");} else {gchar **disabled_eventhandlers = NULL;path = EVENTDIR;item = janus_config_get(config, config_general, janus_config_type_item, "events_folder");if(item && item->value)path = (char *)item->value;JANUS_LOG(LOG_INFO, "Event handler plugins folder: %s\n", path);dir = opendir(path);if(!dir) {/* Not really fatal, we don't care and go on anyway: event handlers are not fundamental */JANUS_LOG(LOG_FATAL, "\tCouldn't access event handler plugins folder...\n");} else {item = janus_config_get(config, config_events, janus_config_type_item, "stats_period");if(item && item->value) {/* Check if we need to use a larger period for pushing statistics to event handlers */int period = atoi(item->value);if(period < 0) {JANUS_LOG(LOG_WARN, "Invalid event handlers statistics period, using default value (1 second)\n");} else if(period == 0) {janus_ice_set_event_stats_period(0);JANUS_LOG(LOG_WARN, "Disabling event handlers statistics period, no media statistics will be pushed to event handlers\n");} else {janus_ice_set_event_stats_period(period);JANUS_LOG(LOG_INFO, "Setting event handlers statistics period to %d seconds\n", period);}}/* Any event handlers to ignore? */item = janus_config_get(config, config_events, janus_config_type_item, "disable");if(item && item->value)disabled_eventhandlers = g_strsplit(item->value, ",", -1);/* Open the shared objects */struct dirent *eventent = NULL;char eventpath[1024];while((eventent = readdir(dir))) {int len = strlen(eventent->d_name);if (len < 4) {continue;}if (strcasecmp(eventent->d_name+len-strlen(SHLIB_EXT), SHLIB_EXT)) {continue;}/* Check if this event handler has been disabled in the configuration file */if(disabled_eventhandlers != NULL) {gchar *index = disabled_eventhandlers[0];if(index != NULL) {int i=0;gboolean skip = FALSE;while(index != NULL) {while(isspace(*index))index++;if(strlen(index) && !strcmp(index, eventent->d_name)) {JANUS_LOG(LOG_WARN, "Event handler plugin '%s' has been disabled, skipping...\n", eventent->d_name);skip = TRUE;break;}i++;index = disabled_eventhandlers[i];}if(skip)continue;}}JANUS_LOG(LOG_INFO, "Loading event handler plugin '%s'...\n", eventent->d_name);memset(eventpath, 0, 1024);g_snprintf(eventpath, 1024, "%s/%s", path, eventent->d_name);void *event = dlopen(eventpath, RTLD_NOW | RTLD_GLOBAL);if (!event) {JANUS_LOG(LOG_ERR, "\tCouldn't load event handler plugin '%s': %s\n", eventent->d_name, dlerror());} else {create_e *create = (create_e*) dlsym(event, "create");const char *dlsym_error = dlerror();if (dlsym_error) {JANUS_LOG(LOG_ERR, "\tCouldn't load symbol 'create': %s\n", dlsym_error);continue;}janus_eventhandler *janus_eventhandler = create();if(!janus_eventhandler) {JANUS_LOG(LOG_ERR, "\tCouldn't use function 'create'...\n");continue;}/* Are all the mandatory methods and callbacks implemented? */if(!janus_eventhandler->init || !janus_eventhandler->destroy ||!janus_eventhandler->get_api_compatibility ||!janus_eventhandler->get_version ||!janus_eventhandler->get_version_string ||!janus_eventhandler->get_description ||!janus_eventhandler->get_package ||!janus_eventhandler->get_name ||!janus_eventhandler->incoming_event) {JANUS_LOG(LOG_ERR, "\tMissing some mandatory methods/callbacks, skipping this event handler plugin...\n");continue;}if(janus_eventhandler->get_api_compatibility() < JANUS_EVENTHANDLER_API_VERSION) {JANUS_LOG(LOG_ERR, "The '%s' event handler plugin was compiled against an older version of the API (%d < %d), skipping it: update it to enable it again\n",janus_eventhandler->get_package(), janus_eventhandler->get_api_compatibility(), JANUS_EVENTHANDLER_API_VERSION);continue;}janus_eventhandler->init(configs_folder);JANUS_LOG(LOG_VERB, "\tVersion: %d (%s)\n", janus_eventhandler->get_version(), janus_eventhandler->get_version_string());JANUS_LOG(LOG_VERB, "\t   [%s] %s\n", janus_eventhandler->get_package(), janus_eventhandler->get_name());JANUS_LOG(LOG_VERB, "\t   %s\n", janus_eventhandler->get_description());JANUS_LOG(LOG_VERB, "\t   Plugin API version: %d\n", janus_eventhandler->get_api_compatibility());JANUS_LOG(LOG_VERB, "\t   Subscriptions:");if(janus_eventhandler->events_mask == 0) {JANUS_LOG(LOG_VERB, " none");} else {if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_SESSION))JANUS_LOG(LOG_VERB, " sessions");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_HANDLE))JANUS_LOG(LOG_VERB, " handles");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_JSEP))JANUS_LOG(LOG_VERB, " jsep");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_WEBRTC))JANUS_LOG(LOG_VERB, " webrtc");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_MEDIA))JANUS_LOG(LOG_VERB, " media");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_PLUGIN))JANUS_LOG(LOG_VERB, " plugins");if(janus_flags_is_set(&janus_eventhandler->events_mask, JANUS_EVENT_TYPE_TRANSPORT))JANUS_LOG(LOG_VERB, " transports");}JANUS_LOG(LOG_VERB, "\n");if(eventhandlers == NULL)eventhandlers = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(eventhandlers, (gpointer)janus_eventhandler->get_package(), janus_eventhandler);if(eventhandlers_so == NULL)eventhandlers_so = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(eventhandlers_so, (gpointer)janus_eventhandler->get_package(), event);}}}closedir(dir);if(disabled_eventhandlers != NULL)g_strfreev(disabled_eventhandlers);disabled_eventhandlers = NULL;/* Initialize the event broadcaster */if(janus_events_init(enable_events, (server_name ? server_name : (char *)JANUS_SERVER_NAME), eventhandlers) < 0) {JANUS_LOG(LOG_FATAL, "Error initializing the Event handlers mechanism...\n");exit(1);}}

其中，janus_events_init(enable_events, (server_name ? server_name : (char *)JANUS_SERVER_NAME), eventhandlers)初始化启动了全局的事件处理线程 janus_events_thread

2.3 Plugins处理代码

详细代码：

	/* Load plugins */path = PLUGINDIR;item = janus_config_get(config, config_general, janus_config_type_item, "plugins_folder");if(item && item->value)path = (char *)item->value;JANUS_LOG(LOG_INFO, "Plugins folder: %s\n", path);dir = opendir(path);if(!dir) {JANUS_LOG(LOG_FATAL, "\tCouldn't access plugins folder...\n");exit(1);}/* Any plugin to ignore? */gchar **disabled_plugins = NULL;item = janus_config_get(config, config_plugins, janus_config_type_item, "disable");if(item && item->value)disabled_plugins = g_strsplit(item->value, ",", -1);/* Open the shared objects */struct dirent *pluginent = NULL;char pluginpath[1024];while((pluginent = readdir(dir))) {int len = strlen(pluginent->d_name);if (len < 4) {continue;}if (strcasecmp(pluginent->d_name+len-strlen(SHLIB_EXT), SHLIB_EXT)) {continue;}/* Check if this plugins has been disabled in the configuration file */if(disabled_plugins != NULL) {gchar *index = disabled_plugins[0];if(index != NULL) {int i=0;gboolean skip = FALSE;while(index != NULL) {while(isspace(*index))index++;if(strlen(index) && !strcmp(index, pluginent->d_name)) {JANUS_LOG(LOG_WARN, "Plugin '%s' has been disabled, skipping...\n", pluginent->d_name);skip = TRUE;break;}i++;index = disabled_plugins[i];}if(skip)continue;}}JANUS_LOG(LOG_INFO, "Loading plugin '%s'...\n", pluginent->d_name);memset(pluginpath, 0, 1024);g_snprintf(pluginpath, 1024, "%s/%s", path, pluginent->d_name);void *plugin = dlopen(pluginpath, RTLD_NOW | RTLD_GLOBAL);if (!plugin) {JANUS_LOG(LOG_ERR, "\tCouldn't load plugin '%s': %s\n", pluginent->d_name, dlerror());} else {create_p *create = (create_p*) dlsym(plugin, "create");const char *dlsym_error = dlerror();if (dlsym_error) {JANUS_LOG(LOG_ERR, "\tCouldn't load symbol 'create': %s\n", dlsym_error);continue;}janus_plugin *janus_plugin = create();if(!janus_plugin) {JANUS_LOG(LOG_ERR, "\tCouldn't use function 'create'...\n");continue;}/* Are all the mandatory methods and callbacks implemented? */if(!janus_plugin->init || !janus_plugin->destroy ||!janus_plugin->get_api_compatibility ||!janus_plugin->get_version ||!janus_plugin->get_version_string ||!janus_plugin->get_description ||!janus_plugin->get_package ||!janus_plugin->get_name ||!janus_plugin->create_session ||!janus_plugin->query_session ||!janus_plugin->destroy_session ||!janus_plugin->handle_message ||!janus_plugin->setup_media ||!janus_plugin->hangup_media) {JANUS_LOG(LOG_ERR, "\tMissing some mandatory methods/callbacks, skipping this plugin...\n");continue;}if(janus_plugin->get_api_compatibility() < JANUS_PLUGIN_API_VERSION) {JANUS_LOG(LOG_ERR, "The '%s' plugin was compiled against an older version of the API (%d < %d), skipping it: update it to enable it again\n",janus_plugin->get_package(), janus_plugin->get_api_compatibility(), JANUS_PLUGIN_API_VERSION);continue;}if(janus_plugin->init(&janus_handler_plugin, configs_folder) < 0) {JANUS_LOG(LOG_WARN, "The '%s' plugin could not be initialized\n", janus_plugin->get_package());dlclose(plugin);continue;}JANUS_LOG(LOG_VERB, "\tVersion: %d (%s)\n", janus_plugin->get_version(), janus_plugin->get_version_string());JANUS_LOG(LOG_VERB, "\t   [%s] %s\n", janus_plugin->get_package(), janus_plugin->get_name());JANUS_LOG(LOG_VERB, "\t   %s\n", janus_plugin->get_description());JANUS_LOG(LOG_VERB, "\t   Plugin API version: %d\n", janus_plugin->get_api_compatibility());if(!janus_plugin->incoming_rtp && !janus_plugin->incoming_rtcp && !janus_plugin->incoming_data) {JANUS_LOG(LOG_WARN, "The '%s' plugin doesn't implement any callback for RTP/RTCP/data... is this on purpose?\n",janus_plugin->get_package());}if(!janus_plugin->incoming_rtp && !janus_plugin->incoming_rtcp && janus_plugin->incoming_data) {JANUS_LOG(LOG_WARN, "The '%s' plugin will only handle data channels (no RTP/RTCP)... is this on purpose?\n",janus_plugin->get_package());}if(plugins == NULL)plugins = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(plugins, (gpointer)janus_plugin->get_package(), janus_plugin);if(plugins_so == NULL)plugins_so = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(plugins_so, (gpointer)janus_plugin->get_package(), plugin);}}closedir(dir);if(disabled_plugins != NULL)g_strfreev(disabled_plugins);disabled_plugins = NULL;

其中，janus_plugin->init(&janus_handler_plugin, configs_folder)初始化设置了插件的回调处理 janus_handler_plugin

2.4 Transports处理代码

详细代码：

	/* Load transports */gboolean janus_api_enabled = FALSE, admin_api_enabled = FALSE;path = TRANSPORTDIR;item = janus_config_get(config, config_general, janus_config_type_item, "transports_folder");if(item && item->value)path = (char *)item->value;JANUS_LOG(LOG_INFO, "Transport plugins folder: %s\n", path);dir = opendir(path);if(!dir) {JANUS_LOG(LOG_FATAL, "\tCouldn't access transport plugins folder...\n");exit(1);}/* Any transport to ignore? */gchar **disabled_transports = NULL;item = janus_config_get(config, config_transports, janus_config_type_item, "disable");if(item && item->value)disabled_transports = g_strsplit(item->value, ",", -1);/* Open the shared objects */struct dirent *transportent = NULL;char transportpath[1024];while((transportent = readdir(dir))) {int len = strlen(transportent->d_name);if (len < 4) {continue;}if (strcasecmp(transportent->d_name+len-strlen(SHLIB_EXT), SHLIB_EXT)) {continue;}/* Check if this transports has been disabled in the configuration file */if(disabled_transports != NULL) {gchar *index = disabled_transports[0];if(index != NULL) {int i=0;gboolean skip = FALSE;while(index != NULL) {while(isspace(*index))index++;if(strlen(index) && !strcmp(index, transportent->d_name)) {JANUS_LOG(LOG_WARN, "Transport plugin '%s' has been disabled, skipping...\n", transportent->d_name);skip = TRUE;break;}i++;index = disabled_transports[i];}if(skip)continue;}}JANUS_LOG(LOG_INFO, "Loading transport plugin '%s'...\n", transportent->d_name);memset(transportpath, 0, 1024);g_snprintf(transportpath, 1024, "%s/%s", path, transportent->d_name);void *transport = dlopen(transportpath, RTLD_NOW | RTLD_GLOBAL);if (!transport) {JANUS_LOG(LOG_ERR, "\tCouldn't load transport plugin '%s': %s\n", transportent->d_name, dlerror());} else {create_t *create = (create_t*) dlsym(transport, "create");const char *dlsym_error = dlerror();if (dlsym_error) {JANUS_LOG(LOG_ERR, "\tCouldn't load symbol 'create': %s\n", dlsym_error);continue;}janus_transport *janus_transport = create();if(!janus_transport) {JANUS_LOG(LOG_ERR, "\tCouldn't use function 'create'...\n");continue;}/* Are all the mandatory methods and callbacks implemented? */if(!janus_transport->init || !janus_transport->destroy ||!janus_transport->get_api_compatibility ||!janus_transport->get_version ||!janus_transport->get_version_string ||!janus_transport->get_description ||!janus_transport->get_package ||!janus_transport->get_name ||!janus_transport->send_message ||!janus_transport->is_janus_api_enabled ||!janus_transport->is_admin_api_enabled ||!janus_transport->session_created ||!janus_transport->session_over ||!janus_transport->session_claimed) {JANUS_LOG(LOG_ERR, "\tMissing some mandatory methods/callbacks, skipping this transport plugin...\n");continue;}if(janus_transport->get_api_compatibility() < JANUS_TRANSPORT_API_VERSION) {JANUS_LOG(LOG_ERR, "The '%s' transport plugin was compiled against an older version of the API (%d < %d), skipping it: update it to enable it again\n",janus_transport->get_package(), janus_transport->get_api_compatibility(), JANUS_TRANSPORT_API_VERSION);continue;}if(janus_transport->init(&janus_handler_transport, configs_folder) < 0) {JANUS_LOG(LOG_WARN, "The '%s' plugin could not be initialized\n", janus_transport->get_package());dlclose(transport);continue;}JANUS_LOG(LOG_VERB, "\tVersion: %d (%s)\n", janus_transport->get_version(), janus_transport->get_version_string());JANUS_LOG(LOG_VERB, "\t   [%s] %s\n", janus_transport->get_package(), janus_transport->get_name());JANUS_LOG(LOG_VERB, "\t   %s\n", janus_transport->get_description());JANUS_LOG(LOG_VERB, "\t   Plugin API version: %d\n", janus_transport->get_api_compatibility());JANUS_LOG(LOG_VERB, "\t   Janus API: %s\n", janus_transport->is_janus_api_enabled() ? "enabled" : "disabled");JANUS_LOG(LOG_VERB, "\t   Admin API: %s\n", janus_transport->is_admin_api_enabled() ? "enabled" : "disabled");janus_api_enabled = janus_api_enabled || janus_transport->is_janus_api_enabled();admin_api_enabled = admin_api_enabled || janus_transport->is_admin_api_enabled();if(transports == NULL)transports = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(transports, (gpointer)janus_transport->get_package(), janus_transport);if(transports_so == NULL)transports_so = g_hash_table_new(g_str_hash, g_str_equal);g_hash_table_insert(transports_so, (gpointer)janus_transport->get_package(), transport);}}closedir(dir);if(disabled_transports != NULL)g_strfreev(disabled_transports);disabled_transports = NULL;

其中，janus_transport->init(&janus_handler_transport, configs_folder)初始化设置了Transport的回调处理 janus_handler_transport

3. 请求处理过程

过程如下所述：

1. Janus客户端连接Transport发起请求；
2. Transport收到请求后，执行janus_xxx_handler处理，调用Janus主线程的回调函数incoming_request;
3. Janus主线程的incoming_request将请求加入请求队列;
4. 请求处理线程从请求队列中获取请求信息，执行janus_transport_requests处理；
   • 若请求需同步处理，请求处理线程执行janus_process_incoming_request，调用Plugin的handle_message函数进行处理；
   • 若请求需异步处理，请求处理线程将请求加入任务列表，任务处理线程池执行janus_transport_task函数，执行janus_process_incoming_request，调用Plugin的handle_message函数进行处理；
5. Plugin的handle_message将处理结果加入Plugin消息队列;
6. Plugin消息处理线程从Plugin消息队列获取处理结果，执行janus_yyy_handler处理，调用Janus主线程的回调函数push_event；
7. Janus主线程的push_event调用Transport的send_message进行处理；
8. Transport将处理完的结果返回给Janus客户端；

3.1 Transport启动服务时设置请求处理函数（以transport/janus_http.c为例）

janus_http_handler 详细代码：

/* WebServer requests handler */
int janus_http_handler(void *cls, struct MHD_Connection *connection, const char *url, const char *method, const char *version, const char *upload_data, size_t *upload_data_size, void **ptr)
{char *payload = NULL;json_t *root = NULL;struct MHD_Response *response = NULL;int ret = MHD_NO;gchar *session_path = NULL, *handle_path = NULL;gchar **basepath = NULL, **path = NULL;guint64 session_id = 0, handle_id = 0;/* Is this the first round? */int firstround = 0;janus_transport_session *ts = (janus_transport_session *)*ptr;janus_http_msg *msg = NULL;if(ts == NULL) {firstround = 1;JANUS_LOG(LOG_DBG, "Got a HTTP %s request on %s...\n", method, url);JANUS_LOG(LOG_DBG, " ... Just parsing headers for now...\n");msg = g_malloc0(sizeof(janus_http_msg));msg->connection = connection;janus_mutex_init(&msg->wait_mutex);janus_condition_init(&msg->wait_cond);ts = janus_transport_session_create(msg, janus_http_msg_destroy);janus_mutex_lock(&messages_mutex);g_hash_table_insert(messages, ts, ts);janus_mutex_unlock(&messages_mutex);*ptr = ts;MHD_get_connection_values(connection, MHD_HEADER_KIND, &janus_http_headers, msg);ret = MHD_YES;/* Notify handlers about this new transport instance */if(notify_events && gateway->events_is_enabled()) {json_t *info = json_object();json_object_set_new(info, "event", json_string("request"));json_object_set_new(info, "admin_api", json_false());const union MHD_ConnectionInfo *conninfo = MHD_get_connection_info(connection, MHD_CONNECTION_INFO_CLIENT_ADDRESS);if(conninfo != NULL) {janus_network_address addr;janus_network_address_string_buffer addr_buf;if(janus_network_address_from_sockaddr((struct sockaddr *)conninfo->client_addr, &addr) == 0 &&janus_network_address_to_string_buffer(&addr, &addr_buf) == 0) {const char *ip = janus_network_address_string_from_buffer(&addr_buf);json_object_set_new(info, "ip", json_string(ip));}uint16_t port = janus_http_sockaddr_to_port((struct sockaddr *)conninfo->client_addr);json_object_set_new(info, "port", json_integer(port));}gateway->notify_event(&janus_http_transport, ts, info);}} else {JANUS_LOG(LOG_DBG, "Processing HTTP %s request on %s...\n", method, url);msg = (janus_http_msg *)ts->transport_p;}/* Parse request */if (strcasecmp(method, "GET") && strcasecmp(method, "POST") && strcasecmp(method, "OPTIONS")) {ret = janus_http_return_error(ts, 0, NULL, JANUS_ERROR_TRANSPORT_SPECIFIC, "Unsupported method %s", method);goto done;}if (!strcasecmp(method, "OPTIONS")) {response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_OK, response);MHD_destroy_response(response);}/* Get path components */if(strcasecmp(url, ws_path)) {if(strlen(ws_path) > 1) {basepath = g_strsplit(url, ws_path, -1);} else {/* The base path is the web server too itself, we process the url itself */basepath = g_malloc_n(3, sizeof(char *));basepath[0] = g_strdup("/");basepath[1] = g_strdup(url);basepath[2] = NULL;}if(basepath[0] == NULL || basepath[1] == NULL || basepath[1][0] != '/') {JANUS_LOG(LOG_ERR, "Invalid url %s\n", url);response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_NOT_FOUND, response);MHD_destroy_response(response);}if(firstround) {g_strfreev(basepath);return ret;}path = g_strsplit(basepath[1], "/", -1);if(path == NULL || path[1] == NULL) {JANUS_LOG(LOG_ERR, "Invalid path %s (%s)\n", basepath[1], path ? path[1] : "");response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_NOT_FOUND, response);MHD_destroy_response(response);g_strfreev(basepath);g_strfreev(path);return ret;}}if(firstround) {g_strfreev(basepath);g_strfreev(path);return ret;}JANUS_LOG(LOG_DBG, " ... parsing request...\n");if(path != NULL && path[1] != NULL && strlen(path[1]) > 0) {session_path = g_strdup(path[1]);JANUS_LOG(LOG_HUGE, "Session: %s\n", session_path);}if(session_path != NULL && path[2] != NULL && strlen(path[2]) > 0) {handle_path = g_strdup(path[2]);JANUS_LOG(LOG_HUGE, "Handle: %s\n", handle_path);}if(session_path != NULL && handle_path != NULL && path[3] != NULL && strlen(path[3]) > 0) {JANUS_LOG(LOG_ERR, "Too many components...\n");response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_NOT_FOUND, response);MHD_destroy_response(response);goto done;}/* Get payload, if any */if(!strcasecmp(method, "POST")) {JANUS_LOG(LOG_HUGE, "Processing POST data (%s) (%zu bytes)...\n", msg->contenttype, *upload_data_size);if(*upload_data_size != 0) {msg->payload = g_realloc(msg->payload, msg->len+*upload_data_size+1);memcpy(msg->payload+msg->len, upload_data, *upload_data_size);msg->len += *upload_data_size;memset(msg->payload + msg->len, '\0', 1);JANUS_LOG(LOG_DBG, "  -- Data we have now (%zu bytes)\n", msg->len);*upload_data_size = 0;	/* Go on */ret = MHD_YES;goto done;}JANUS_LOG(LOG_DBG, "Done getting payload, we can answer\n");if(msg->payload == NULL) {JANUS_LOG(LOG_ERR, "No payload :-(\n");ret = MHD_NO;goto done;}payload = msg->payload;JANUS_LOG(LOG_HUGE, "%s\n", payload);}/* Is this a generic request for info? */if(session_path != NULL && !strcmp(session_path, "info")) {/* The info REST endpoint, if contacted through a GET, provides information on the Janus core */if(strcasecmp(method, "GET")) {response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_BAD_REQUEST, response);MHD_destroy_response(response);goto done;}/* Turn this into a fake "info" request */method = "POST";char tr[12];janus_http_random_string(12, (char *)&tr);root = json_object();json_object_set_new(root, "janus", json_string("info"));json_object_set_new(root, "transaction", json_string(tr));goto parsingdone;}/* Or maybe a long poll */if(!strcasecmp(method, "GET") || !payload) {session_id = session_path ? g_ascii_strtoull(session_path, NULL, 10) : 0;if(session_id < 1) {JANUS_LOG(LOG_ERR, "Invalid session %s\n", session_path);response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_NOT_FOUND, response);MHD_destroy_response(response);goto done;}msg->session_id = session_id;/* Since we handle long polls ourselves, the core isn't involved (if not for providing us with events)* A long poll, though, can act as a keepalive, so we pass a fake one to the core to avoid undesirable timeouts *//* First of all, though, API secret and token based authentication may be enabled in the core, so since* we're bypassing it for notifications we'll have to check those ourselves */const char *secret = MHD_lookup_connection_value(connection, MHD_GET_ARGUMENT_KIND, "apisecret");const char *token = MHD_lookup_connection_value(connection, MHD_GET_ARGUMENT_KIND, "token");gboolean secret_authorized = FALSE, token_authorized = FALSE;if(!gateway->is_api_secret_needed(&janus_http_transport) && !gateway->is_auth_token_needed(&janus_http_transport)) {/* Nothing to check */secret_authorized = TRUE;token_authorized = TRUE;} else {if(gateway->is_api_secret_valid(&janus_http_transport, secret)) {/* API secret is valid */secret_authorized = TRUE;}if(gateway->is_auth_token_valid(&janus_http_transport, token)) {/* Token is valid */token_authorized = TRUE;}/* We consider a request authorized if either the proper API secret or a valid token has been provided */if(!secret_authorized && !token_authorized) {response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_FORBIDDEN, response);MHD_destroy_response(response);goto done;}}/* Ok, go on with the keepalive */char tr[12];janus_http_random_string(12, (char *)&tr);root = json_object();json_object_set_new(root, "janus", json_string("keepalive"));json_object_set_new(root, "session_id", json_integer(session_id));json_object_set_new(root, "transaction", json_string(tr));if(secret)json_object_set_new(root, "apisecret", json_string(secret));if(token)json_object_set_new(root, "token", json_string(token));gateway->incoming_request(&janus_http_transport, ts, (void *)keepalive_id, FALSE, root, NULL);/* Ok, go on */if(handle_path) {char *location = g_malloc(strlen(ws_path) + strlen(session_path) + 2);g_sprintf(location, "%s/%s", ws_path, session_path);JANUS_LOG(LOG_ERR, "Invalid GET to %s, redirecting to %s\n", url, location);response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);MHD_add_response_header(response, "Location", location);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, 302, response);MHD_destroy_response(response);g_free(location);goto done;}janus_mutex_lock(&sessions_mutex);janus_http_session *session = g_hash_table_lookup(sessions, &session_id);janus_mutex_unlock(&sessions_mutex);if(!session || g_atomic_int_get(&session->destroyed)) {JANUS_LOG(LOG_ERR, "Couldn't find any session %"SCNu64"...\n", session_id);response = MHD_create_response_from_buffer(0, NULL, MHD_RESPMEM_PERSISTENT);janus_http_add_cors_headers(msg, response);ret = MHD_queue_response(connection, MHD_HTTP_NOT_FOUND, response);MHD_destroy_response(response);goto done;}janus_refcount_increase(&ts->ref);janus_refcount_increase(&session->ref);/* How many messages can we send back in a single response? (just one by default) */int max_events = 1;const char *maxev = MHD_lookup_connection_value(connection, MHD_GET_ARGUMENT_KIND, "maxev");if(maxev != NULL) {max_events = atoi(maxev);if(max_events < 1) {JANUS_LOG(LOG_WARN, "Invalid maxev parameter passed (%d), defaulting to 1\n", max_events);max_events = 1;}}JANUS_LOG(LOG_VERB, "Session %"SCNu64" found... returning up to %d messages\n", session_id, max_events);/* Handle GET, taking the first message from the list */json_t *event = g_async_queue_try_pop(session->events);if(event != NULL) {if(max_events == 1) {/* Return just this message and leave */gchar *event_text = json_dumps(event, json_format);json_decref(event);ret = janus_http_return_success(ts, event_text);} else {/* The application is willing to receive more events at the same time, anything to report? */json_t *list = json_array();json_array_append_new(list, event);int events = 1;while(events < max_events) {event = g_async_queue_try_pop(session->events);if(event == NULL)break;json_array_append_new(list, event);events++;}/* Return the array of messages and leave */gchar *list_text = json_dumps(list, json_format);json_decref(list);ret = janus_http_return_success(ts, list_text);}} else {/* Still no message, wait */ret = janus_http_notifier(ts, session, max_events);}janus_refcount_decrease(&session->ref);janus_refcount_decrease(&ts->ref);goto done;}json_error_t error;/* Parse the JSON payload */root = json_loads(payload, 0, &error);if(!root) {ret = janus_http_return_error(ts, 0, NULL, JANUS_ERROR_INVALID_JSON, "JSON error: on line %d: %s", error.line, error.text);goto done;}if(!json_is_object(root)) {ret = janus_http_return_error(ts, 0, NULL, JANUS_ERROR_INVALID_JSON_OBJECT, "JSON error: not an object");json_decref(root);goto done;}parsingdone:/* Check if we have session and handle identifiers, and how they were provided */session_id = json_integer_value(json_object_get(root, "session_id"));if(session_id && session_path && g_ascii_strtoull(session_path, NULL, 10) != session_id) {ret = janus_http_return_error(ts, 0, NULL, JANUS_ERROR_INVALID_REQUEST_PATH, "Conflicting session ID (payload and path)");json_decref(root);goto done;}if(!session_id) {/* No session ID in the JSON object, maybe in the path? */session_id = session_path ? g_ascii_strtoull(session_path, NULL, 10) : 0;if(session_id > 0)json_object_set_new(root, "session_id", json_integer(session_id));}handle_id = json_integer_value(json_object_get(root, "handle_id"));if(handle_id && handle_path && g_ascii_strtoull(handle_path, NULL, 10) != handle_id) {ret = janus_http_return_error(ts, 0, NULL, JANUS_ERROR_INVALID_REQUEST_PATH, "Conflicting handle ID (payload and path)");json_decref(root);goto done;}if(!handle_id) {/* No session ID in the JSON object, maybe in the path? */handle_id = handle_path ? g_ascii_strtoull(handle_path, NULL, 10) : 0;if(handle_id > 0)json_object_set_new(root, "handle_id", json_integer(handle_id));}/* Suspend the connection and pass the ball to the core */JANUS_LOG(LOG_HUGE, "Forwarding request to the core (%p)\n", ts);gateway->incoming_request(&janus_http_transport, ts, ts, FALSE, root, &error);/* Wait for a response (but not forever) */
#ifndef USE_PTHREAD_MUTEXgint64 wakeup = janus_get_monotonic_time() + 10*G_TIME_SPAN_SECOND;janus_mutex_lock(&msg->wait_mutex);while(!msg->got_response) {int res = janus_condition_wait_until(&msg->wait_cond, &msg->wait_mutex, wakeup);if(msg->got_response || !res)break;}janus_mutex_unlock(&msg->wait_mutex);
#elsestruct timeval now;gettimeofday(&now, NULL);struct timespec wakeup;wakeup.tv_sec = now.tv_sec+10;	/* Wait at max 10 seconds for a response */wakeup.tv_nsec = now.tv_usec*1000UL;janus_mutex_lock(&msg->wait_mutex);while(!msg->got_response) {int res = janus_condition_timedwait(&msg->wait_cond, &msg->wait_mutex, &wakeup);if(msg->got_response || res == ETIMEDOUT)break;}janus_mutex_unlock(&msg->wait_mutex);
#endifif(!msg->response) {ret = MHD_NO;} else {char *response_text = json_dumps(msg->response, json_format);json_decref(msg->response);msg->response = NULL;ret = janus_http_return_success(ts, response_text);}done:g_strfreev(basepath);g_strfreev(path);g_free(session_path);g_free(handle_path);return ret;
}

回调 incoming_request ，并等待响应直至返回结果或超时：

3.2 Janus将请求加入请求队列

3.3 请求处理线程处理函数janus_transport_requests

• 若是管理请求，执行janus_process_incoming_admin_request同步处理；
• 若不是管理请求
  • janus==message，将请求加入任务队列，采用janus_transport_task异步处理，调用janus_process_incoming_request进行处理；
  • janus!=message，执行janus_process_incoming_request同步处理；

3.4 处理请求函数janus_process_incoming_request

int janus_process_incoming_request(janus_request *request) {int ret = -1;if(request == NULL) {JANUS_LOG(LOG_ERR, "Missing request or payload to process, giving up...\n");return ret;}int error_code = 0;char error_cause[100];json_t *root = request->message;/* Ok, let's start with the ids */guint64 session_id = 0, handle_id = 0;json_t *s = json_object_get(root, "session_id");if(s && json_is_integer(s))session_id = json_integer_value(s);json_t *h = json_object_get(root, "handle_id");if(h && json_is_integer(h))handle_id = json_integer_value(h);janus_session *session = NULL;janus_ice_handle *handle = NULL;/* Get transaction and message request */JANUS_VALIDATE_JSON_OBJECT(root, incoming_request_parameters,error_code, error_cause, FALSE,JANUS_ERROR_MISSING_MANDATORY_ELEMENT, JANUS_ERROR_INVALID_ELEMENT_TYPE);if(error_code != 0) {ret = janus_process_error_string(request, session_id, NULL, error_code, error_cause);goto jsondone;}json_t *transaction = json_object_get(root, "transaction");const gchar *transaction_text = json_string_value(transaction);json_t *message = json_object_get(root, "janus");const gchar *message_text = json_string_value(message);if(session_id == 0 && handle_id == 0) {/* Can only be a 'Create new session', a 'Get info' or a 'Ping/Pong' request */if(!strcasecmp(message_text, "info")) {ret = janus_process_success(request, janus_info(transaction_text));goto jsondone;}if(!strcasecmp(message_text, "ping")) {/* Prepare JSON reply */json_t *reply = janus_create_message("pong", 0, transaction_text);ret = janus_process_success(request, reply);goto jsondone;}if(strcasecmp(message_text, "create")) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}/* Make sure we're accepting new sessions */if(!accept_new_sessions) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_NOT_ACCEPTING_SESSIONS, NULL);goto jsondone;}/* Any secret/token to check? */ret = janus_request_check_secret(request, session_id, transaction_text);if(ret != 0) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNAUTHORIZED, NULL);goto jsondone;}session_id = 0;json_t *id = json_object_get(root, "id");if(id != NULL) {/* The application provided the session ID to use */session_id = json_integer_value(id);if(session_id > 0 && (session = janus_session_find(session_id)) != NULL) {/* Session ID already taken */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_SESSION_CONFLICT, "Session ID already in use");goto jsondone;}}/* Handle it */session = janus_session_create(session_id);if(session == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNKNOWN, "Memory error");goto jsondone;}session_id = session->session_id;/* We increase the counter as this request is using the session */janus_refcount_increase(&session->ref);/* Take note of the request source that originated this session (HTTP, WebSockets, RabbitMQ?) */session->source = janus_request_new(request->transport, request->instance, NULL, FALSE, NULL);/* Notify the source that a new session has been created */request->transport->session_created(request->instance, session->session_id);/* Notify event handlers */if(janus_events_is_enabled()) {/* Session created, add info on the transport that originated it */json_t *transport = json_object();json_object_set_new(transport, "transport", json_string(session->source->transport->get_package()));char id[32];memset(id, 0, sizeof(id));g_snprintf(id, sizeof(id), "%p", session->source->instance);json_object_set_new(transport, "id", json_string(id));janus_events_notify_handlers(JANUS_EVENT_TYPE_SESSION, session_id, "created", transport);}/* Prepare JSON reply */json_t *reply = janus_create_message("success", 0, transaction_text);json_t *data = json_object();json_object_set_new(data, "id", json_integer(session_id));json_object_set_new(reply, "data", data);/* Send the success reply */ret = janus_process_success(request, reply);goto jsondone;}if(session_id < 1) {JANUS_LOG(LOG_ERR, "Invalid session\n");ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_SESSION_NOT_FOUND, NULL);goto jsondone;}if(h && handle_id < 1) {JANUS_LOG(LOG_ERR, "Invalid handle\n");ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_SESSION_NOT_FOUND, NULL);goto jsondone;}/* Go on with the processing */ret = janus_request_check_secret(request, session_id, transaction_text);if(ret != 0) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNAUTHORIZED, NULL);goto jsondone;}/* If we got here, make sure we have a session (and/or a handle) */session = janus_session_find(session_id);if(!session) {JANUS_LOG(LOG_ERR, "Couldn't find any session %"SCNu64"...\n", session_id);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_SESSION_NOT_FOUND, "No such session %"SCNu64"", session_id);goto jsondone;}/* Update the last activity timer */session->last_activity = janus_get_monotonic_time();handle = NULL;if(handle_id > 0) {handle = janus_session_handles_find(session, handle_id);if(!handle) {JANUS_LOG(LOG_ERR, "Couldn't find any handle %"SCNu64" in session %"SCNu64"...\n", handle_id, session_id);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_HANDLE_NOT_FOUND, "No such handle %"SCNu64" in session %"SCNu64"", handle_id, session_id);goto jsondone;}}/* What is this? */if(!strcasecmp(message_text, "keepalive")) {/* Just a keep-alive message, reply with an ack */JANUS_LOG(LOG_VERB, "Got a keep-alive on session %"SCNu64"\n", session_id);json_t *reply = janus_create_message("ack", session_id, transaction_text);/* Send the success reply */ret = janus_process_success(request, reply);} else if(!strcasecmp(message_text, "attach")) {if(handle != NULL) {/* Attach is a session-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}JANUS_VALIDATE_JSON_OBJECT(root, attach_parameters,error_code, error_cause, FALSE,JANUS_ERROR_MISSING_MANDATORY_ELEMENT, JANUS_ERROR_INVALID_ELEMENT_TYPE);if(error_code != 0) {ret = janus_process_error_string(request, session_id, transaction_text, error_code, error_cause);goto jsondone;}json_t *plugin = json_object_get(root, "plugin");const gchar *plugin_text = json_string_value(plugin);janus_plugin *plugin_t = janus_plugin_find(plugin_text);if(plugin_t == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_NOT_FOUND, "No such plugin '%s'", plugin_text);goto jsondone;}/* If the auth token mechanism is enabled, we should check if this token can access this plugin */if(janus_auth_is_enabled()) {json_t *token = json_object_get(root, "token");if(token != NULL) {const char *token_value = json_string_value(token);if(token_value && !janus_auth_check_plugin(token_value, plugin_t)) {JANUS_LOG(LOG_ERR, "Token '%s' can't access plugin '%s'\n", token_value, plugin_text);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNAUTHORIZED_PLUGIN, "Provided token can't access plugin '%s'", plugin_text);goto jsondone;}}}json_t *opaque = json_object_get(root, "opaque_id");const char *opaque_id = opaque ? json_string_value(opaque) : NULL;/* Create handle */handle = janus_ice_handle_create(session, opaque_id);if(handle == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNKNOWN, "Memory error");goto jsondone;}handle_id = handle->handle_id;/* We increase the counter as this request is using the handle */janus_refcount_increase(&handle->ref);/* Attach to the plugin */int error = 0;if((error = janus_ice_handle_attach_plugin(session, handle, plugin_t)) != 0) {/* TODO Make error struct to pass verbose information */janus_session_handles_remove(session, handle);JANUS_LOG(LOG_ERR, "Couldn't attach to plugin '%s', error '%d'\n", plugin_text, error);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_ATTACH, "Couldn't attach to plugin: error '%d'", error);goto jsondone;}/* Prepare JSON reply */json_t *reply = janus_create_message("success", session_id, transaction_text);json_t *data = json_object();json_object_set_new(data, "id", json_integer(handle_id));json_object_set_new(reply, "data", data);/* Send the success reply */ret = janus_process_success(request, reply);} else if(!strcasecmp(message_text, "destroy")) {if(handle != NULL) {/* Query is a session-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}janus_mutex_lock(&sessions_mutex);g_hash_table_remove(sessions, &session->session_id);janus_mutex_unlock(&sessions_mutex);/* Notify the source that the session has been destroyed */if(session->source && session->source->transport) {session->source->transport->session_over(session->source->instance, session->session_id, FALSE, FALSE);}/* Schedule the session for deletion */janus_session_destroy(session);/* Prepare JSON reply */json_t *reply = janus_create_message("success", session_id, transaction_text);/* Send the success reply */ret = janus_process_success(request, reply);/* Notify event handlers as well */if(janus_events_is_enabled())janus_events_notify_handlers(JANUS_EVENT_TYPE_SESSION, session_id, "destroyed", NULL);} else if(!strcasecmp(message_text, "detach")) {if(handle == NULL) {/* Query is an handle-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}if(handle->app == NULL || handle->app_handle == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_DETACH, "No plugin to detach from");goto jsondone;}int error = janus_session_handles_remove(session, handle);if(error != 0) {/* TODO Make error struct to pass verbose information */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_DETACH, "Couldn't detach from plugin: error '%d'", error);/* TODO Delete handle instance */goto jsondone;}/* Prepare JSON reply */json_t *reply = janus_create_message("success", session_id, transaction_text);/* Send the success reply */ret = janus_process_success(request, reply);} else if(!strcasecmp(message_text, "hangup")) {if(handle == NULL) {/* Query is an handle-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}if(handle->app == NULL || handle->app_handle == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_DETACH, "No plugin attached");goto jsondone;}janus_ice_webrtc_hangup(handle, "Janus API");/* Prepare JSON reply */json_t *reply = janus_create_message("success", session_id, transaction_text);/* Send the success reply */ret = janus_process_success(request, reply);} else if(!strcasecmp(message_text, "claim")) {janus_mutex_lock(&session->mutex);if(session->source != NULL) {/* Notify the old transport that this session is over for them, but has been reclaimed */session->source->transport->session_over(session->source->instance, session->session_id, FALSE, TRUE);janus_request_destroy(session->source);session->source = NULL;}session->source = janus_request_new(request->transport, request->instance, NULL, FALSE, NULL);/* Notify the new transport that it has claimed a session */session->source->transport->session_claimed(session->source->instance, session->session_id);/* Previous transport may be gone, clear flag. */g_atomic_int_set(&session->transport_gone, 0);janus_mutex_unlock(&session->mutex);/* Prepare JSON reply */json_t *reply = json_object();json_object_set_new(reply, "janus", json_string("success"));json_object_set_new(reply, "session_id", json_integer(session_id));json_object_set_new(reply, "transaction", json_string(transaction_text));/* Send the success reply */ret = janus_process_success(request, reply);} else if(!strcasecmp(message_text, "message")) {if(handle == NULL) {/* Query is an handle-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}if(handle->app == NULL || handle->app_handle == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE, "No plugin to handle this message");goto jsondone;}janus_plugin *plugin_t = (janus_plugin *)handle->app;JANUS_LOG(LOG_VERB, "[%"SCNu64"] There's a message for %s\n", handle->handle_id, plugin_t->get_name());JANUS_VALIDATE_JSON_OBJECT(root, body_parameters,error_code, error_cause, FALSE,JANUS_ERROR_MISSING_MANDATORY_ELEMENT, JANUS_ERROR_INVALID_ELEMENT_TYPE);if(error_code != 0) {ret = janus_process_error_string(request, session_id, transaction_text, error_code, error_cause);goto jsondone;}json_t *body = json_object_get(root, "body");/* Is there an SDP attached? */json_t *jsep = json_object_get(root, "jsep");char *jsep_type = NULL;char *jsep_sdp = NULL, *jsep_sdp_stripped = NULL;gboolean renegotiation = FALSE;if(jsep != NULL) {if(!json_is_object(jsep)) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_JSON_OBJECT, "Invalid jsep object");goto jsondone;}JANUS_VALIDATE_JSON_OBJECT_FORMAT("JSEP error: missing mandatory element (%s)","JSEP error: invalid element type (%s should be %s)",jsep, jsep_parameters, error_code, error_cause, FALSE,JANUS_ERROR_MISSING_MANDATORY_ELEMENT, JANUS_ERROR_INVALID_ELEMENT_TYPE);if(error_code != 0) {ret = janus_process_error_string(request, session_id, transaction_text, error_code, error_cause);goto jsondone;}json_t *type = json_object_get(jsep, "type");jsep_type = g_strdup(json_string_value(type));type = NULL;gboolean do_trickle = TRUE;json_t *jsep_trickle = json_object_get(jsep, "trickle");do_trickle = jsep_trickle ? json_is_true(jsep_trickle) : TRUE;/* Are we still cleaning up from a previous media session? */if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_CLEANING)) {JANUS_LOG(LOG_VERB, "[%"SCNu64"] Still cleaning up from a previous media session, let's wait a bit...\n", handle->handle_id);gint64 waited = 0;while(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_CLEANING)) {g_usleep(100000);waited += 100000;if(waited >= 3*G_USEC_PER_SEC) {JANUS_LOG(LOG_VERB, "[%"SCNu64"]   -- Waited 3 seconds, that's enough!\n", handle->handle_id);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_WEBRTC_STATE, "Still cleaning a previous session");goto jsondone;}}}/* Check if we're renegotiating (if we have an answer, we did an offer/answer round already) */renegotiation = janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_NEGOTIATED);/* Check the JSEP type */janus_mutex_lock(&handle->mutex);int offer = 0;if(!strcasecmp(jsep_type, "offer")) {offer = 1;janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_OFFER);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_ANSWER);} else if(!strcasecmp(jsep_type, "answer")) {janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_ANSWER);if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_OFFER))janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_NEGOTIATED);offer = 0;} else {/* TODO Handle other message types as well */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_JSEP_UNKNOWN_TYPE, "JSEP error: unknown message type '%s'", jsep_type);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);janus_mutex_unlock(&handle->mutex);goto jsondone;}json_t *sdp = json_object_get(jsep, "sdp");jsep_sdp = (char *)json_string_value(sdp);JANUS_LOG(LOG_VERB, "[%"SCNu64"] Remote SDP:\n%s", handle->handle_id, jsep_sdp);/* Is this valid SDP? */char error_str[512];int audio = 0, video = 0, data = 0;janus_sdp *parsed_sdp = janus_sdp_preparse(handle, jsep_sdp, error_str, sizeof(error_str), &audio, &video, &data);if(parsed_sdp == NULL) {/* Invalid SDP */ret = janus_process_error_string(request, session_id, transaction_text, JANUS_ERROR_JSEP_INVALID_SDP, error_str);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);janus_mutex_unlock(&handle->mutex);goto jsondone;}/* Notify event handlers */if(janus_events_is_enabled()) {janus_events_notify_handlers(JANUS_EVENT_TYPE_JSEP,session_id, handle_id, handle->opaque_id, "remote", jsep_type, jsep_sdp);}/* FIXME We're only handling single audio/video lines for now... */JANUS_LOG(LOG_VERB, "[%"SCNu64"] Audio %s been negotiated, Video %s been negotiated, SCTP/DataChannels %s been negotiated\n",handle->handle_id,audio ? "has" : "has NOT",video ? "has" : "has NOT",data ? "have" : "have NOT");if(audio > 1) {JANUS_LOG(LOG_WARN, "[%"SCNu64"] More than one audio line? only going to negotiate one...\n", handle->handle_id);}if(video > 1) {JANUS_LOG(LOG_WARN, "[%"SCNu64"] More than one video line? only going to negotiate one...\n", handle->handle_id);}if(data > 1) {JANUS_LOG(LOG_WARN, "[%"SCNu64"] More than one data line? only going to negotiate one...\n", handle->handle_id);}
#ifndef HAVE_SCTPif(data) {JANUS_LOG(LOG_WARN, "[%"SCNu64"]   -- DataChannels have been negotiated, but support for them has not been compiled...\n", handle->handle_id);}
#endif/* We behave differently if it's a new session or an update... */if(!renegotiation) {/* New session */if(offer) {/* Setup ICE locally (we received an offer) */if(janus_ice_setup_local(handle, offer, audio, video, data, do_trickle) < 0) {JANUS_LOG(LOG_ERR, "Error setting ICE locally\n");janus_sdp_destroy(parsed_sdp);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNKNOWN, "Error setting ICE locally");janus_mutex_unlock(&handle->mutex);goto jsondone;}} else {/* Make sure we're waiting for an ANSWER in the first place */if(!handle->agent) {JANUS_LOG(LOG_ERR, "Unexpected ANSWER (did we offer?)\n");janus_sdp_destroy(parsed_sdp);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNEXPECTED_ANSWER, "Unexpected ANSWER (did we offer?)");janus_mutex_unlock(&handle->mutex);goto jsondone;}}if(janus_sdp_process(handle, parsed_sdp, FALSE) < 0) {JANUS_LOG(LOG_ERR, "Error processing SDP\n");janus_sdp_destroy(parsed_sdp);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_JSEP_INVALID_SDP, "Error processing SDP");janus_mutex_unlock(&handle->mutex);goto jsondone;}if(!offer) {/* Set remote candidates now (we received an answer) */if(do_trickle) {janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_TRICKLE);} else {janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_TRICKLE);}janus_request_ice_handle_answer(handle, audio, video, data, jsep_sdp);} else {/* Check if the mid RTP extension is being negotiated */handle->stream->mid_ext_id = janus_rtp_header_extension_get_id(jsep_sdp, JANUS_RTP_EXTMAP_MID);/* Check if the RTP Stream ID extension is being negotiated */handle->stream->rid_ext_id = janus_rtp_header_extension_get_id(jsep_sdp, JANUS_RTP_EXTMAP_RID);handle->stream->ridrtx_ext_id = janus_rtp_header_extension_get_id(jsep_sdp, JANUS_RTP_EXTMAP_REPAIRED_RID);/* Check if the frame marking ID extension is being negotiated */handle->stream->framemarking_ext_id = janus_rtp_header_extension_get_id(jsep_sdp, JANUS_RTP_EXTMAP_FRAME_MARKING);/* Check if transport wide CC is supported */int transport_wide_cc_ext_id = janus_rtp_header_extension_get_id(jsep_sdp, JANUS_RTP_EXTMAP_TRANSPORT_WIDE_CC);handle->stream->do_transport_wide_cc = transport_wide_cc_ext_id > 0 ? TRUE : FALSE;handle->stream->transport_wide_cc_ext_id = transport_wide_cc_ext_id;}} else {/* FIXME This is a renegotiation: we can currently only handle simple changes in media* direction and ICE restarts: anything more complex than that will result in an error */JANUS_LOG(LOG_INFO, "[%"SCNu64"] Negotiation update, checking what changed...\n", handle->handle_id);if(janus_sdp_process(handle, parsed_sdp, TRUE) < 0) {JANUS_LOG(LOG_ERR, "Error processing SDP\n");janus_sdp_destroy(parsed_sdp);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNEXPECTED_ANSWER, "Error processing SDP");janus_mutex_unlock(&handle->mutex);goto jsondone;}if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_ICE_RESTART)) {JANUS_LOG(LOG_INFO, "[%"SCNu64"] Restarting ICE...\n", handle->handle_id);/* Update remote credentials for ICE */if(handle->stream) {nice_agent_set_remote_credentials(handle->agent, handle->stream->stream_id,handle->stream->ruser, handle->stream->rpass);}/* FIXME We only need to do that for offers: if it's an answer, we did that already */if(offer) {janus_ice_restart(handle);} else {janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_ICE_RESTART);}/* If we're full-trickling, we'll need to resend the candidates later */if(janus_ice_is_full_trickle_enabled()) {janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_RESEND_TRICKLES);}}
#ifdef HAVE_SCTPif(!offer) {/* Were datachannels just added? */if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_DATA_CHANNELS)) {janus_ice_stream *stream = handle->stream;if(stream != NULL && stream->component != NULL&& stream->component->dtls != NULL && stream->component->dtls->sctp == NULL) {/* Create SCTP association as well */JANUS_LOG(LOG_WARN, "[%"SCNu64"] Creating datachannels...\n", handle->handle_id);janus_dtls_srtp_create_sctp(stream->component->dtls);}}}
#endif}char *tmp = handle->remote_sdp;handle->remote_sdp = g_strdup(jsep_sdp);g_free(tmp);janus_mutex_unlock(&handle->mutex);/* Anonymize SDP */if(janus_sdp_anonymize(parsed_sdp) < 0) {/* Invalid SDP */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_JSEP_INVALID_SDP, "JSEP error: invalid SDP");janus_sdp_destroy(parsed_sdp);g_free(jsep_type);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);goto jsondone;}jsep_sdp_stripped = janus_sdp_write(parsed_sdp);janus_sdp_destroy(parsed_sdp);sdp = NULL;janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);}/* Make sure the app handle is still valid */if(handle->app == NULL || !janus_plugin_session_is_alive(handle->app_handle)) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE, "No plugin to handle this message");g_free(jsep_type);g_free(jsep_sdp_stripped);janus_flags_clear(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER);goto jsondone;}/* Send the message to the plugin (which must eventually free transaction_text and unref the two objects, body and jsep) */json_incref(body);json_t *body_jsep = NULL;if(jsep_sdp_stripped) {body_jsep = json_pack("{ssss}", "type", jsep_type, "sdp", jsep_sdp_stripped);/* Check if simulcasting is enabled */if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_HAS_VIDEO)) {if(handle->stream && (handle->stream->rid[0] || handle->stream->video_ssrc_peer[1])) {json_t *simulcast = json_object();/* If we have rids, pass those, otherwise pass the SSRCs */if(handle->stream->rid[0]) {json_t *rids = json_array();json_array_append_new(rids, json_string(handle->stream->rid[0]));if(handle->stream->rid[1])json_array_append_new(rids, json_string(handle->stream->rid[1]));if(handle->stream->rid[2])json_array_append_new(rids, json_string(handle->stream->rid[2]));json_object_set_new(simulcast, "rids", rids);json_object_set_new(simulcast, "rid-ext", json_integer(handle->stream->rid_ext_id));} else {json_t *ssrcs = json_array();json_array_append_new(ssrcs, json_integer(handle->stream->video_ssrc_peer[0]));if(handle->stream->video_ssrc_peer[1])json_array_append_new(ssrcs, json_integer(handle->stream->video_ssrc_peer[1]));if(handle->stream->video_ssrc_peer[2])json_array_append_new(ssrcs, json_integer(handle->stream->video_ssrc_peer[2]));json_object_set_new(simulcast, "ssrcs", ssrcs);}if(handle->stream->framemarking_ext_id > 0)json_object_set_new(simulcast, "framemarking-ext", json_integer(handle->stream->framemarking_ext_id));json_object_set_new(body_jsep, "simulcast", simulcast);}}/* Check if this is a renegotiation or update */if(renegotiation)json_object_set_new(body_jsep, "update", json_true());}janus_plugin_result *result = plugin_t->handle_message(handle->app_handle,g_strdup((char *)transaction_text), body, body_jsep);g_free(jsep_type);g_free(jsep_sdp_stripped);if(result == NULL) {/* Something went horribly wrong! */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE, "Plugin didn't give a result");goto jsondone;}if(result->type == JANUS_PLUGIN_OK) {/* The plugin gave a result already (synchronous request/response) */if(result->content == NULL || !json_is_object(result->content)) {/* Missing content, or not a JSON object */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE,result->content == NULL ?"Plugin didn't provide any content for this synchronous response" :"Plugin returned an invalid JSON response");janus_plugin_result_destroy(result);goto jsondone;}/* Reference the content, as destroying the result instance will decref it */json_incref(result->content);/* Prepare JSON response */json_t *reply = janus_create_message("success", session->session_id, transaction_text);json_object_set_new(reply, "sender", json_integer(handle->handle_id));if(janus_is_opaqueid_in_api_enabled() && handle->opaque_id != NULL)json_object_set_new(reply, "opaque_id", json_string(handle->opaque_id));json_t *plugin_data = json_object();json_object_set_new(plugin_data, "plugin", json_string(plugin_t->get_package()));json_object_set_new(plugin_data, "data", result->content);json_object_set_new(reply, "plugindata", plugin_data);/* Send the success reply */ret = janus_process_success(request, reply);} else if(result->type == JANUS_PLUGIN_OK_WAIT) {/* The plugin received the request but didn't process it yet, send an ack (asynchronous notifications may follow) */json_t *reply = janus_create_message("ack", session_id, transaction_text);if(result->text)json_object_set_new(reply, "hint", json_string(result->text));/* Send the success reply */ret = janus_process_success(request, reply);} else {/* Something went horribly wrong! */ret = janus_process_error_string(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE,(char *)(result->text ? result->text : "Plugin returned a severe (unknown) error"));janus_plugin_result_destroy(result);goto jsondone;}janus_plugin_result_destroy(result);} else if(!strcasecmp(message_text, "trickle")) {if(handle == NULL) {/* Trickle is an handle-level command */ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_REQUEST_PATH, "Unhandled request '%s' at this path", message_text);goto jsondone;}if(handle->app == NULL || !janus_plugin_session_is_alive(handle->app_handle)) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_PLUGIN_MESSAGE, "No plugin to handle this trickle candidate");goto jsondone;}json_t *candidate = json_object_get(root, "candidate");json_t *candidates = json_object_get(root, "candidates");if(candidate == NULL && candidates == NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_MISSING_MANDATORY_ELEMENT, "Missing mandatory element (candidate|candidates)");goto jsondone;}if(candidate != NULL && candidates != NULL) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_JSON, "Can't have both candidate and candidates");goto jsondone;}if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_CLEANING)) {JANUS_LOG(LOG_ERR, "[%"SCNu64"] Received a trickle, but still cleaning a previous session\n", handle->handle_id);ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_WEBRTC_STATE, "Still cleaning a previous session");goto jsondone;}janus_mutex_lock(&handle->mutex);if(!janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_TRICKLE)) {/* It looks like this peer supports Trickle, after all */JANUS_LOG(LOG_VERB, "Handle %"SCNu64" supports trickle even if it didn't negotiate it...\n", handle->handle_id);janus_flags_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_TRICKLE);}/* Is there any stream ready? this trickle may get here before the SDP it relates to */if(handle->stream == NULL) {JANUS_LOG(LOG_WARN, "[%"SCNu64"] No stream, queueing this trickle as it got here before the SDP...\n", handle->handle_id);/* Enqueue this trickle candidate(s), we'll process this later */janus_ice_trickle *early_trickle = janus_ice_trickle_new(transaction_text, candidate ? candidate : candidates);handle->pending_trickles = g_list_append(handle->pending_trickles, early_trickle);/* Send the ack right away, an event will tell the application if the candidate(s) failed */goto trickledone;}/* Is the ICE stack ready already? */if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER) ||!janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_OFFER) ||!janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_ANSWER)) {const char *cause = NULL;if(janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_PROCESSING_OFFER))cause = "processing the offer";else if(!janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_ANSWER))cause = "waiting for the answer";else if(!janus_flags_is_set(&handle->webrtc_flags, JANUS_ICE_HANDLE_WEBRTC_GOT_OFFER))cause = "waiting for the offer";JANUS_LOG(LOG_VERB, "[%"SCNu64"] Still %s, queueing this trickle to wait until we're done there...\n",handle->handle_id, cause);/* Enqueue this trickle candidate(s), we'll process this later */janus_ice_trickle *early_trickle = janus_ice_trickle_new(transaction_text, candidate ? candidate : candidates);handle->pending_trickles = g_list_append(handle->pending_trickles, early_trickle);/* Send the ack right away, an event will tell the application if the candidate(s) failed */goto trickledone;}if(candidate != NULL) {/* We got a single candidate */int error = 0;const char *error_string = NULL;if((error = janus_ice_trickle_parse(handle, candidate, &error_string)) != 0) {ret = janus_process_error(request, session_id, transaction_text, error, "%s", error_string);janus_mutex_unlock(&handle->mutex);goto jsondone;}} else {/* We got multiple candidates in an array */if(!json_is_array(candidates)) {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_INVALID_ELEMENT_TYPE, "candidates is not an array");janus_mutex_unlock(&handle->mutex);goto jsondone;}JANUS_LOG(LOG_VERB, "Got multiple candidates (%zu)\n", json_array_size(candidates));if(json_array_size(candidates) > 0) {/* Handle remote candidates */size_t i = 0;for(i=0; i<json_array_size(candidates); i++) {json_t *c = json_array_get(candidates, i);/* FIXME We don't care if any trickle fails to parse */janus_ice_trickle_parse(handle, c, NULL);}}}trickledone:janus_mutex_unlock(&handle->mutex);/* We reply right away, not to block the web server... */json_t *reply = janus_create_message("ack", session_id, transaction_text);/* Send the success reply */ret = janus_process_success(request, reply);} else {ret = janus_process_error(request, session_id, transaction_text, JANUS_ERROR_UNKNOWN_REQUEST, "Unknown request '%s'", message_text);}jsondone:/* Done processing */if(handle != NULL)janus_refcount_decrease(&handle->ref);if(session != NULL)janus_refcount_decrease(&session->ref);return ret;
}