#define DEFAULT_UDP_RECONNECT TRUE
#define DEFAULT_MULTICAST_IFACE NULL
#define DEFAULT_NTP_SYNC FALSE
+#define DEFAULT_USE_PIPELINE_CLOCK FALSE
enum
{
PROP_UDP_RECONNECT,
PROP_MULTICAST_IFACE,
PROP_NTP_SYNC,
+ PROP_USE_PIPELINE_CLOCK,
PROP_LAST
};
"Synchronize received streams to the NTP clock", DEFAULT_NTP_SYNC,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+ g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
+ g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
+ "Use the pipeline running-time to set the NTP time in the RTCP SR messages",
+ DEFAULT_USE_PIPELINE_CLOCK,
+ G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
/**
* GstRTSPSrc::handle-request:
* @rtspsrc: a #GstRTSPSrc
*
* Handle a server request in @request and prepare @response.
*
+ * This signal is called from the streaming thread, you should therefore not
+ * do any state changes on @rtspsrc because this might deadlock. If you want
+ * to modify the state as a result of this signal, post a
+ * #GST_MESSAGE_REQUEST_STATE message on the bus or signal the main thread
+ * in some other way.
+ *
* Since: 1.2
*/
gst_rtspsrc_signals[SIGNAL_HANDLE_REQUEST] =
src->udp_reconnect = DEFAULT_UDP_RECONNECT;
src->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
src->ntp_sync = DEFAULT_NTP_SYNC;
+ src->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
/* get a list of all extensions */
src->extensions = gst_rtsp_ext_list_get ();
case PROP_NTP_SYNC:
rtspsrc->ntp_sync = g_value_get_boolean (value);
break;
+ case PROP_USE_PIPELINE_CLOCK:
+ rtspsrc->use_pipeline_clock = g_value_get_boolean (value);
+ break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
case PROP_NTP_SYNC:
g_value_set_boolean (value, rtspsrc->ntp_sync);
break;
+ case PROP_USE_PIPELINE_CLOCK:
+ g_value_set_boolean (value, rtspsrc->use_pipeline_clock);
+ break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
* server told us to really use the UDP ports. */
stream->udpsrc[0] = gst_object_ref_sink (udpsrc0);
stream->udpsrc[1] = gst_object_ref_sink (udpsrc1);
+ gst_element_set_locked_state (stream->udpsrc[0], TRUE);
+ gst_element_set_locked_state (stream->udpsrc[1], TRUE);
/* keep track of next available port number when we have a range
* configured */
}
static void
+gst_rtspsrc_set_state (GstRTSPSrc * src, GstState state)
+{
+ GList *walk;
+
+ if (src->manager)
+ gst_element_set_state (GST_ELEMENT_CAST (src->manager), state);
+
+ for (walk = src->streams; walk; walk = g_list_next (walk)) {
+ GstRTSPStream *stream = (GstRTSPStream *) walk->data;
+ gint i;
+
+ for (i = 0; i < 2; i++) {
+ if (stream->udpsrc[i])
+ gst_element_set_state (stream->udpsrc[i], state);
+ }
+ }
+}
+
+static void
gst_rtspsrc_flush (GstRTSPSrc * src, gboolean flush, gboolean playing)
{
GstEvent *event;
- gint cmd, i;
+ gint cmd;
GstState state;
- GList *walk;
if (flush) {
event = gst_event_new_flush_start ();
}
gst_rtspsrc_push_event (src, event);
gst_rtspsrc_loop_send_cmd (src, cmd, CMD_LOOP);
-
- /* to manage jitterbuffer buffer mode */
- if (src->manager)
- gst_element_set_state (GST_ELEMENT_CAST (src->manager), state);
-
- /* make running time start start at 0 again */
- for (walk = src->streams; walk; walk = g_list_next (walk)) {
- GstRTSPStream *stream = (GstRTSPStream *) walk->data;
-
- for (i = 0; i < 2; i++) {
- /* for udp case */
- if (stream->udpsrc[i]) {
- gst_element_set_state (stream->udpsrc[i], state);
- }
- }
- }
+ gst_rtspsrc_set_state (src, state);
}
static GstRTSPResult
/* configure the manager */
if (src->manager == NULL) {
GObjectClass *klass;
- GstState target;
if (!(src->manager = gst_element_factory_make (manager, "manager"))) {
/* fallback */
}
/* we manage this element */
+ gst_element_set_locked_state (src->manager, TRUE);
gst_bin_add (GST_BIN_CAST (src), src->manager);
- GST_OBJECT_LOCK (src);
- target = GST_STATE_TARGET (src);
- GST_OBJECT_UNLOCK (src);
-
- ret = gst_element_set_state (src->manager, target);
+ ret = gst_element_set_state (src->manager, GST_STATE_PAUSED);
if (ret == GST_STATE_CHANGE_FAILURE)
goto start_manager_failure;
g_object_set (src->manager, "ntp-sync", src->ntp_sync, NULL);
}
+ if (g_object_class_find_property (klass, "use-pipeline-clock")) {
+ g_object_set (src->manager, "use-pipeline-clock",
+ src->use_pipeline_clock, NULL);
+ }
+
if (g_object_class_find_property (klass, "drop-on-latency")) {
g_object_set (src->manager, "drop-on-latency", src->drop_on_latency,
NULL);
src->multi_iface, NULL);
/* change state */
+ gst_element_set_locked_state (stream->udpsrc[0], TRUE);
gst_element_set_state (stream->udpsrc[0], GST_STATE_PAUSED);
}
/* creating another UDP source for RTCP */
if (max != -1) {
+ GstCaps *caps;
+
uri = g_strdup_printf ("udp://%s:%d", destination, max);
stream->udpsrc[1] =
gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
if (stream->udpsrc[1] == NULL)
goto no_element;
+ caps = gst_caps_new_empty_simple ("application/x-rtcp");
+ g_object_set (stream->udpsrc[1], "caps", caps, NULL);
+ gst_caps_unref (caps);
+
/* take ownership */
gst_object_ref_sink (stream->udpsrc[1]);
/* we manage the UDP elements now. For unicast, the UDP sources where
* allocated in the stream when we suggested a transport. */
if (stream->udpsrc[0]) {
+ gst_element_set_locked_state (stream->udpsrc[0], TRUE);
gst_bin_add (GST_BIN_CAST (src), stream->udpsrc[0]);
GST_DEBUG_OBJECT (src, "setting up UDP source");
/* RTCP port */
if (stream->udpsrc[1]) {
+ GstCaps *caps;
+
+ gst_element_set_locked_state (stream->udpsrc[1], TRUE);
gst_bin_add (GST_BIN_CAST (src), stream->udpsrc[1]);
+ caps = gst_caps_new_empty_simple ("application/x-rtcp");
+ g_object_set (stream->udpsrc[1], "caps", caps, NULL);
+ gst_caps_unref (caps);
+
if (stream->channelpad[1]) {
GstPad *pad;
GST_DEBUG_OBJECT (src, "set flushing %d", flush);
GST_RTSP_STATE_LOCK (src);
- if (src->conninfo.connection) {
+ if (src->conninfo.connection && src->conninfo.flushing != flush) {
GST_DEBUG_OBJECT (src, "connection flush");
gst_rtsp_connection_flush (src->conninfo.connection, flush);
+ src->conninfo.flushing = flush;
}
for (walk = src->streams; walk; walk = g_list_next (walk)) {
GstRTSPStream *stream = (GstRTSPStream *) walk->data;
- GST_DEBUG_OBJECT (src, "stream %p flush", stream);
- if (stream->conninfo.connection)
+ if (stream->conninfo.connection && stream->conninfo.flushing != flush) {
+ GST_DEBUG_OBJECT (src, "stream %p flush", stream);
gst_rtsp_connection_flush (stream->conninfo.connection, flush);
+ stream->conninfo.flushing = flush;
+ }
}
GST_RTSP_STATE_UNLOCK (src);
}
}
static GstFlowReturn
-gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
+gst_rtspsrc_handle_data (GstRTSPSrc * src, GstRTSPMessage * message)
{
- GstRTSPMessage message = { 0 };
- GstRTSPResult res;
+ GstFlowReturn ret = GST_FLOW_OK;
gint channel;
GstRTSPStream *stream;
GstPad *outpad = NULL;
guint8 *data;
guint size;
- GstFlowReturn ret = GST_FLOW_OK;
GstBuffer *buf;
- gboolean is_rtcp, have_data;
+ gboolean is_rtcp;
GstEvent *event;
- /* here we are only interested in data messages */
- have_data = FALSE;
- do {
- GTimeVal tv_timeout;
-
- /* get the next timeout interval */
- gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
-
- /* see if the timeout period expired */
- if ((tv_timeout.tv_sec | tv_timeout.tv_usec) == 0) {
- GST_DEBUG_OBJECT (src, "timout, sending keep-alive");
- /* send keep-alive, only act on interrupt, a warning will be posted for
- * other errors. */
- if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
- goto interrupt;
- /* get new timeout */
- gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
- }
-
- GST_DEBUG_OBJECT (src, "doing receive with timeout %ld seconds, %ld usec",
- tv_timeout.tv_sec, tv_timeout.tv_usec);
-
- /* protect the connection with the connection lock so that we can see when
- * we are finished doing server communication */
- res =
- gst_rtspsrc_connection_receive (src, src->conninfo.connection,
- &message, src->ptcp_timeout);
-
- switch (res) {
- case GST_RTSP_OK:
- GST_DEBUG_OBJECT (src, "we received a server message");
- break;
- case GST_RTSP_EINTR:
- /* we got interrupted this means we need to stop */
- goto interrupt;
- case GST_RTSP_ETIMEOUT:
- /* no reply, send keep alive */
- GST_DEBUG_OBJECT (src, "timeout, sending keep-alive");
- if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
- goto interrupt;
- continue;
- case GST_RTSP_EEOF:
- /* go EOS when the server closed the connection */
- goto server_eof;
- default:
- goto receive_error;
- }
-
- switch (message.type) {
- case GST_RTSP_MESSAGE_REQUEST:
- /* server sends us a request message, handle it */
- res =
- gst_rtspsrc_handle_request (src, src->conninfo.connection,
- &message);
- if (res == GST_RTSP_EEOF)
- goto server_eof;
- else if (res < 0)
- goto handle_request_failed;
- break;
- case GST_RTSP_MESSAGE_RESPONSE:
- /* we ignore response messages */
- GST_DEBUG_OBJECT (src, "ignoring response message");
- if (src->debug)
- gst_rtsp_message_dump (&message);
- break;
- case GST_RTSP_MESSAGE_DATA:
- GST_DEBUG_OBJECT (src, "got data message");
- have_data = TRUE;
- break;
- default:
- GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
- message.type);
- break;
- }
- }
- while (!have_data);
-
- channel = message.type_data.data.channel;
+ channel = message->type_data.data.channel;
stream = find_stream (src, &channel, (gpointer) find_stream_by_channel);
if (!stream)
}
/* take a look at the body to figure out what we have */
- gst_rtsp_message_get_body (&message, &data, &size);
+ gst_rtsp_message_get_body (message, &data, &size);
if (size < 2)
goto invalid_length;
goto unknown_stream;
/* take the message body for further processing */
- gst_rtsp_message_steal_body (&message, &data, &size);
+ gst_rtsp_message_steal_body (message, &data, &size);
/* strip the trailing \0 */
size -= 1;
gst_memory_new_wrapped (0, data, size, 0, size, data, g_free));
/* don't need message anymore */
- gst_rtsp_message_unset (&message);
+ gst_rtsp_message_unset (message);
GST_DEBUG_OBJECT (src, "pushing data of size %d on channel %d", size,
channel);
if (src->need_activate) {
+ gchar *stream_id;
+ GstEvent *event;
+ GChecksum *cs;
+ gchar *uri;
+
+ /* generate an SHA256 sum of the URI */
+ cs = g_checksum_new (G_CHECKSUM_SHA256);
+ uri = src->conninfo.location;
+ g_checksum_update (cs, (const guchar *) uri, strlen (uri));
+ stream_id =
+ g_strdup_printf ("%s/%d", g_checksum_get_string (cs), stream->id);
+ g_checksum_free (cs);
+ event = gst_event_new_stream_start (stream_id);
+ g_free (stream_id);
+ gst_rtspsrc_push_event (src, event);
+
gst_rtspsrc_activate_streams (src);
src->need_activate = FALSE;
}
unknown_stream:
{
GST_DEBUG_OBJECT (src, "unknown stream on channel %d, ignored", channel);
- gst_rtsp_message_unset (&message);
+ gst_rtsp_message_unset (message);
+ return GST_FLOW_OK;
+ }
+invalid_length:
+ {
+ GST_ELEMENT_WARNING (src, RESOURCE, READ, (NULL),
+ ("Short message received, ignoring."));
+ gst_rtsp_message_unset (message);
return GST_FLOW_OK;
}
+}
+
+static GstFlowReturn
+gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
+{
+ GstRTSPMessage message = { 0 };
+ GstRTSPResult res;
+ GstFlowReturn ret = GST_FLOW_OK;
+ GTimeVal tv_timeout;
+
+ while (TRUE) {
+ /* get the next timeout interval */
+ gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
+
+ /* see if the timeout period expired */
+ if ((tv_timeout.tv_sec | tv_timeout.tv_usec) == 0) {
+ GST_DEBUG_OBJECT (src, "timout, sending keep-alive");
+ /* send keep-alive, only act on interrupt, a warning will be posted for
+ * other errors. */
+ if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
+ goto interrupt;
+ /* get new timeout */
+ gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
+ }
+
+ GST_DEBUG_OBJECT (src, "doing receive with timeout %ld seconds, %ld usec",
+ tv_timeout.tv_sec, tv_timeout.tv_usec);
+
+ /* protect the connection with the connection lock so that we can see when
+ * we are finished doing server communication */
+ res =
+ gst_rtspsrc_connection_receive (src, src->conninfo.connection,
+ &message, src->ptcp_timeout);
+
+ switch (res) {
+ case GST_RTSP_OK:
+ GST_DEBUG_OBJECT (src, "we received a server message");
+ break;
+ case GST_RTSP_EINTR:
+ /* we got interrupted this means we need to stop */
+ goto interrupt;
+ case GST_RTSP_ETIMEOUT:
+ /* no reply, send keep alive */
+ GST_DEBUG_OBJECT (src, "timeout, sending keep-alive");
+ if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
+ goto interrupt;
+ continue;
+ case GST_RTSP_EEOF:
+ /* go EOS when the server closed the connection */
+ goto server_eof;
+ default:
+ goto receive_error;
+ }
+
+ switch (message.type) {
+ case GST_RTSP_MESSAGE_REQUEST:
+ /* server sends us a request message, handle it */
+ res =
+ gst_rtspsrc_handle_request (src, src->conninfo.connection,
+ &message);
+ if (res == GST_RTSP_EEOF)
+ goto server_eof;
+ else if (res < 0)
+ goto handle_request_failed;
+ break;
+ case GST_RTSP_MESSAGE_RESPONSE:
+ /* we ignore response messages */
+ GST_DEBUG_OBJECT (src, "ignoring response message");
+ if (src->debug)
+ gst_rtsp_message_dump (&message);
+ break;
+ case GST_RTSP_MESSAGE_DATA:
+ GST_DEBUG_OBJECT (src, "got data message");
+ ret = gst_rtspsrc_handle_data (src, &message);
+ if (ret != GST_FLOW_OK)
+ goto handle_data_failed;
+ break;
+ default:
+ GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
+ message.type);
+ break;
+ }
+ }
+ g_assert_not_reached ();
+
+ /* ERRORS */
server_eof:
{
GST_DEBUG_OBJECT (src, "we got an eof from the server");
interrupt:
{
gst_rtsp_message_unset (&message);
- GST_DEBUG_OBJECT (src, "got interrupted: stop connection flush");
- gst_rtspsrc_connection_flush (src, FALSE);
+ GST_DEBUG_OBJECT (src, "got interrupted");
return GST_FLOW_FLUSHING;
}
receive_error:
gst_rtsp_message_unset (&message);
return GST_FLOW_ERROR;
}
-invalid_length:
+handle_data_failed:
{
- GST_ELEMENT_WARNING (src, RESOURCE, READ, (NULL),
- ("Short message received, ignoring."));
- gst_rtsp_message_unset (&message);
- return GST_FLOW_OK;
+ GST_DEBUG_OBJECT (src, "could no handle data message");
+ return ret;
}
}
break;
}
}
+ g_assert_not_reached ();
/* we get here when the connection got interrupted */
interrupt:
{
gst_rtsp_message_unset (&message);
- GST_DEBUG_OBJECT (src, "got interrupted: stop connection flush");
- gst_rtspsrc_connection_flush (src, FALSE);
+ GST_DEBUG_OBJECT (src, "got interrupted");
return GST_FLOW_FLUSHING;
}
connect_error:
src->pending_cmd = CMD_WAIT;
GST_OBJECT_UNLOCK (src);
/* cancel previous request */
- GST_DEBUG_OBJECT (src, "cancel previous request");
+ GST_DEBUG_OBJECT (src, "cancel previous request %d", old);
gst_rtspsrc_loop_cancel_cmd (src, old);
GST_OBJECT_LOCK (src);
}
break;
case GST_RTSP_MESSAGE_DATA:
/* get next response */
- GST_DEBUG_OBJECT (src, "ignoring data response message");
+ GST_DEBUG_OBJECT (src, "handle data response message");
+ gst_rtspsrc_handle_data (src, response);
goto next;
default:
GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
gst_object_unref (src->provided_clock);
src->provided_clock = netclock;
+ gst_element_post_message (GST_ELEMENT_CAST (src),
+ gst_message_new_clock_provide (GST_OBJECT_CAST (src),
+ src->provided_clock, TRUE));
+
res = TRUE;
cleanup:
g_strfreev (fields);
GST_DEBUG_OBJECT (src, "TEARDOWN...");
+ gst_rtspsrc_set_state (src, GST_STATE_READY);
+
if (src->state < GST_RTSP_STATE_READY) {
GST_DEBUG_OBJECT (src, "not ready, doing cleanup");
goto close;
* udp sources */
gst_rtspsrc_send_dummy_packets (src);
- /* activate receive elements;
- * only in async case, since receive elements may not have been affected
- * by overall state change (e.g. not around yet),
- * do not mess with state in sync case (e.g. seeking) */
- if (async) {
- /* state change might be happening in the application thread. A
- * specific case is when chaging state to NULL where we will wait
- * for this task to finish (gst_rtspsrc_stop). However this task
- * will try to change the state to PLAYING causing a deadlock. */
-
- /* make sure we are not in the middle of a state change. The
- * state lock is a recursive lock so it's safe to lock twice from
- * the same thread */
- if (GST_STATE_TRYLOCK (src)) {
- gst_element_set_state (GST_ELEMENT_CAST (src), GST_STATE_PLAYING);
- GST_STATE_UNLOCK (src);
- } else {
- res = GST_RTSP_ERROR;
- goto changing_state;
- }
- }
+ /* require new SR packets */
+ if (src->manager)
+ g_signal_emit_by_name (src->manager, "reset-sync", NULL);
+
+ gst_rtspsrc_set_state (src, GST_STATE_PLAYING);
/* construct a control url */
if (src->control)
GST_DEBUG_OBJECT (src, "we were already PLAYING");
goto done;
}
-changing_state:
- {
- GST_DEBUG_OBJECT (src, "failed going to PLAYING, already changing state");
- goto done;
- }
create_request_failed:
{
gchar *str = gst_rtsp_strresult (res);
break;
}
+ /* change element states now */
+ gst_rtspsrc_set_state (src, GST_STATE_PAUSED);
+
no_connection:
src->state = GST_RTSP_STATE_READY;
GST_OBJECT_LOCK (src);
cmd = src->pending_cmd;
- if (cmd == CMD_RECONNECT || cmd == CMD_PLAY || cmd == CMD_LOOP)
+ if (cmd == CMD_RECONNECT || cmd == CMD_PLAY || cmd == CMD_PAUSE
+ || cmd == CMD_LOOP)
src->pending_cmd = CMD_LOOP;
else
src->pending_cmd = CMD_WAIT;
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
/* unblock the tcp tasks and make the loop waiting */
gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_WAIT, CMD_LOOP);
+ /* make sure it is waiting before we send PAUSE or PLAY below */
+ GST_RTSP_STREAM_LOCK (rtspsrc);
+ GST_RTSP_STREAM_UNLOCK (rtspsrc);
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
break;
goto done;
switch (transition) {
+ case GST_STATE_CHANGE_NULL_TO_READY:
+ ret = GST_STATE_CHANGE_SUCCESS;
+ break;
+ case GST_STATE_CHANGE_READY_TO_PAUSED:
+ ret = GST_STATE_CHANGE_NO_PREROLL;
+ break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_PLAY, 0);
+ ret = GST_STATE_CHANGE_SUCCESS;
break;
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
/* send pause request and keep the idle task around */
gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_PAUSE, CMD_LOOP);
ret = GST_STATE_CHANGE_NO_PREROLL;
break;
- case GST_STATE_CHANGE_READY_TO_PAUSED:
- ret = GST_STATE_CHANGE_NO_PREROLL;
- break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_CLOSE, CMD_PAUSE);
+ ret = GST_STATE_CHANGE_SUCCESS;
break;
case GST_STATE_CHANGE_READY_TO_NULL:
gst_rtspsrc_stop (rtspsrc);
+ ret = GST_STATE_CHANGE_SUCCESS;
break;
default:
break;
gst_rtspsrc_uri_get_protocols (GType type)
{
static const gchar *protocols[] =
- { "rtsp", "rtspu", "rtspt", "rtsph", "rtsp-sdp", NULL };
+ { "rtsp", "rtspu", "rtspt", "rtsph", "rtsp-sdp",
+ "rtsps", "rtspsu", "rtspst", "rtspsh", NULL
+ };
return protocols;
}
projects / platform / upstream / gstreamer.git / blobdiff