gst/tcp/gsttcpclientsink.c (gst_tcpclientsink_base_init): Actually add the pad template.

Original commit message from CVS:
2005-09-28  Andy Wingo  <wingo@pobox.com>

* gst/tcp/gsttcpclientsink.c (gst_tcpclientsink_base_init):
Actually add the pad template.
(gst_tcpclientsink_get_type): We're a base sink. Woot, works.

* gst/tcp/gsttcpserversrc.c: Go ahead and fix up serversrc while
I'm at it...

ChangeLog

@@ -1,5 +1,12 @@
 2005-09-28  Andy Wingo  <wingo@pobox.com>
 
+	* gst/tcp/gsttcpclientsink.c (gst_tcpclientsink_base_init):
+	Actually add the pad template.
+	(gst_tcpclientsink_get_type): We're a base sink. Woot, works.
+
+	* gst/tcp/gsttcpserversrc.c: Go ahead and fix up serversrc while
+	I'm at it...
+
 	* gst/tcp/gsttcpclientsrc.c: Make interruptable -- code stolen
 	from fdsrc. Get caps in create() instead of start() so it can be
 	interrupted. Interruption somewhat untested.

gst/tcp/gsttcpclientsink.c

@@ -53,6 +53,11 @@ enum
       /* FILL ME */
 };
 
+static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
 static void gst_tcpclientsink_base_init (gpointer g_class);
 static void gst_tcpclientsink_class_init (GstTCPClientSink * klass);
 static void gst_tcpclientsink_init (GstTCPClientSink * tcpclientsink);
@@ -95,7 +100,7 @@ gst_tcpclientsink_get_type (void)
     };
 
     tcpclientsink_type =
-        g_type_register_static (GST_TYPE_ELEMENT, "GstTCPClientSink",
+        g_type_register_static (GST_TYPE_BASE_SINK, "GstTCPClientSink",
         &tcpclientsink_info, 0);
   }
   return tcpclientsink_type;
@@ -106,6 +111,9 @@ gst_tcpclientsink_base_init (gpointer g_class)
 {
   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 
+  gst_element_class_add_pad_template (element_class,
+      gst_static_pad_template_get (&sinktemplate));
+
   gst_element_class_set_details (element_class, &gst_tcpclientsink_details);
 }
 

gst/tcp/gsttcpclientsrc.c

@@ -34,19 +34,19 @@
 
 /* control stuff stolen from fdsrc */
 #define CONTROL_STOP            'S'     /* stop the select call */
-#define CONTROL_SOCKETS(src)   src->control_fds
-#define WRITE_SOCKET(src)      src->control_fds[1]
-#define READ_SOCKET(src)       src->control_fds[0]
+#define CONTROL_SOCKETS(o)      o->control_fds
+#define WRITE_SOCKET(o)         o->control_fds[1]
+#define READ_SOCKET(o)          o->control_fds[0]
 
-#define SEND_COMMAND(src, command)          \
+#define SEND_COMMAND(o, command)          \
 G_STMT_START {                              \
   unsigned char c; c = command;             \
-  write (WRITE_SOCKET(src), &c, 1);         \
+  write (WRITE_SOCKET(o), &c, 1);         \
 } G_STMT_END
 
-#define READ_COMMAND(src, command, res)        \
+#define READ_COMMAND(o, command, res)        \
 G_STMT_START {                                 \
-  res = read(READ_SOCKET(src), &command, 1);   \
+  res = read(READ_SOCKET(o), &command, 1);   \
 } G_STMT_END
 
 

gst/tcp/gsttcpserversrc.c

@@ -28,6 +28,25 @@
 #include "gsttcpserversrc.h"
 #include <unistd.h>
 #include <sys/ioctl.h>
+#include <fcntl.h>
+
+
+/* control stuff stolen from fdsrc */
+#define CONTROL_STOP            'S'     /* stop the select call */
+#define CONTROL_SOCKETS(o)      o->control_fds
+#define WRITE_SOCKET(o)         o->control_fds[1]
+#define READ_SOCKET(o)          o->control_fds[0]
+
+#define SEND_COMMAND(o, command)          \
+G_STMT_START {                              \
+  unsigned char c; c = command;             \
+  write (WRITE_SOCKET(o), &c, 1);         \
+} G_STMT_END
+
+#define READ_COMMAND(o, command, res)        \
+G_STMT_START {                                 \
+  res = read(READ_SOCKET(o), &command, 1);   \
+} G_STMT_END
 
 
 GST_DEBUG_CATEGORY (tcpserversrc_debug);
@@ -66,6 +85,7 @@ static void gst_tcpserversrc_finalize (GObject * gobject);
 
 static gboolean gst_tcpserversrc_start (GstBaseSrc * bsrc);
 static gboolean gst_tcpserversrc_stop (GstBaseSrc * bsrc);
+static gboolean gst_tcpserversrc_unlock (GstBaseSrc * bsrc);
 static GstFlowReturn gst_tcpserversrc_create (GstPushSrc * psrc,
     GstBuffer ** buf);
 
@@ -113,6 +133,7 @@ gst_tcpserversrc_class_init (GstTCPServerSrcClass * klass)
 
   gstbasesrc_class->start = gst_tcpserversrc_start;
   gstbasesrc_class->stop = gst_tcpserversrc_stop;
+  gstbasesrc_class->unlock = gst_tcpserversrc_unlock;
 
   gstpush_src_class->create = gst_tcpserversrc_create;
 
@@ -130,6 +151,9 @@ gst_tcpserversrc_init (GstTCPServerSrc * src, GstTCPServerSrcClass * g_class)
   src->curoffset = 0;
   src->protocol = GST_TCP_PROTOCOL_NONE;
 
+  READ_SOCKET (src) = -1;
+  WRITE_SOCKET (src) = -1;
+
   GST_FLAG_UNSET (src, GST_TCPSERVERSRC_OPEN);
 }
 
@@ -156,8 +180,8 @@ gst_tcpserversrc_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 
   switch (src->protocol) {
     case GST_TCP_PROTOCOL_NONE:
-      ret = gst_tcp_read_buffer (GST_ELEMENT (src), src->client_sock_fd, -1,
-          outbuf);
+      ret = gst_tcp_read_buffer (GST_ELEMENT (src), src->client_sock_fd,
+          READ_SOCKET (src), outbuf);
       break;
 
     case GST_TCP_PROTOCOL_GDP:
@@ -165,8 +189,8 @@ gst_tcpserversrc_create (GstPushSrc * psrc, GstBuffer ** outbuf)
         GstCaps *caps;
         gchar *string;
 
-        ret = gst_tcp_gdp_read_caps (GST_ELEMENT (src), src->client_sock_fd, -1,
-            &caps);
+        ret = gst_tcp_gdp_read_caps (GST_ELEMENT (src), src->client_sock_fd,
+            READ_SOCKET (src), &caps);
 
         if (ret != GST_FLOW_OK)
           goto gdp_caps_read_error;
@@ -179,8 +203,8 @@ gst_tcpserversrc_create (GstPushSrc * psrc, GstBuffer ** outbuf)
         gst_pad_set_caps (GST_BASE_SRC_PAD (psrc), caps);
       }
 
-      ret = gst_tcp_gdp_read_buffer (GST_ELEMENT (src), src->client_sock_fd, -1,
-          outbuf);
+      ret = gst_tcp_gdp_read_buffer (GST_ELEMENT (src), src->client_sock_fd,
+          READ_SOCKET (src), outbuf);
 
       if (ret == GST_FLOW_OK)
         gst_buffer_set_caps (*outbuf, GST_PAD_CAPS (GST_BASE_SRC_PAD (src)));
@@ -277,6 +301,13 @@ gst_tcpserversrc_start (GstBaseSrc * bsrc)
   int ret;
   GstTCPServerSrc *src = GST_TCPSERVERSRC (bsrc);
 
+  /* create the control sockets before anything */
+  if (socketpair (PF_UNIX, SOCK_STREAM, 0, CONTROL_SOCKETS (src)) < 0)
+    goto socket_pair;
+
+  fcntl (READ_SOCKET (src), F_SETFL, O_NONBLOCK);
+  fcntl (WRITE_SOCKET (src), F_SETFL, O_NONBLOCK);
+
   /* reset caps_received flag */
   src->caps_received = FALSE;
 
@@ -334,6 +365,12 @@ gst_tcpserversrc_start (GstBaseSrc * bsrc)
   return TRUE;
 
   /* ERRORS */
+socket_pair:
+  {
+    GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ_WRITE, (NULL),
+        GST_ERROR_SYSTEM);
+    return FALSE;
+  }
 socket_error:
   {
     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL), GST_ERROR_SYSTEM);
@@ -392,5 +429,21 @@ gst_tcpserversrc_stop (GstBaseSrc * bsrc)
   }
   GST_FLAG_UNSET (src, GST_TCPSERVERSRC_OPEN);
 
+  close (READ_SOCKET (src));
+  close (WRITE_SOCKET (src));
+  READ_SOCKET (src) = -1;
+  WRITE_SOCKET (src) = -1;
+
+  return TRUE;
+}
+
+/* will be called only between calls to start() and stop() */
+static gboolean
+gst_tcpserversrc_unlock (GstBaseSrc * bsrc)
+{
+  GstTCPServerSrc *src = GST_TCPSERVERSRC (bsrc);
+
+  SEND_COMMAND (src, CONTROL_STOP);
+
   return TRUE;
 }

gst/tcp/gsttcpserversrc.h

@@ -72,6 +72,8 @@ struct _GstTCPServerSrc {
   socklen_t client_sin_len;
   int client_sock_fd;
 
+  int control_fds[2];
+
   /* number of bytes we've gotten */
   off_t curoffset;