tcpclientsink: Port to GIO

gst/tcp/gsttcpclientsink.c

@@ -1,6 +1,8 @@
 /* GStreamer
  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
  * Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
+ * Copyright (C) <2011> Collabora Ltd.
+ *     Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
  *
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Library General Public
@@ -20,7 +22,7 @@
 
 /**
  * SECTION:element-tcpclientsink
- * @see_also: #tcpclientsrc
+ * @see_also: #tcpclientsink
  *
  * <refsect2>
  * <title>Example launch line</title>
@@ -28,7 +30,7 @@
  * # server:
  * nc -l -p 3000
  * # client:
- * gst-launch fdsrc fd=1 ! tcpclientsink port=3000
+ * gst-launch fdsink fd=1 ! tcpclientsink port=3000
  * ]| everything you type in the client is shown on the server
  * </refsect2>
  */
@@ -37,9 +39,9 @@
 #include "config.h"
 #endif
 #include <gst/gst-i18n-plugin.h>
+
 #include "gsttcp.h"
 #include "gsttcpclientsink.h"
-#include <string.h>             /* memset */
 
 /* TCPClientSink signals and args */
 enum
@@ -54,9 +56,9 @@ GST_DEBUG_CATEGORY_STATIC (tcpclientsink_debug);
 
 enum
 {
-  ARG_0,
+  PROP_0,
-  ARG_HOST,
+  PROP_HOST,
-  ARG_PORT
+  PROP_PORT
 };
 
 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
@@ -70,8 +72,10 @@ static gboolean gst_tcp_client_sink_setcaps (GstBaseSink * bsink,
     GstCaps * caps);
 static GstFlowReturn gst_tcp_client_sink_render (GstBaseSink * bsink,
     GstBuffer * buf);
-static GstStateChangeReturn gst_tcp_client_sink_change_state (GstElement *
+static gboolean gst_tcp_client_sink_start (GstBaseSink * bsink);
-    element, GstStateChange transition);
+static gboolean gst_tcp_client_sink_stop (GstBaseSink * bsink);
+static gboolean gst_tcp_client_sink_unlock (GstBaseSink * bsink);
+static gboolean gst_tcp_client_sink_unlock_stop (GstBaseSink * bsink);
 
 static void gst_tcp_client_sink_set_property (GObject * object, guint prop_id,
     const GValue * value, GParamSpec * pspec);
@@ -101,10 +105,10 @@ gst_tcp_client_sink_class_init (GstTCPClientSinkClass * klass)
   gobject_class->get_property = gst_tcp_client_sink_get_property;
   gobject_class->finalize = gst_tcp_client_sink_finalize;
 
-  g_object_class_install_property (gobject_class, ARG_HOST,
+  g_object_class_install_property (gobject_class, PROP_HOST,
       g_param_spec_string ("host", "Host", "The host/IP to send the packets to",
           TCP_DEFAULT_HOST, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
-  g_object_class_install_property (gobject_class, ARG_PORT,
+  g_object_class_install_property (gobject_class, PROP_PORT,
       g_param_spec_int ("port", "Port", "The port to send the packets to",
           0, TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
@@ -117,10 +121,12 @@ gst_tcp_client_sink_class_init (GstTCPClientSinkClass * klass)
       "Send data as a client over the network via TCP",
       "Thomas Vander Stichele <thomas at apestaart dot org>");
 
-  gstelement_class->change_state = gst_tcp_client_sink_change_state;
+  gstbasesink_class->start = gst_tcp_client_sink_start;
-
+  gstbasesink_class->stop = gst_tcp_client_sink_stop;
   gstbasesink_class->set_caps = gst_tcp_client_sink_setcaps;
   gstbasesink_class->render = gst_tcp_client_sink_render;
+  gstbasesink_class->unlock = gst_tcp_client_sink_unlock;
+  gstbasesink_class->unlock_stop = gst_tcp_client_sink_unlock_stop;
 
   GST_DEBUG_CATEGORY_INIT (tcpclientsink_debug, "tcpclientsink", 0, "TCP sink");
 }
@@ -131,7 +137,9 @@ gst_tcp_client_sink_init (GstTCPClientSink * this)
   this->host = g_strdup (TCP_DEFAULT_HOST);
   this->port = TCP_DEFAULT_PORT;
 
-  this->sock_fd.fd = -1;
+  this->socket = NULL;
+  this->cancellable = g_cancellable_new ();
+
   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SINK_OPEN);
 }
 
@@ -140,7 +148,16 @@ gst_tcp_client_sink_finalize (GObject * gobject)
 {
   GstTCPClientSink *this = GST_TCP_CLIENT_SINK (gobject);
 
+  if (this->cancellable)
+    g_object_unref (this->cancellable);
+  this->cancellable = NULL;
+
+  if (this->socket)
+    g_object_unref (this->socket);
+  this->socket = NULL;
+
   g_free (this->host);
+  this->host = NULL;
 
   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 }
@@ -154,10 +171,11 @@ gst_tcp_client_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
 static GstFlowReturn
 gst_tcp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf)
 {
-  size_t wrote = 0;
   GstTCPClientSink *sink;
   guint8 *data;
-  gsize size;
+  gsize size, written = 0;
+  gssize rret;
+  GError *err = NULL;
 
   sink = GST_TCP_CLIENT_SINK (bsink);
 
@@ -169,24 +187,39 @@ gst_tcp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf)
       size);
 
   /* write buffer data */
-  wrote = gst_tcp_socket_write (sink->sock_fd.fd, data, size);
+  while (written < size) {
+    rret =
+        g_socket_send (sink->socket, (gchar *) data + written, size - written,
+        sink->cancellable, &err);
+    if (rret < 0)
+      goto write_error;
+    written += rret;
+  }
   gst_buffer_unmap (buf, data, size);
 
-  if (wrote < size)
+  sink->data_written += written;
-    goto write_error;
-
-  sink->data_written += wrote;
 
   return GST_FLOW_OK;
 
   /* ERRORS */
 write_error:
   {
-    GST_ELEMENT_ERROR (sink, RESOURCE, WRITE,
+    GstFlowReturn ret;
-        (_("Error while sending data to \"%s:%d\"."), sink->host, sink->port),
+
-        ("Only %" G_GSIZE_FORMAT " of %" G_GSIZE_FORMAT " bytes written: %s",
+    gst_buffer_unmap (buf, data, size);
-            wrote, size, g_strerror (errno)));
+
-    return GST_FLOW_ERROR;
+    if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+      ret = GST_FLOW_WRONG_STATE;
+      GST_DEBUG_OBJECT (sink, "Cancelled reading from socket");
+    } else {
+      GST_ELEMENT_ERROR (sink, RESOURCE, WRITE,
+          (_("Error while sending data to \"%s:%d\"."), sink->host, sink->port),
+          ("Only %" G_GSIZE_FORMAT " of %" G_GSIZE_FORMAT " bytes written: %s",
+              written, size, err->message));
+      ret = GST_FLOW_ERROR;
+    }
+    g_clear_error (&err);
+    return ret;
   }
 }
 
@@ -200,7 +233,7 @@ gst_tcp_client_sink_set_property (GObject * object, guint prop_id,
   tcpclientsink = GST_TCP_CLIENT_SINK (object);
 
   switch (prop_id) {
-    case ARG_HOST:
+    case PROP_HOST:
       if (!g_value_get_string (value)) {
         g_warning ("host property cannot be NULL");
         break;
@@ -208,7 +241,7 @@ gst_tcp_client_sink_set_property (GObject * object, guint prop_id,
       g_free (tcpclientsink->host);
       tcpclientsink->host = g_strdup (g_value_get_string (value));
       break;
-    case ARG_PORT:
+    case PROP_PORT:
       tcpclientsink->port = g_value_get_int (value);
       break;
 
@@ -228,13 +261,12 @@ gst_tcp_client_sink_get_property (GObject * object, guint prop_id,
   tcpclientsink = GST_TCP_CLIENT_SINK (object);
 
   switch (prop_id) {
-    case ARG_HOST:
+    case PROP_HOST:
       g_value_set_string (value, tcpclientsink->host);
       break;
-    case ARG_PORT:
+    case PROP_PORT:
       g_value_set_int (value, tcpclientsink->port);
       break;
-
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -244,114 +276,146 @@ gst_tcp_client_sink_get_property (GObject * object, guint prop_id,
 
 /* create a socket for sending to remote machine */
 static gboolean
-gst_tcp_client_sink_start (GstTCPClientSink * this)
+gst_tcp_client_sink_start (GstBaseSink * bsink)
 {
-  int ret;
+  GstTCPClientSink *this = GST_TCP_CLIENT_SINK (bsink);
-  gchar *ip;
+  GError *err = NULL;
+  GInetAddress *addr;
+  GSocketAddress *saddr;
 
   if (GST_OBJECT_FLAG_IS_SET (this, GST_TCP_CLIENT_SINK_OPEN))
     return TRUE;
 
-  /* reset caps_sent flag */
-  this->caps_sent = FALSE;
-
   /* create sending client socket */
   GST_DEBUG_OBJECT (this, "opening sending client socket to %s:%d", this->host,
       this->port);
-  if ((this->sock_fd.fd = socket (AF_INET, SOCK_STREAM, 0)) == -1) {
+  this->socket =
-    GST_ELEMENT_ERROR (this, RESOURCE, OPEN_WRITE, (NULL), GST_ERROR_SYSTEM);
+      g_socket_new (G_SOCKET_FAMILY_IPV4, G_SOCKET_TYPE_STREAM,
-    return FALSE;
+      G_SOCKET_PROTOCOL_TCP, &err);
-  }
+  if (!this->socket)
-  GST_DEBUG_OBJECT (this, "opened sending client socket with fd %d",
+    goto no_socket;
-      this->sock_fd.fd);
+
+  GST_DEBUG_OBJECT (this, "opened sending client socket");
 
   /* look up name if we need to */
-  ip = gst_tcp_host_to_ip (GST_ELEMENT (this), this->host);
+  addr = g_inet_address_new_from_string (this->host);
-  if (!ip) {
+  if (!addr) {
-    gst_tcp_socket_close (&this->sock_fd);
+    GResolver *resolver = g_resolver_get_default ();
-    return FALSE;
+    GList *results;
+
+    results =
+        g_resolver_lookup_by_name (resolver, this->host, this->cancellable,
+        &err);
+    if (!results)
+      goto name_resolve;
+    addr = G_INET_ADDRESS (g_object_ref (results->data));
+
+    g_resolver_free_addresses (results);
+    g_object_unref (resolver);
   }
-  GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
+#ifndef GST_DISABLE_GST_DEBUG
+  {
+    gchar *ip = g_inet_address_to_string (addr);
+
+    GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
+    g_free (ip);
+  }
+#endif
 
   /* connect to server */
-  memset (&this->server_sin, 0, sizeof (this->server_sin));
+  saddr = g_inet_socket_address_new (addr, this->port);
-  this->server_sin.sin_family = AF_INET;        /* network socket */
+  if (!g_socket_connect (this->socket, saddr, this->cancellable, &err))
-  this->server_sin.sin_port = htons (this->port);       /* on port */
+    goto connect_failed;
-  this->server_sin.sin_addr.s_addr = inet_addr (ip);    /* on host ip */
-  g_free (ip);
 
-  GST_DEBUG_OBJECT (this, "connecting to server");
+  g_object_unref (saddr);
-  ret = connect (this->sock_fd.fd, (struct sockaddr *) &this->server_sin,
+  g_object_unref (addr);
-      sizeof (this->server_sin));
-
-  if (ret) {
-    gst_tcp_socket_close (&this->sock_fd);
-    switch (errno) {
-      case ECONNREFUSED:
-        GST_ELEMENT_ERROR (this, RESOURCE, OPEN_WRITE,
-            (_("Connection to %s:%d refused."), this->host, this->port),
-            (NULL));
-        return FALSE;
-        break;
-      default:
-        GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
-            ("connect to %s:%d failed: %s", this->host, this->port,
-                g_strerror (errno)));
-        return FALSE;
-        break;
-    }
-  }
 
   GST_OBJECT_FLAG_SET (this, GST_TCP_CLIENT_SINK_OPEN);
 
   this->data_written = 0;
 
   return TRUE;
+no_socket:
+  {
+    GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
+        ("Failed to create socket: %s", err->message));
+    g_clear_error (&err);
+    return FALSE;
+  }
+name_resolve:
+  {
+    if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+      GST_DEBUG_OBJECT (this, "Cancelled name resolval");
+    } else {
+      GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
+          ("Failed to resolve host '%s': %s", this->host, err->message));
+    }
+    g_clear_error (&err);
+    gst_tcp_client_sink_stop (GST_BASE_SINK (this));
+    return FALSE;
+  }
+connect_failed:
+  {
+    if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+      GST_DEBUG_OBJECT (this, "Cancelled connecting");
+    } else {
+      GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
+          ("Failed to connect to host '%s:%d': %s", this->host, this->port,
+              err->message));
+    }
+    g_clear_error (&err);
+    g_object_unref (saddr);
+    g_object_unref (addr);
+    gst_tcp_client_sink_stop (GST_BASE_SINK (this));
+    return FALSE;
+  }
 }
 
 static gboolean
-gst_tcp_client_sink_stop (GstTCPClientSink * this)
+gst_tcp_client_sink_stop (GstBaseSink * bsink)
 {
+  GstTCPClientSink *this = GST_TCP_CLIENT_SINK (bsink);
+  GError *err = NULL;
+
   if (!GST_OBJECT_FLAG_IS_SET (this, GST_TCP_CLIENT_SINK_OPEN))
     return TRUE;
 
-  gst_tcp_socket_close (&this->sock_fd);
+  if (this->socket) {
+    GST_DEBUG_OBJECT (this, "closing socket");
+
+    if (!g_socket_close (this->socket, &err)) {
+      GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
+      g_clear_error (&err);
+    }
+    g_object_unref (this->socket);
+    this->socket = NULL;
+  }
 
   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SINK_OPEN);
 
   return TRUE;
 }
 
-static GstStateChangeReturn
+/* will be called only between calls to start() and stop() */
-gst_tcp_client_sink_change_state (GstElement * element,
+static gboolean
-    GstStateChange transition)
+gst_tcp_client_sink_unlock (GstBaseSink * bsink)
 {
-  GstTCPClientSink *sink;
+  GstTCPClientSink *sink = GST_TCP_CLIENT_SINK (bsink);
-  GstStateChangeReturn res;
 
-  sink = GST_TCP_CLIENT_SINK (element);
+  GST_DEBUG_OBJECT (sink, "set to flushing");
+  g_cancellable_cancel (sink->cancellable);
 
-  switch (transition) {
+  return TRUE;
-    case GST_STATE_CHANGE_NULL_TO_READY:
+}
-    case GST_STATE_CHANGE_READY_TO_PAUSED:
+
-      if (!gst_tcp_client_sink_start (sink))
+/* will be called only between calls to start() and stop() */
-        goto start_failure;
+static gboolean
-      break;
+gst_tcp_client_sink_unlock_stop (GstBaseSink * bsink)
-    default:
+{
-      break;
+  GstTCPClientSink *sink = GST_TCP_CLIENT_SINK (bsink);
-  }
+
-  res = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+  GST_DEBUG_OBJECT (sink, "unset flushing");
-
+  g_cancellable_reset (sink->cancellable);
-  switch (transition) {
+
-    case GST_STATE_CHANGE_READY_TO_NULL:
+  return TRUE;
-      gst_tcp_client_sink_stop (sink);
-    default:
-      break;
-  }
-  return res;
-
-start_failure:
-  {
-    return GST_STATE_CHANGE_FAILURE;
-  }
 }

gst/tcp/gsttcpclientsink.h

@@ -25,25 +25,12 @@
 #include <gst/gst.h>
 #include <gst/base/gstbasesink.h>
 
+#include <gio/gio.h>
+
 #include "gsttcp.h"
 
 G_BEGIN_DECLS
 
-#include <stdio.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <errno.h>
-#include <string.h>
-#include <sys/time.h>
-#include <sys/types.h>
-#include <netinet/in.h>
-#include <netdb.h>
-#include <sys/socket.h>
-#include <sys/wait.h>
-#include <fcntl.h>
-#include <arpa/inet.h>
-#include "gsttcp.h"
-
 #define GST_TYPE_TCP_CLIENT_SINK \
   (gst_tcp_client_sink_get_type())
 #define GST_TCP_CLIENT_SINK(obj) \
@@ -70,13 +57,12 @@ struct _GstTCPClientSink {
   /* server information */
   int port;
   gchar *host;
-  struct sockaddr_in server_sin;
 
   /* socket */
-  GstPollFD sock_fd;
+  GSocket *socket;
+  GCancellable *cancellable;
 
   size_t data_written; /* how much bytes have we written ? */
-  gboolean caps_sent; /* whether or not we sent caps already */
 };
 
 struct _GstTCPClientSinkClass {