Gstreamer的webrtcbin插件
1、输入参数
static GOptionEntry entries[] = {
{"peer-id", 0, 0, G_OPTION_ARG_STRING, &peer_id, "String ID of the peer to connect to", "ID"},
{"server", 0, 0, G_OPTION_ARG_STRING, &server_url, "Signalling server to connect to", "URL"},
{"disable-ssl", 0, 0, G_OPTION_ARG_NONE, &disable_ssl, "Disable ssl", NULL},
{"remote-offerer", 0, 0, G_OPTION_ARG_NONE, &remote_is_offerer, "Request that the peer generate the offer and we'll answer", NULL},
{NULL},
};
2、解析参数并连接到信令服务器
int main(int argc, char *argv[])
{
GOptionContext *context;
GError *error = NULL;
sigint_setup();
context = g_option_context_new("- gstreamer webrtc sendrecv demo");
g_option_context_add_main_entries(context, entries, NULL);
g_option_context_add_group(context, gst_init_get_option_group());
if (!g_option_context_parse(context, &argc, &argv, &error)){
g_printerr("Error initializing: %s\n", error->message);
return -1;
}
if (!check_plugins())
return -1;
if (!peer_id){
g_printerr("--peer-id is a required argument\n");
return -1;
}
//运行 localhost 服务器时禁用 ssl，因为它可能是具有自签名证书的测试服务器
{
GstUri *uri = gst_uri_from_string(server_url);
if (g_strcmp0("localhost", gst_uri_get_host(uri)) == 0 ||
g_strcmp0("127.0.0.1", gst_uri_get_host(uri)) == 0)
disable_ssl = TRUE;
gst_uri_unref(uri);
}
loop = g_main_loop_new(NULL, FALSE);
connect_to_websocket_server_async();
g_main_loop_run(loop);
g_main_loop_unref(loop);
if (pipe1) {
gst_element_set_state(GST_ELEMENT(pipe1), GST_STATE_NULL);
g_print("Pipeline stopped\n");
gst_object_unref(pipe1);
}
return 0;
}
3、连接到
wss://webrtc.nirbheek.in:8443信令服务器
/* 连接到信令服务器。这是其他一切的入口点 */
static const gchar *server_url = "wss://webrtc.nirbheek.in:8443";
static void
connect_to_websocket_server_async(void)
{
SoupLogger *logger;
SoupMessage *message;
SoupSession *session;
const char *https_aliases[] = {"wss", NULL};
session = soup_session_new_with_options(SOUP_SESSION_SSL_STRICT, !disable_ssl,
SOUP_SESSION_SSL_USE_SYSTEM_CA_FILE, TRUE,
//SOUP_SESSION_SSL_CA_FILE, "/etc/ssl/certs/ca-bundle.crt",
SOUP_SESSION_HTTPS_ALIASES, https_aliases, NULL);
logger = soup_logger_new(SOUP_LOGGER_LOG_BODY, -1);
soup_session_add_feature(session, SOUP_SESSION_FEATURE(logger));
g_object_unref(logger);
message = soup_message_new(SOUP_METHOD_GET, server_url);
g_print("Connecting to server...\n");
/* Once connected, we will register */
soup_session_websocket_connect_async(session, message, NULL, NULL, NULL, (GAsyncReadyCallback)on_server_connected, message);
app_state = SERVER_CONNECTING;
}
4、连接成功后监听消息和关闭事件并注册服务器
static void
on_server_connected(SoupSession *session, GAsyncResult *res,
SoupMessage *msg)
{
GError *error = NULL;
ws_conn = soup_session_websocket_connect_finish(session, res, &error);
if (error)
{
cleanup_and_quit_loop(error->message, SERVER_CONNECTION_ERROR);
g_error_free(error);
return;
}
g_assert_nonnull(ws_conn);
app_state = SERVER_CONNECTED;
g_print("Connected to signalling server\n");
g_signal_connect(ws_conn, "closed", G_CALLBACK(on_server_closed), NULL);
g_signal_connect(ws_conn, "message", G_CALLBACK(on_server_message), NULL);
/* Register with the server so it knows about us and can accept commands */
register_with_server();
}
5、向信令服务器注册
static gboolean
register_with_server(void)
{
gchar *hello;
gint32 our_id;
if (soup_websocket_connection_get_state(ws_conn) !=
SOUP_WEBSOCKET_STATE_OPEN)
return FALSE;
our_id = g_random_int_range(10, 10000);
g_print("Registering id %i with server\n", our_id);
app_state = SERVER_REGISTERING;
// 使用随机整数 ID 向服务器注册。回复将通过 on_server_message() 接收
hello = g_strdup_printf("HELLO %i", our_id);
soup_websocket_connection_send_text(ws_conn, hello);
g_free(hello);
return TRUE;
}
6、告诉信令服务器连接对端peer_id
static gboolean
setup_call(void)
{
gchar *msg;
if (soup_websocket_connection_get_state(ws_conn) != SOUP_WEBSOCKET_STATE_OPEN)
return FALSE;
if (!peer_id)
return FALSE;
g_print("Setting up signalling server call with %s\n", peer_id);
app_state = PEER_CONNECTING;
msg = g_strdup_printf("SESSION %s", peer_id);
soup_websocket_connection_send_text(ws_conn, msg);
g_free(msg);
return TRUE;
}
7、处理回复消息
/* One mega message handler for our asynchronous calling mechanism */
static void
on_server_message(SoupWebsocketConnection *conn, SoupWebsocketDataType type,
GBytes *message, gpointer user_data)
{
// 检测数据是否为文本类型
gchar *text;
switch (type) {
case SOUP_WEBSOCKET_DATA_BINARY:
g_printerr("Received unknown binary message, ignoring\n");
return;
case SOUP_WEBSOCKET_DATA_TEXT: {
gsize size;
const gchar *data = g_bytes_get_data(message, &size);
/* Convert to NULL-terminated string */
text = g_strndup(data, size);
break;
}
default:
g_assert_not_reached();
}
// 服务器已接受我们的注册，我们已准备好发送命令
if (g_strcmp0(text, "HELLO") == 0) {
if (app_state != SERVER_REGISTERING) {
cleanup_and_quit_loop("ERROR: Received HELLO when not registering", APP_STATE_ERROR);
goto out;
}
app_state = SERVER_REGISTERED;
g_print("Registered with server\n");
// 要求信令服务器将我们与特定对等点连接起来
if (!setup_call()) {
cleanup_and_quit_loop("ERROR: Failed to setup call", PEER_CALL_ERROR);
goto out;
}
/* Call has been setup by the server, now we can start negotiation */
} else if (g_strcmp0(text, "SESSION_OK") == 0) {
if (app_state != PEER_CONNECTING) {
cleanup_and_quit_loop("ERROR: Received SESSION_OK when not calling", PEER_CONNECTION_ERROR);
goto out;
}
app_state = PEER_CONNECTED;
// 开始谈判（交换 SDP 和 ICE 候选人）
if (!start_pipeline())
cleanup_and_quit_loop("ERROR: failed to start pipeline", PEER_CALL_ERROR);
} else if (g_str_has_prefix(text, "ERROR")) { //处理错误消息
switch (app_state) {
case SERVER_CONNECTING:
app_state = SERVER_CONNECTION_ERROR;
break;
case SERVER_REGISTERING:
app_state = SERVER_REGISTRATION_ERROR;
break;
case PEER_CONNECTING:
app_state = PEER_CONNECTION_ERROR;
break;
case PEER_CONNECTED:
case PEER_CALL_NEGOTIATING:
app_state = PEER_CALL_ERROR;
break;
default:
app_state = APP_STATE_ERROR;
}
cleanup_and_quit_loop(text, 0);
/* Look for JSON messages containing SDP and ICE candidates */
} else {
JsonNode *root;
JsonObject *object, *child;
JsonParser *parser = json_parser_new();
if (!json_parser_load_from_data(parser, text, -1, NULL)){
g_printerr("Unknown message '%s', ignoring", text);
g_object_unref(parser);
goto out;
}
root = json_parser_get_root(parser);
if (!JSON_NODE_HOLDS_OBJECT(root)){
g_printerr("Unknown json message '%s', ignoring", text);
g_object_unref(parser);
goto out;
}
object = json_node_get_object(root);
if (json_object_has_member(object, "sdp")) { //远端的sdp消息
int ret;
GstSDPMessage *sdp;
const gchar *text, *sdptype;
GstWebRTCSessionDescription *answer;
g_assert_cmphex(app_state, ==, PEER_CALL_NEGOTIATING);
child = json_object_get_object_member(object, "sdp");
if (!json_object_has_member(child, "type")) {
cleanup_and_quit_loop("ERROR: received SDP without 'type'", PEER_CALL_ERROR);
goto out;
}
sdptype = json_object_get_string_member(child, "type");
// 在这个例子中，我们默认创建了要约并收到一个答案，但可以注释掉要约创建并等待要约，因此我们在这里处理。
// 请参阅 gst-plugins-bad 中的 tests/examples/webrtcbidirectional.c 以获取另一个示例，了解如何处理来自同行的要约并使用 webrtcbin 回复答案。
text = json_object_get_string_member(child, "sdp");
ret = gst_sdp_message_new(&sdp);
g_assert_cmphex(ret, ==, GST_SDP_OK);
ret = gst_sdp_message_parse_buffer((guint8 *)text, strlen(text), sdp);
g_assert_cmphex(ret, ==, GST_SDP_OK);
if (g_str_equal(sdptype, "answer")){ // answer sdp消息
g_print("Received answer:\n%s\n", text);
answer = gst_webrtc_session_description_new(GST_WEBRTC_SDP_TYPE_ANSWER, sdp);
g_assert_nonnull(answer);
// 设置远端的sdp到pipeline
{
GstPromise *promise = gst_promise_new();
g_signal_emit_by_name(webrtc1, "set-remote-description", answer, promise);
gst_promise_interrupt(promise);
gst_promise_unref(promise);
}
app_state = PEER_CALL_STARTED;
} else { // offer sdp消息
g_print("Received offer:\n%s\n", text);
on_offer_received(sdp);
}
} else if (json_object_has_member(object, "ice")) { // ICE候选人信息
const gchar *candidate;
gint sdpmlineindex;
child = json_object_get_object_member(object, "ice");
candidate = json_object_get_string_member(child, "candidate");
sdpmlineindex = json_object_get_int_member(child, "sdpMLineIndex");
/* Add ice candidate sent by remote peer */
g_signal_emit_by_name(webrtc1, "add-ice-candidate", sdpmlineindex, candidate);
} else {
g_printerr("Ignoring unknown JSON message:\n%s\n", text);
}
g_object_unref(parser);
}
out:
g_free(text);
}
8、创建webrtcbin的Pipeline
#define STUN_SERVER " stun-server=stun://stun.l.google.com:19302 "
#define RTP_CAPS_OPUS "application/x-rtp,media=audio,encoding-name=OPUS,payload="
#define RTP_CAPS_VP8 "application/x-rtp,media=video,encoding-name=VP8,payload="
static gboolean
start_pipeline(void)
{
GstStateChangeReturn ret;
GError *error = NULL;
pipe1 = gst_parse_launch("webrtcbin bundle-policy=max-bundle name=sendrecv " STUN_SERVER
"videotestsrc is-live=true pattern=ball ! videoconvert ! queue ! vp8enc deadline=1 ! rtpvp8pay ! "
"queue ! " RTP_CAPS_VP8 "96 ! sendrecv. " //发送视频
"audiotestsrc is-live=true wave=red-noise ! audioconvert ! audioresample ! queue ! opusenc ! rtpopuspay ! "
"queue ! " RTP_CAPS_OPUS "97 ! sendrecv. ", //发送音频
&error);
if (error) {
g_printerr("Failed to parse launch: %s\n", error->message);
g_error_free(error);
goto err;
}
// 获取webrtcbin插件对象
webrtc1 = gst_bin_get_by_name(GST_BIN(pipe1), "sendrecv");
g_assert_nonnull(webrtc1);
// 这是我们create offer sdp的 gstwebrtc 入口点。当管道进入播放状态时，它将被调用。
g_signal_connect(webrtc1, "on-negotiation-needed", G_CALLBACK(on_negotiation_needed), NULL);
// 我们需要通过 websockets 信令服务器将此 ICE 候选传输到浏览器。来自浏览器的传入 ice 候选也需要我们添加，请参阅 on_server_message()
g_signal_connect(webrtc1, "on-ice-candidate", G_CALLBACK(send_ice_candidate_message), NULL);
g_signal_connect(webrtc1, "notify::ice-gathering-state", G_CALLBACK(on_ice_gathering_state_notify), NULL);
gst_element_set_state(pipe1, GST_STATE_READY);
g_signal_emit_by_name(webrtc1, "create-data-channel", "channel", NULL, &send_channel);
if (send_channel) {
g_print("Created data channel\n");
connect_data_channel_signals(send_channel);
} else {
g_print("Could not create data channel, is usrsctp available?\n");
}
g_signal_connect(webrtc1, "on-data-channel", G_CALLBACK(on_data_channel), NULL);
// 接收到音视频数据on_incoming_stream回调函数处理
g_signal_connect(webrtc1, "pad-added", G_CALLBACK(on_incoming_stream), pipe1);
// 生命周期与管道本身相同, 启动管道
gst_object_unref(webrtc1);
g_print("Starting pipeline\n");
ret = gst_element_set_state(GST_ELEMENT(pipe1), GST_STATE_PLAYING);
if (ret == GST_STATE_CHANGE_FAILURE)
goto err;
return TRUE;
err:
if (pipe1)
g_clear_object(&pipe1);
if (webrtc1)
webrtc1 = NULL;
return FALSE;
}
9、data-channel数据通道数据通信
static void
connect_data_channel_signals(GObject *data_channel)
{
g_signal_connect(data_channel, "on-error", G_CALLBACK(data_channel_on_error), NULL);
g_signal_connect(data_channel, "on-open", G_CALLBACK(data_channel_on_open), NULL);
g_signal_connect(data_channel, "on-close", G_CALLBACK(data_channel_on_close), NULL);
g_signal_connect(data_channel, "on-message-string", G_CALLBACK(data_channel_on_message_string), NULL);
}
10、协商处理
static void
on_negotiation_needed(GstElement *element, gpointer user_data)
{
app_state = PEER_CALL_NEGOTIATING;
// 是否接收远端的sdp，音视频流
if (remote_is_offerer) {
//发送指令获取sdp
gchar *msg = g_strdup_printf("OFFER_REQUEST");
soup_websocket_connection_send_text(ws_conn, msg);
g_free(msg);
} else {
//只发送模式
GArray *transceivers;
g_signal_emit_by_name(element, "get-transceivers", &transceivers);
GstWebRTCRTPTransceiver *transceiver = g_array_index(transceivers, GstWebRTCRTPTransceiver *, 0);
transceiver->direction = GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_SENDONLY;
g_object_set(transceiver, "fec-type", GST_WEBRTC_FEC_TYPE_ULP_RED, NULL);
g_object_set(transceiver, "do-nack", TRUE, NULL);
//创建本地offer sdp，获取本地音视频信息
GstPromise *promise;
promise = gst_promise_new_with_change_func(on_offer_created, user_data, NULL);
g_signal_emit_by_name(webrtc1, "create-offer", NULL, promise);
}
}
11、获取offer sdp后设置本地并且发送给对端
/* Offer created by our pipeline, to be sent to the peer */
static void
on_offer_created(GstPromise *promise, gpointer user_data)
{
GstWebRTCSessionDescription *offer = NULL;
const GstStructure *reply;
g_assert_cmphex(app_state, ==, PEER_CALL_NEGOTIATING);
g_assert_cmphex(gst_promise_wait(promise), ==, GST_PROMISE_RESULT_REPLIED);
reply = gst_promise_get_reply(promise);
gst_structure_get(reply, "offer", GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &offer, NULL);
gst_promise_unref(promise);
promise = gst_promise_new();
g_signal_emit_by_name(webrtc1, "set-local-description", offer, promise);
gst_promise_interrupt(promise);
gst_promise_unref(promise);
/* Send offer to peer */
send_sdp_to_peer(offer);
gst_webrtc_session_description_free(offer);
}
12、接收音视频数据并解码后调用
on_incoming_decodebin_stream
static void
on_incoming_stream(GstElement *webrtc, GstPad *pad, GstElement *pipe)
{
GstElement *decodebin;
GstPad *sinkpad;
if (GST_PAD_DIRECTION(pad) != GST_PAD_SRC)
return;
decodebin = gst_element_factory_make("decodebin", NULL);
g_signal_connect(decodebin, "pad-added", G_CALLBACK(on_incoming_decodebin_stream), pipe);
gst_bin_add(GST_BIN(pipe), decodebin);
gst_element_sync_state_with_parent(decodebin);
sinkpad = gst_element_get_static_pad(decodebin, "sink");
gst_pad_link(pad, sinkpad);
gst_object_unref(sinkpad);
}
13、接收解码后的音视频数据并播放
static void
on_incoming_decodebin_stream(GstElement *decodebin, GstPad *pad,
GstElement *pipe)
{
GstCaps *caps;
const gchar *name;
if (!gst_pad_has_current_caps(pad)) {
g_printerr("Pad '%s' has no caps, can't do anything, ignoring\n", GST_PAD_NAME(pad));
return;
}
caps = gst_pad_get_current_caps(pad);
name = gst_structure_get_name(gst_caps_get_structure(caps, 0));
if (g_str_has_prefix(name, "video")){
handle_media_stream(pad, pipe, "videoconvert", "autovideosink");
} else if (g_str_has_prefix(name, "audio")) {
handle_media_stream(pad, pipe, "audioconvert", "autoaudiosink");
} else {
g_printerr("Unknown pad %s, ignoring", GST_PAD_NAME(pad));
}
}