gst/rtsp/gstrtspsrc.c: Send RTCP messages back to the server over the TCP connection.

Original commit message from CVS:
* gst/rtsp/gstrtspsrc.c: (gst_rtspsrc_handle_src_event),
(gst_rtspsrc_handle_src_query), (gst_rtspsrc_sink_chain),
(gst_rtspsrc_stream_configure_manager),
(gst_rtspsrc_stream_free_udp), (gst_rtspsrc_stream_configure_tcp),
(gst_rtspsrc_stream_configure_mcast),
(gst_rtspsrc_stream_configure_udp),
(gst_rtspsrc_stream_configure_udp_sink),
(gst_rtspsrc_stream_configure_transport):
Send RTCP messages back to the server over the TCP connection.
* gst/rtsp/rtspconnection.c: (rtsp_connection_write),
(rtsp_connection_send), (rtsp_connection_read), (read_body),
(rtsp_connection_receive):
* gst/rtsp/rtspconnection.h:
Factor out and expose lowlevel _write and _read methods.
Implement sending data messages to the server.

4 files changed, 186 insertions, 59 deletions

diff --git a/ChangeLog b/ChangeLog
index d626b2c0..0ba27a1d 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,22 @@
+2007-05-04  Wim Taymans  <wim@fluendo.com>
+
+	* gst/rtsp/gstrtspsrc.c: (gst_rtspsrc_handle_src_event),
+	(gst_rtspsrc_handle_src_query), (gst_rtspsrc_sink_chain),
+	(gst_rtspsrc_stream_configure_manager),
+	(gst_rtspsrc_stream_free_udp), (gst_rtspsrc_stream_configure_tcp),
+	(gst_rtspsrc_stream_configure_mcast),
+	(gst_rtspsrc_stream_configure_udp),
+	(gst_rtspsrc_stream_configure_udp_sink),
+	(gst_rtspsrc_stream_configure_transport):
+	Send RTCP messages back to the server over the TCP connection.
+
+	* gst/rtsp/rtspconnection.c: (rtsp_connection_write),
+	(rtsp_connection_send), (rtsp_connection_read), (read_body),
+	(rtsp_connection_receive):
+	* gst/rtsp/rtspconnection.h:
+	Factor out and expose lowlevel _write and _read methods.
+	Implement sending data messages to the server.
+
 2007-05-03  Wim Taymans  <wim@fluendo.com>
 
 	* gst/multipart/multipartmux.c: (gst_multipart_mux_queue_pads),
diff --git a/gst/rtsp/gstrtspsrc.c b/gst/rtsp/gstrtspsrc.c
index 36179101..e7064b33 100644
--- a/gst/rtsp/gstrtspsrc.c
+++ b/gst/rtsp/gstrtspsrc.c
@@ -120,12 +120,19 @@ static GstStaticPadTemplate rtptemplate = GST_STATIC_PAD_TEMPLATE ("stream%d",
     GST_PAD_SOMETIMES,
     GST_STATIC_CAPS ("application/x-rtp; application/x-rdt"));
 
-/* template used internally */
-static GstStaticPadTemplate anytemplate = GST_STATIC_PAD_TEMPLATE ("internal%d",
+/* templates used internally */
+static GstStaticPadTemplate anysrctemplate =
+GST_STATIC_PAD_TEMPLATE ("internalsrc%d",
     GST_PAD_SRC,
     GST_PAD_SOMETIMES,
     GST_STATIC_CAPS_ANY);
 
+static GstStaticPadTemplate anysinktemplate =
+GST_STATIC_PAD_TEMPLATE ("internalsink%d",
+    GST_PAD_SINK,
+    GST_PAD_SOMETIMES,
+    GST_STATIC_CAPS_ANY);
+
 enum
 {
   /* FILL ME */
@@ -1081,6 +1088,40 @@ gst_rtspsrc_handle_src_query (GstPad * pad, GstQuery * query)
   return res;
 }
 
+/* callback for RTCP messages to be sent to the server when operating in TCP
+ * mode. */
+static GstFlowReturn
+gst_rtspsrc_sink_chain (GstPad * pad, GstBuffer * buffer)
+{
+  GstRTSPSrc *src;
+  GstRTSPStream *stream;
+  GstFlowReturn res = GST_FLOW_OK;
+  guint8 *data;
+  guint size;
+  RTSPResult ret;
+  RTSPMessage message = { 0 };
+
+  stream = (GstRTSPStream *) gst_pad_get_element_private (pad);
+  src = stream->parent;
+
+  data = GST_BUFFER_DATA (buffer);
+  size = GST_BUFFER_SIZE (buffer);
+
+  rtsp_message_init_data (&message, stream->channel[1]);
+
+  rtsp_message_take_body (&message, data, size);
+
+  GST_DEBUG_OBJECT (src, "sending %u bytes RTCP", size);
+  ret = rtsp_connection_send (src->connection, &message, NULL);
+  GST_DEBUG_OBJECT (src, "sent RTCP, %d", ret);
+
+  rtsp_message_steal_body (&message, &data, &size);
+
+  gst_buffer_unref (buffer);
+
+  return res;
+}
+
 static void
 pad_unblocked (GstPad * pad, gboolean blocked, GstRTSPSrc * src)
 {
@@ -1350,14 +1391,12 @@ gst_rtspsrc_stream_configure_tcp (GstRTSPSrc * src, GstRTSPStream * stream,
 
     *outpad = gst_object_ref (stream->channelpad[0]);
   } else {
-    GstPadTemplate *template;
-
     GST_DEBUG_OBJECT (src, "using manager source pad");
 
-    template = gst_static_pad_template_get (&anytemplate);
+    template = gst_static_pad_template_get (&anysrctemplate);
 
     /* allocate pads for sending the channel data into the manager */
-    pad0 = gst_pad_new_from_template (template, "internal0");
+    pad0 = gst_pad_new_from_template (template, "internalsrc0");
     gst_pad_set_event_function (pad0, gst_rtspsrc_handle_src_event);
     gst_pad_set_query_function (pad0, gst_rtspsrc_handle_src_query);
     gst_pad_link (pad0, stream->channelpad[0]);
@@ -1367,12 +1406,31 @@ gst_rtspsrc_stream_configure_tcp (GstRTSPSrc * src, GstRTSPStream * stream,
     if (stream->channelpad[1]) {
       /* if we have a sinkpad for the other channel, create a pad and link to the
        * manager. */
-      pad1 = gst_pad_new_from_template (template, "internal1");
+      pad1 = gst_pad_new_from_template (template, "internalsrc1");
       gst_pad_link (pad1, stream->channelpad[1]);
       stream->channelpad[1] = pad1;
     }
     gst_object_unref (template);
   }
+  /* setup RTCP transport back to the server */
+  if (src->session) {
+    GstPad *pad;
+
+    template = gst_static_pad_template_get (&anysinktemplate);
+
+    stream->rtcppad = gst_pad_new_from_template (template, "internalsink0");
+    gst_pad_set_chain_function (stream->rtcppad, gst_rtspsrc_sink_chain);
+    gst_pad_set_element_private (stream->rtcppad, stream);
+    gst_pad_set_active (stream->rtcppad, TRUE);
+
+    /* get session RTCP pad */
+    name = g_strdup_printf ("send_rtcp_src_%d", stream->id);
+    pad = gst_element_get_request_pad (src->session, name);
+    g_free (name);
+
+    /* and link */
+    gst_pad_link (pad, stream->rtcppad);
+  }
   return TRUE;
 }
 
@@ -1500,6 +1558,10 @@ gst_rtspsrc_stream_configure_udp_sink (GstRTSPSrc * src, GstRTSPStream * stream,
   gint port;
   gchar *destination, *uri, *name;
 
+  /* no session, we're done */
+  if (src->session == NULL)
+    return TRUE;
+
   /* get host and port */
   if (transport->lower_transport == RTSP_LOWER_TRANS_UDP_MCAST)
     port = transport->port.max;
diff --git a/gst/rtsp/rtspconnection.c b/gst/rtsp/rtspconnection.c
index e5ad277b..038a4aca 100644
--- a/gst/rtsp/rtspconnection.c
+++ b/gst/rtsp/rtspconnection.c
@@ -265,12 +265,43 @@ append_auth_header (RTSPConnection * conn, RTSPMessage * message, GString * str)
 }
 
 RTSPResult
+rtsp_connection_write (RTSPConnection * conn, const guint8 * data, guint size,
+    GTimeVal * timeout)
+{
+  guint towrite;
+
+  g_return_val_if_fail (conn != NULL, RTSP_EINVAL);
+  g_return_val_if_fail (data != NULL || size == 0, RTSP_EINVAL);
+
+  towrite = size;
+
+  while (towrite > 0) {
+    gint written;
+
+    written = write (conn->fd, data, towrite);
+    if (written < 0) {
+      if (errno != EAGAIN && errno != EINTR)
+        goto write_error;
+    } else {
+      towrite -= written;
+      data += written;
+    }
+  }
+  return RTSP_OK;
+
+  /* ERRORS */
+write_error:
+  {
+    return RTSP_ESYS;
+  }
+}
+
+RTSPResult
 rtsp_connection_send (RTSPConnection * conn, RTSPMessage * message,
     GTimeVal * timeout)
 {
-  GString *str;
-  gint towrite;
-  gchar *data;
+  GString *str = NULL;
+  RTSPResult res;
 
 #ifdef G_OS_WIN32
   WSADATA w;
@@ -305,57 +336,69 @@ rtsp_connection_send (RTSPConnection * conn, RTSPMessage * message,
       g_string_append_printf (str, "RTSP/1.0 %d %s\r\n",
           message->type_data.response.code, message->type_data.response.reason);
       break;
+    case RTSP_MESSAGE_DATA:
+    {
+      guint8 data_header[4];
+
+      /* prepare data header */
+      data_header[0] = '$';
+      data_header[1] = message->type_data.data.channel;
+      data_header[2] = (message->body_size >> 8) & 0xff;
+      data_header[3] = message->body_size & 0xff;
+
+      /* create string with header and data */
+      str = g_string_append_len (str, (gchar *) data_header, 4);
+      str =
+          g_string_append_len (str, (gchar *) message->body,
+          message->body_size);
+      break;
+    }
     default:
       g_assert_not_reached ();
       break;
   }
 
-  /* append session id if we have one */
-  if (conn->session_id[0] != '\0') {
-    append_header (RTSP_HDR_SESSION, conn->session_id, str);
-  }
-
-  /* append headers */
-  g_hash_table_foreach (message->hdr_fields, (GHFunc) append_header, str);
-
-  /* Append any authentication headers */
-  append_auth_header (conn, message, str);
-
-  /* append Content-Length and body if needed */
-  if (message->body != NULL && message->body_size > 0) {
-    gchar *len;
-
-    len = g_strdup_printf ("%d", message->body_size);
-    append_header (RTSP_HDR_CONTENT_LENGTH, len, str);
-    g_free (len);
-    /* header ends here */
-    g_string_append (str, "\r\n");
-    str =
-        g_string_append_len (str, (gchar *) message->body, message->body_size);
-  } else {
-    /* just end headers */
-    g_string_append (str, "\r\n");
+  /* append specific headers and body */
+  switch (message->type) {
+    case RTSP_MESSAGE_REQUEST:
+    case RTSP_MESSAGE_RESPONSE:
+      /* append session id if we have one */
+      if (conn->session_id[0] != '\0') {
+        append_header (RTSP_HDR_SESSION, conn->session_id, str);
+      }
+      /* append headers */
+      g_hash_table_foreach (message->hdr_fields, (GHFunc) append_header, str);
+
+      /* Append any authentication headers */
+      append_auth_header (conn, message, str);
+
+      /* append Content-Length and body if needed */
+      if (message->body != NULL && message->body_size > 0) {
+        gchar *len;
+
+        len = g_strdup_printf ("%d", message->body_size);
+        append_header (RTSP_HDR_CONTENT_LENGTH, len, str);
+        g_free (len);
+        /* header ends here */
+        g_string_append (str, "\r\n");
+        str =
+            g_string_append_len (str, (gchar *) message->body,
+            message->body_size);
+      } else {
+        /* just end headers */
+        g_string_append (str, "\r\n");
+      }
+      break;
+    default:
+      break;
   }
 
   /* write request */
-  towrite = str->len;
-  data = str->str;
-
-  while (towrite > 0) {
-    gint written;
+  res = rtsp_connection_write (conn, (guint8 *) str->str, str->len, timeout);
 
-    written = write (conn->fd, data, towrite);
-    if (written < 0) {
-      if (errno != EAGAIN && errno != EINTR)
-        goto write_error;
-    } else {
-      towrite -= written;
-      data += written;
-    }
-  }
   g_string_free (str, TRUE);
 
-  return RTSP_OK;
+  return res;
 
 #ifdef G_OS_WIN32
 startup_error:
@@ -371,11 +414,6 @@ version_error:
     return RTSP_EWSAVERSION;
   }
 #endif
-write_error:
-  {
-    g_string_free (str, TRUE);
-    return RTSP_ESYS;
-  }
 }
 
 static RTSPResult
@@ -550,7 +588,7 @@ no_column:
 }
 
 RTSPResult
-rtsp_connection_read (RTSPConnection * conn, gpointer data, guint size,
+rtsp_connection_read (RTSPConnection * conn, guint8 * data, guint size,
     GTimeVal * timeout)
 {
   fd_set readfds;
@@ -631,7 +669,7 @@ rtsp_connection_read (RTSPConnection * conn, gpointer data, guint size,
         goto read_error;
     } else {
       toread -= bytes;
-      data = (char *) data + bytes;
+      data += bytes;
     }
   }
   return RTSP_OK;
@@ -663,7 +701,7 @@ static RTSPResult
 read_body (RTSPConnection * conn, glong content_length, RTSPMessage * msg,
     GTimeVal * timeout)
 {
-  gchar *body;
+  guint8 *body;
   RTSPResult res;
 
   if (content_length <= 0) {
@@ -732,7 +770,8 @@ rtsp_connection_receive (RTSPConnection * conn, RTSPMessage * msg,
       rtsp_message_init_data (msg, (gint) c);
 
       /* next two bytes are the length of the data */
-      RTSP_CHECK (rtsp_connection_read (conn, &size, 2, timeout), read_error);
+      RTSP_CHECK (rtsp_connection_read (conn, (guint8 *) & size, 2, timeout),
+          read_error);
 
       size = GUINT16_FROM_BE (size);
 
diff --git a/gst/rtsp/rtspconnection.h b/gst/rtsp/rtspconnection.h
index 8ddaa4ca..b2fcd072 100644
--- a/gst/rtsp/rtspconnection.h
+++ b/gst/rtsp/rtspconnection.h
@@ -78,10 +78,17 @@ RTSPResult      rtsp_connection_connect  (RTSPConnection *conn, GTimeVal *timeou
 RTSPResult      rtsp_connection_close    (RTSPConnection *conn);
 RTSPResult      rtsp_connection_free     (RTSPConnection *conn);
 
+/* sending/receiving raw bytes */
+RTSPResult      rtsp_connection_read     (RTSPConnection * conn, guint8 * data,
+                                          guint size, GTimeVal * timeout);
+RTSPResult      rtsp_connection_write    (RTSPConnection * conn, const guint8 * data, 
+		                          guint size, GTimeVal * timeout);
+
 /* sending/receiving messages */
 RTSPResult      rtsp_connection_send     (RTSPConnection *conn, RTSPMessage *message, GTimeVal *timeout);
 RTSPResult      rtsp_connection_receive  (RTSPConnection *conn, RTSPMessage *message, GTimeVal *timeout);
 
+/* flushing state */
 RTSPResult      rtsp_connection_flush    (RTSPConnection *conn, gboolean flush);
 
 /* Configure Authentication data */