1 #include <gst/gst.h>
2 #include <gst/sdp/sdp.h>
3 #include <gst/webrtc/webrtc.h>
4
5 #include <string.h>
6
7 static GMainLoop *loop;
8 static GstElement *pipe1, *webrtc1, *webrtc2, *extra_src;
9 static GstBus *bus1;
10
11 #define SEND_SRC(pattern) "videotestsrc is-live=true pattern=" pattern " ! timeoverlay ! queue ! vp8enc ! rtpvp8pay ! queue ! " \
12 "capsfilter caps=application/x-rtp,media=video,payload=96,encoding-name=VP8"
13
14 static void
_element_message(GstElement * parent,GstMessage * msg)15 _element_message (GstElement * parent, GstMessage * msg)
16 {
17 switch (GST_MESSAGE_TYPE (msg)) {
18 case GST_MESSAGE_EOS:{
19 GstElement *receive, *webrtc;
20 GstPad *pad, *peer;
21
22 g_print ("Got element EOS message from %s parent %s\n",
23 GST_OBJECT_NAME (msg->src), GST_OBJECT_NAME (parent));
24
25 receive = GST_ELEMENT (msg->src);
26
27 pad = gst_element_get_static_pad (receive, "sink");
28 peer = gst_pad_get_peer (pad);
29
30 webrtc = GST_ELEMENT (gst_pad_get_parent (peer));
31 gst_bin_remove (GST_BIN (pipe1), receive);
32
33 gst_pad_unlink (peer, pad);
34 gst_element_release_request_pad (webrtc, peer);
35
36 gst_object_unref (pad);
37 gst_object_unref (peer);
38
39 gst_element_set_state (receive, GST_STATE_NULL);
40 break;
41 }
42 default:
43 break;
44 }
45 }
46
47 static gboolean
_bus_watch(GstBus * bus,GstMessage * msg,GstElement * pipe)48 _bus_watch (GstBus * bus, GstMessage * msg, GstElement * pipe)
49 {
50 switch (GST_MESSAGE_TYPE (msg)) {
51 case GST_MESSAGE_STATE_CHANGED:
52 if (GST_ELEMENT (msg->src) == pipe) {
53 GstState old, new, pending;
54
55 gst_message_parse_state_changed (msg, &old, &new, &pending);
56
57 {
58 gchar *dump_name = g_strconcat ("state_changed-",
59 gst_element_state_get_name (old), "_",
60 gst_element_state_get_name (new), NULL);
61 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (msg->src),
62 GST_DEBUG_GRAPH_SHOW_ALL, dump_name);
63 g_free (dump_name);
64 }
65 }
66 break;
67 case GST_MESSAGE_ERROR:{
68 GError *err = NULL;
69 gchar *dbg_info = NULL;
70
71 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (pipe),
72 GST_DEBUG_GRAPH_SHOW_ALL, "error");
73
74 gst_message_parse_error (msg, &err, &dbg_info);
75 g_printerr ("ERROR from element %s: %s\n",
76 GST_OBJECT_NAME (msg->src), err->message);
77 g_printerr ("Debugging info: %s\n", (dbg_info) ? dbg_info : "none");
78 g_error_free (err);
79 g_free (dbg_info);
80 g_main_loop_quit (loop);
81 break;
82 }
83 case GST_MESSAGE_EOS:{
84 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (pipe),
85 GST_DEBUG_GRAPH_SHOW_ALL, "eos");
86 g_print ("EOS received\n");
87 g_main_loop_quit (loop);
88 break;
89 }
90 case GST_MESSAGE_ELEMENT:{
91 const GstStructure *s = gst_message_get_structure (msg);
92 if (g_strcmp0 (gst_structure_get_name (s), "GstBinForwarded") == 0) {
93 GstMessage *sub_msg;
94
95 gst_structure_get (s, "message", GST_TYPE_MESSAGE, &sub_msg, NULL);
96 _element_message (GST_ELEMENT (msg->src), sub_msg);
97 gst_message_unref (sub_msg);
98 }
99 break;
100 }
101 default:
102 break;
103 }
104
105 return TRUE;
106 }
107
108 static void
_webrtc_pad_added(GstElement * webrtc,GstPad * new_pad,GstElement * pipe)109 _webrtc_pad_added (GstElement * webrtc, GstPad * new_pad, GstElement * pipe)
110 {
111 GstElement *out;
112 GstPad *sink;
113
114 if (GST_PAD_DIRECTION (new_pad) != GST_PAD_SRC)
115 return;
116
117 out = gst_parse_bin_from_description ("queue ! rtpvp8depay ! vp8dec ! "
118 "videoconvert ! queue ! xvimagesink", TRUE, NULL);
119 gst_bin_add (GST_BIN (pipe), out);
120 gst_element_sync_state_with_parent (out);
121
122 sink = out->sinkpads->data;
123
124 gst_pad_link (new_pad, sink);
125 }
126
127 static void
_on_answer_received(GstPromise * promise,gpointer user_data)128 _on_answer_received (GstPromise * promise, gpointer user_data)
129 {
130 GstWebRTCSessionDescription *answer = NULL;
131 const GstStructure *reply;
132 gchar *desc;
133
134 g_assert (gst_promise_wait (promise) == GST_PROMISE_RESULT_REPLIED);
135 reply = gst_promise_get_reply (promise);
136 gst_structure_get (reply, "answer",
137 GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &answer, NULL);
138 gst_promise_unref (promise);
139 desc = gst_sdp_message_as_text (answer->sdp);
140 g_print ("Created answer:\n%s\n", desc);
141 g_free (desc);
142
143 /* this is one way to tell webrtcbin that we don't want to be notified when
144 * this task is complete: set a NULL promise */
145 g_signal_emit_by_name (webrtc1, "set-remote-description", answer, NULL);
146 /* this is another way to tell webrtcbin that we don't want to be notified
147 * when this task is complete: interrupt the promise */
148 promise = gst_promise_new ();
149 g_signal_emit_by_name (webrtc2, "set-local-description", answer, promise);
150 gst_promise_interrupt (promise);
151 gst_promise_unref (promise);
152
153 gst_webrtc_session_description_free (answer);
154 }
155
156 static void
_on_offer_received(GstPromise * promise,gpointer user_data)157 _on_offer_received (GstPromise * promise, gpointer user_data)
158 {
159 GstWebRTCSessionDescription *offer = NULL;
160 const GstStructure *reply;
161 gchar *desc;
162
163 g_assert (gst_promise_wait (promise) == GST_PROMISE_RESULT_REPLIED);
164 reply = gst_promise_get_reply (promise);
165 gst_structure_get (reply, "offer",
166 GST_TYPE_WEBRTC_SESSION_DESCRIPTION, &offer, NULL);
167 gst_promise_unref (promise);
168 desc = gst_sdp_message_as_text (offer->sdp);
169 g_print ("Created offer:\n%s\n", desc);
170 g_free (desc);
171
172 g_signal_emit_by_name (webrtc1, "set-local-description", offer, NULL);
173 g_signal_emit_by_name (webrtc2, "set-remote-description", offer, NULL);
174
175 promise = gst_promise_new_with_change_func (_on_answer_received, user_data,
176 NULL);
177 g_signal_emit_by_name (webrtc2, "create-answer", NULL, promise);
178
179 gst_webrtc_session_description_free (offer);
180 }
181
182 static void
_on_negotiation_needed(GstElement * element,gpointer user_data)183 _on_negotiation_needed (GstElement * element, gpointer user_data)
184 {
185 GstPromise *promise;
186
187 promise = gst_promise_new_with_change_func (_on_offer_received, user_data,
188 NULL);
189 g_signal_emit_by_name (webrtc1, "create-offer", NULL, promise);
190 }
191
192 static void
_on_ice_candidate(GstElement * webrtc,guint mlineindex,gchar * candidate,GstElement * other)193 _on_ice_candidate (GstElement * webrtc, guint mlineindex, gchar * candidate,
194 GstElement * other)
195 {
196 g_signal_emit_by_name (other, "add-ice-candidate", mlineindex, candidate);
197 }
198
199 static gboolean
stream_change(gpointer data)200 stream_change (gpointer data)
201 {
202 if (!extra_src) {
203 g_print ("Adding extra stream\n");
204 extra_src =
205 gst_parse_bin_from_description (SEND_SRC ("circular"), TRUE, NULL);
206
207 gst_element_set_locked_state (extra_src, TRUE);
208 gst_bin_add (GST_BIN (pipe1), extra_src);
209 gst_element_link (extra_src, webrtc1);
210 gst_element_set_locked_state (extra_src, FALSE);
211 gst_element_sync_state_with_parent (extra_src);
212 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (pipe1),
213 GST_DEBUG_GRAPH_SHOW_ALL, "add");
214 } else {
215 GstPad *pad, *peer;
216 GstWebRTCRTPTransceiver *transceiver;
217
218 g_print ("Removing extra stream\n");
219 pad = gst_element_get_static_pad (extra_src, "src");
220 peer = gst_pad_get_peer (pad);
221 gst_element_send_event (extra_src, gst_event_new_eos ());
222
223 g_object_get (peer, "transceiver", &transceiver, NULL);
224 g_object_set (transceiver, "direction",
225 GST_WEBRTC_RTP_TRANSCEIVER_DIRECTION_INACTIVE, NULL);
226
227 gst_element_set_locked_state (extra_src, TRUE);
228 gst_element_set_state (extra_src, GST_STATE_NULL);
229 gst_pad_unlink (pad, peer);
230 gst_element_release_request_pad (webrtc1, peer);
231
232 gst_object_unref (transceiver);
233 gst_object_unref (peer);
234 gst_object_unref (pad);
235
236 gst_bin_remove (GST_BIN (pipe1), extra_src);
237 extra_src = NULL;
238 GST_DEBUG_BIN_TO_DOT_FILE_WITH_TS (GST_BIN (pipe1),
239 GST_DEBUG_GRAPH_SHOW_ALL, "remove");
240 }
241
242 return G_SOURCE_CONTINUE;
243 }
244
245 int
main(int argc,char * argv[])246 main (int argc, char *argv[])
247 {
248 gst_init (&argc, &argv);
249
250 loop = g_main_loop_new (NULL, FALSE);
251 pipe1 = gst_parse_launch (SEND_SRC ("smpte")
252 " ! webrtcbin name=smpte bundle-policy=max-bundle " SEND_SRC ("ball")
253 " ! webrtcbin name=ball bundle-policy=max-bundle", NULL);
254 g_object_set (pipe1, "message-forward", TRUE, NULL);
255 bus1 = gst_pipeline_get_bus (GST_PIPELINE (pipe1));
256 gst_bus_add_watch (bus1, (GstBusFunc) _bus_watch, pipe1);
257
258 webrtc1 = gst_bin_get_by_name (GST_BIN (pipe1), "smpte");
259 g_signal_connect (webrtc1, "on-negotiation-needed",
260 G_CALLBACK (_on_negotiation_needed), NULL);
261 g_signal_connect (webrtc1, "pad-added", G_CALLBACK (_webrtc_pad_added),
262 pipe1);
263 webrtc2 = gst_bin_get_by_name (GST_BIN (pipe1), "ball");
264 g_signal_connect (webrtc2, "pad-added", G_CALLBACK (_webrtc_pad_added),
265 pipe1);
266 g_signal_connect (webrtc1, "on-ice-candidate",
267 G_CALLBACK (_on_ice_candidate), webrtc2);
268 g_signal_connect (webrtc2, "on-ice-candidate",
269 G_CALLBACK (_on_ice_candidate), webrtc1);
270
271 g_print ("Starting pipeline\n");
272 gst_element_set_state (GST_ELEMENT (pipe1), GST_STATE_PLAYING);
273
274 g_timeout_add_seconds (5, stream_change, NULL);
275
276 g_main_loop_run (loop);
277
278 gst_element_set_state (GST_ELEMENT (pipe1), GST_STATE_NULL);
279 g_print ("Pipeline stopped\n");
280
281 gst_object_unref (webrtc1);
282 gst_object_unref (webrtc2);
283 gst_bus_remove_watch (bus1);
284 gst_object_unref (bus1);
285 gst_object_unref (pipe1);
286
287 gst_deinit ();
288
289 return 0;
290 }
291