plugins/elements/gstfdsink.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *
   5  * gstfdsink.c:
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 /**
  24  * SECTION:element-fdsink
  25  * @short_description: write to a unix file descriptor
  26  * @see_also: #GstFdSrc
  27  *
  28  * Write data to a unix file descriptor.
  29  *
  30  * This element will sycnhronize on the clock before writing the data on the
  31  * socket. For file descriptors where this does not make sense (files, ...) the
  32  * ::sync property can be used to disable synchronisation.
  33  *
  34  * Last reviewed on 2006-04-28 (0.10.6)
  35  */
  36
  37 #ifdef HAVE_CONFIG_H
  38 #  include "config.h"
  39 #endif
  40
  41 #include "../../gst/gst-i18n-lib.h"
  42
  43 #include <sys/types.h>
  44 #include <sys/stat.h>
  45 #include <sys/socket.h>
  46 #include <fcntl.h>
  47 #include <stdio.h>
  48 #ifdef HAVE_UNISTD_H
  49 #include <unistd.h>
  50 #endif
  51 #ifdef _MSC_VER
  52 #include <io.h>
  53 #endif
  54 #include <errno.h>
  55 #include <string.h>
  56
  57 #include "gstfdsink.h"
  58
  59 /* We add a control socket as in fdsrc to make it shutdown quickly when it's blocking on the fd.
  60  * Select is used to determine when the fd is ready for use. When the element state is changed,
  61  * it happens from another thread while fdsink is select'ing on the fd. The state-change thread
  62  * sends a control message, so fdsink wakes up and changes state immediately otherwise
  63  * it would stay blocked until it receives some data. */
  64
  65 /* the select call is also performed on the control sockets, that way
  66  * we can send special commands to unblock the select call */
  67 #define CONTROL_STOP            'S'     /* stop the select call */
  68 #define CONTROL_SOCKETS(sink)   sink->control_sock
  69 #define WRITE_SOCKET(sink)      sink->control_sock[1]
  70 #define READ_SOCKET(sink)       sink->control_sock[0]
  71
  72 #define SEND_COMMAND(sink, command)          \
  73 G_STMT_START {                              \
  74   unsigned char c; c = command;             \
  75   write (WRITE_SOCKET(sink), &c, 1);         \
  76 } G_STMT_END
  77
  78 #define READ_COMMAND(sink, command, res)        \
  79 G_STMT_START {                                 \
  80   res = read(READ_SOCKET(sink), &command, 1);   \
  81 } G_STMT_END
  82
  83 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  84     GST_PAD_SINK,
  85     GST_PAD_ALWAYS,
  86     GST_STATIC_CAPS_ANY);
  87
  88 GST_DEBUG_CATEGORY_STATIC (gst_fd_sink__debug);
  89 #define GST_CAT_DEFAULT gst_fd_sink__debug
  90
  91 static const GstElementDetails gst_fd_sink__details =
  92 GST_ELEMENT_DETAILS ("Filedescriptor Sink",
  93     "Sink/File",
  94     "Write data to a file descriptor",
  95     "Erik Walthinsen <omega@cse.ogi.edu>");
  96
  97
  98 /* FdSink signals and args */
  99 enum
 100 {
 101   /* FILL ME */
 102   LAST_SIGNAL
 103 };
 104
 105 enum
 106 {
 107   ARG_0,
 108   ARG_FD
 109 };
 110
 111 static void gst_fd_sink_uri_handler_init (gpointer g_iface,
 112     gpointer iface_data);
 113
 114 static void
 115 _do_init (GType gst_fd_sink_type)
 116 {
 117   static const GInterfaceInfo urihandler_info = {
 118     gst_fd_sink_uri_handler_init,
 119     NULL,
 120     NULL
 121   };
 122
 123   g_type_add_interface_static (gst_fd_sink_type, GST_TYPE_URI_HANDLER,
 124       &urihandler_info);
 125
 126   GST_DEBUG_CATEGORY_INIT (gst_fd_sink__debug, "fdsink", 0, "fdsink element");
 127 }
 128
 129 GST_BOILERPLATE_FULL (GstFdSink, gst_fd_sink, GstBaseSink, GST_TYPE_BASE_SINK,
 130     _do_init);
 131
 132 static void gst_fd_sink_set_property (GObject * object, guint prop_id,
 133     const GValue * value, GParamSpec * pspec);
 134 static void gst_fd_sink_get_property (GObject * object, guint prop_id,
 135     GValue * value, GParamSpec * pspec);
 136 static void gst_fd_sink_dispose (GObject * obj);
 137
 138 static gboolean gst_fd_sink_query (GstPad * pad, GstQuery * query);
 139 static GstFlowReturn gst_fd_sink_render (GstBaseSink * sink,
 140     GstBuffer * buffer);
 141 static gboolean gst_fd_sink_start (GstBaseSink * basesink);
 142 static gboolean gst_fd_sink_stop (GstBaseSink * basesink);
 143 static gboolean gst_fd_sink_unlock (GstBaseSink * basesink);
 144
 145 static void
 146 gst_fd_sink_base_init (gpointer g_class)
 147 {
 148   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 149
 150   gst_element_class_add_pad_template (gstelement_class,
 151       gst_static_pad_template_get (&sinktemplate));
 152   gst_element_class_set_details (gstelement_class, &gst_fd_sink__details);
 153 }
 154
 155 static void
 156 gst_fd_sink_class_init (GstFdSinkClass * klass)
 157 {
 158   GObjectClass *gobject_class;
 159   GstBaseSinkClass *gstbasesink_class;
 160
 161   gobject_class = G_OBJECT_CLASS (klass);
 162   gstbasesink_class = GST_BASE_SINK_CLASS (klass);
 163
 164   gobject_class->set_property = gst_fd_sink_set_property;
 165   gobject_class->get_property = gst_fd_sink_get_property;
 166   gobject_class->dispose = gst_fd_sink_dispose;
 167
 168   gstbasesink_class->get_times = NULL;
 169   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_fd_sink_render);
 170   gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_fd_sink_start);
 171   gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_fd_sink_stop);
 172   gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_fd_sink_unlock);
 173   gstbasesink_class->event = NULL;
 174
 175   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_FD,
 176       g_param_spec_int ("fd", "fd", "An open file descriptor to write to",
 177           0, G_MAXINT, 1, G_PARAM_READWRITE));
 178 }
 179
 180 static void
 181 gst_fd_sink_init (GstFdSink * fdsink, GstFdSinkClass * klass)
 182 {
 183   GstPad *pad;
 184
 185   pad = GST_BASE_SINK_PAD (fdsink);
 186   gst_pad_set_query_function (pad, GST_DEBUG_FUNCPTR (gst_fd_sink_query));
 187
 188   fdsink->fd = 1;
 189   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
 190   fdsink->bytes_written = 0;
 191
 192   GST_BASE_SINK (fdsink)->sync = TRUE;
 193 }
 194
 195 static void
 196 gst_fd_sink_dispose (GObject * obj)
 197 {
 198   GstFdSink *fdsink = GST_FD_SINK (obj);
 199
 200   g_free (fdsink->uri);
 201   fdsink->uri = NULL;
 202
 203   G_OBJECT_CLASS (parent_class)->dispose (obj);
 204 }
 205
 206 static gboolean
 207 gst_fd_sink_query (GstPad * pad, GstQuery * query)
 208 {
 209   GstFdSink *fdsink;
 210   GstFormat format;
 211
 212   fdsink = GST_FD_SINK (GST_PAD_PARENT (pad));
 213
 214   switch (GST_QUERY_TYPE (query)) {
 215     case GST_QUERY_POSITION:
 216       gst_query_parse_position (query, &format, NULL);
 217       switch (format) {
 218         case GST_FORMAT_DEFAULT:
 219         case GST_FORMAT_BYTES:
 220           gst_query_set_position (query, GST_FORMAT_BYTES,
 221               fdsink->bytes_written);
 222           return TRUE;
 223         default:
 224           return FALSE;
 225       }
 226
 227     case GST_QUERY_FORMATS:
 228       gst_query_set_formats (query, 2, GST_FORMAT_DEFAULT, GST_FORMAT_BYTES);
 229       return TRUE;
 230
 231     default:
 232       return gst_pad_query_default (pad, query);
 233   }
 234 }
 235
 236 static GstFlowReturn
 237 gst_fd_sink_render (GstBaseSink * sink, GstBuffer * buffer)
 238 {
 239   GstFdSink *fdsink;
 240
 241 #ifndef HAVE_WIN32
 242   fd_set readfds;
 243   fd_set writefds;
 244   gint retval;
 245 #endif
 246   guint8 *data;
 247   guint size;
 248   gint written;
 249
 250   fdsink = GST_FD_SINK (sink);
 251
 252   g_return_val_if_fail (fdsink->fd >= 0, GST_FLOW_ERROR);
 253
 254   data = GST_BUFFER_DATA (buffer);
 255   size = GST_BUFFER_SIZE (buffer);
 256
 257 again:
 258 #ifndef HAVE_WIN32
 259
 260   FD_ZERO (&readfds);
 261   FD_SET (READ_SOCKET (fdsink), &readfds);
 262
 263   FD_ZERO (&writefds);
 264   FD_SET (fdsink->fd, &writefds);
 265
 266   do {
 267     GST_DEBUG_OBJECT (fdsink, "going into select, have %d bytes to write",
 268         size);
 269     retval = select (FD_SETSIZE, &readfds, &writefds, NULL, NULL);
 270   } while ((retval == -1 && errno == EINTR));
 271
 272   if (retval == -1)
 273     goto select_error;
 274
 275   if (FD_ISSET (READ_SOCKET (fdsink), &readfds)) {
 276     /* read all stop commands */
 277     while (TRUE) {
 278       gchar command;
 279       int res;
 280
 281       READ_COMMAND (fdsink, command, res);
 282       if (res < 0) {
 283         GST_LOG_OBJECT (fdsink, "no more commands");
 284         /* no more commands */
 285         break;
 286       }
 287     }
 288     goto stopped;
 289   }
 290 #endif
 291
 292   GST_DEBUG_OBJECT (fdsink, "writing %d bytes to file descriptor %d", size,
 293       fdsink->fd);
 294
 295   written = write (fdsink->fd, data, size);
 296
 297   /* check for errors */
 298   if (G_UNLIKELY (written < 0)) {
 299     /* try to write again on non-fatal errors */
 300     if (errno == EAGAIN || errno == EINTR)
 301       goto again;
 302
 303     /* else go to our error handler */
 304     goto write_error;
 305   }
 306
 307   /* all is fine when we get here */
 308   size -= written;
 309   data += written;
 310   fdsink->bytes_written += written;
 311
 312   GST_DEBUG_OBJECT (fdsink, "wrote %d bytes, %d left", written, size);
 313
 314   /* short write, select and try to write the remainder */
 315   if (G_UNLIKELY (size > 0))
 316     goto again;
 317
 318   return GST_FLOW_OK;
 319
 320 #ifndef HAVE_WIN32
 321 select_error:
 322   {
 323     GST_ELEMENT_ERROR (fdsink, RESOURCE, READ, (NULL),
 324         ("select on file descriptor: %s.", g_strerror (errno)));
 325     GST_DEBUG_OBJECT (fdsink, "Error during select");
 326     return GST_FLOW_ERROR;
 327   }
 328 stopped:
 329   {
 330     GST_DEBUG_OBJECT (fdsink, "Select stopped");
 331     return GST_FLOW_WRONG_STATE;
 332   }
 333 #endif
 334
 335 write_error:
 336   {
 337     switch (errno) {
 338       case ENOSPC:
 339         GST_ELEMENT_ERROR (fdsink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
 340         break;
 341       default:{
 342         GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE,
 343             (_("Error while writing to file descriptor \"%d\"."), fdsink->fd),
 344             ("%s", g_strerror (errno)));
 345       }
 346     }
 347     return GST_FLOW_ERROR;
 348   }
 349 }
 350
 351 static gboolean
 352 gst_fd_sink_check_fd (GstFdSink * fdsink, int fd)
 353 {
 354   struct stat stat_results;
 355   off_t result;
 356
 357   /* see that it is a valid file descriptor */
 358   if (fstat (fd, &stat_results) < 0)
 359     goto invalid;
 360
 361   if (!S_ISREG (stat_results.st_mode))
 362     goto not_seekable;
 363
 364   /* see if it is a seekable stream */
 365   result = lseek (fd, 0, SEEK_CUR);
 366   if (result == -1) {
 367     switch (errno) {
 368       case EINVAL:
 369       case EBADF:
 370         goto invalid;
 371
 372       case ESPIPE:
 373         goto not_seekable;
 374     }
 375   } else
 376     GST_DEBUG_OBJECT (fdsink, "File descriptor \"%d\" is seekable", fd);
 377
 378   return TRUE;
 379
 380 invalid:
 381   {
 382     GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE,
 383         (_("File descriptor \"%d\" is not valid."), fd),
 384         ("%s", g_strerror (errno)));
 385     return FALSE;
 386   }
 387 not_seekable:
 388   {
 389     GST_DEBUG_OBJECT (fdsink, "File descriptor \"%d\" is a pipe", fd);
 390     return TRUE;
 391   }
 392 }
 393
 394 static gboolean
 395 gst_fd_sink_start (GstBaseSink * basesink)
 396 {
 397   GstFdSink *fdsink;
 398   gint control_sock[2];
 399
 400   fdsink = GST_FD_SINK (basesink);
 401   if (!gst_fd_sink_check_fd (fdsink, fdsink->fd))
 402     return FALSE;
 403
 404   if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
 405     goto socket_pair;
 406
 407   READ_SOCKET (fdsink) = control_sock[0];
 408   WRITE_SOCKET (fdsink) = control_sock[1];
 409
 410   fcntl (READ_SOCKET (fdsink), F_SETFL, O_NONBLOCK);
 411   fcntl (WRITE_SOCKET (fdsink), F_SETFL, O_NONBLOCK);
 412
 413   return TRUE;
 414
 415   /* ERRORS */
 416 socket_pair:
 417   {
 418     GST_ELEMENT_ERROR (fdsink, RESOURCE, OPEN_READ_WRITE, (NULL),
 419         GST_ERROR_SYSTEM);
 420     return FALSE;
 421   }
 422 }
 423
 424 static gboolean
 425 gst_fd_sink_stop (GstBaseSink * basesink)
 426 {
 427   GstFdSink *fdsink = GST_FD_SINK (basesink);
 428
 429   close (READ_SOCKET (fdsink));
 430   close (WRITE_SOCKET (fdsink));
 431
 432   return TRUE;
 433 }
 434
 435 static gboolean
 436 gst_fd_sink_unlock (GstBaseSink * basesink)
 437 {
 438   GstFdSink *fdsink = GST_FD_SINK (basesink);
 439
 440   SEND_COMMAND (fdsink, CONTROL_STOP);
 441
 442   return TRUE;
 443 }
 444
 445 static gboolean
 446 gst_fd_sink_update_fd (GstFdSink * fdsink, int new_fd)
 447 {
 448   if (new_fd < 0)
 449     return FALSE;
 450
 451   if (!gst_fd_sink_check_fd (fdsink, new_fd))
 452     goto invalid;
 453
 454   /* assign the fd */
 455   GST_OBJECT_LOCK (fdsink);
 456   fdsink->fd = new_fd;
 457   g_free (fdsink->uri);
 458   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
 459   GST_OBJECT_UNLOCK (fdsink);
 460
 461   return TRUE;
 462
 463 invalid:
 464   {
 465     return FALSE;
 466   }
 467 }
 468
 469 static void
 470 gst_fd_sink_set_property (GObject * object, guint prop_id,
 471     const GValue * value, GParamSpec * pspec)
 472 {
 473   GstFdSink *fdsink;
 474
 475   g_return_if_fail (GST_IS_FD_SINK (object));
 476
 477   fdsink = GST_FD_SINK (object);
 478
 479   switch (prop_id) {
 480     case ARG_FD:{
 481       int fd;
 482
 483       fd = g_value_get_int (value);
 484       gst_fd_sink_update_fd (fdsink, fd);
 485       break;
 486     }
 487     default:
 488       break;
 489   }
 490 }
 491
 492 static void
 493 gst_fd_sink_get_property (GObject * object, guint prop_id, GValue * value,
 494     GParamSpec * pspec)
 495 {
 496   GstFdSink *fdsink;
 497
 498   g_return_if_fail (GST_IS_FD_SINK (object));
 499
 500   fdsink = GST_FD_SINK (object);
 501
 502   switch (prop_id) {
 503     case ARG_FD:
 504       g_value_set_int (value, fdsink->fd);
 505       break;
 506     default:
 507       break;
 508   }
 509 }
 510
 511 /*** GSTURIHANDLER INTERFACE *************************************************/
 512
 513 static guint
 514 gst_fd_sink_uri_get_type (void)
 515 {
 516   return GST_URI_SINK;
 517 }
 518 static gchar **
 519 gst_fd_sink_uri_get_protocols (void)
 520 {
 521   static gchar *protocols[] = { "fd", NULL };
 522
 523   return protocols;
 524 }
 525 static const gchar *
 526 gst_fd_sink_uri_get_uri (GstURIHandler * handler)
 527 {
 528   GstFdSink *sink = GST_FD_SINK (handler);
 529
 530   return sink->uri;
 531 }
 532
 533 static gboolean
 534 gst_fd_sink_uri_set_uri (GstURIHandler * handler, const gchar * uri)
 535 {
 536   gchar *protocol;
 537   GstFdSink *sink = GST_FD_SINK (handler);
 538   gint fd;
 539
 540   protocol = gst_uri_get_protocol (uri);
 541   if (strcmp (protocol, "fd") != 0) {
 542     g_free (protocol);
 543     return FALSE;
 544   }
 545   g_free (protocol);
 546
 547   if (sscanf (uri, "fd://%d", &fd) != 1)
 548     return FALSE;
 549
 550   return gst_fd_sink_update_fd (sink, fd);
 551 }
 552
 553 static void
 554 gst_fd_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
 555 {
 556   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 557
 558   iface->get_type = gst_fd_sink_uri_get_type;
 559   iface->get_protocols = gst_fd_sink_uri_get_protocols;
 560   iface->get_uri = gst_fd_sink_uri_get_uri;
 561   iface->set_uri = gst_fd_sink_uri_set_uri;
 562 }