plugins/elements/gstfdsrc.c

   1 /* GStreamer
   2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
   3  *                    2000 Wim Taymans <wtay@chello.be>
   4  *                    2005 Philippe Khalaf <burger@speedy.org>
   5  *
   6  * gstfdsrc.c:
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  21  * Boston, MA 02111-1307, USA.
  22  */
  23 /**
  24  * SECTION:element-fdsrc
  25  * @short_description: read from a unix file descriptor
  26  * @see_also: #GstFdSink
  27  *
  28  * Read data from a unix file descriptor.
  29  */
  30
  31
  32 #ifdef HAVE_CONFIG_H
  33 #  include "config.h"
  34 #endif
  35 #include "gst/gst_private.h"
  36
  37 #include <sys/types.h>
  38 #include <sys/stat.h>
  39 #include <sys/socket.h>
  40 #include <fcntl.h>
  41 #include <stdio.h>
  42 #ifdef HAVE_UNISTD_H
  43 #include <unistd.h>
  44 #endif
  45 #ifdef _MSC_VER
  46 #include <io.h>
  47 #endif
  48 #include <stdlib.h>
  49 #include <errno.h>
  50
  51 #include "gstfdsrc.h"
  52
  53 /* the select call is also performed on the control sockets, that way
  54  * we can send special commands to unblock the select call */
  55 #define CONTROL_STOP            'S'     /* stop the select call */
  56 #define CONTROL_SOCKETS(src)   src->control_sock
  57 #define WRITE_SOCKET(src)      src->control_sock[1]
  58 #define READ_SOCKET(src)       src->control_sock[0]
  59
  60 #define SEND_COMMAND(src, command)          \
  61 G_STMT_START {                              \
  62   unsigned char c; c = command;             \
  63   write (WRITE_SOCKET(src), &c, 1);         \
  64 } G_STMT_END
  65
  66 #define READ_COMMAND(src, command, res)        \
  67 G_STMT_START {                                 \
  68   res = read(READ_SOCKET(src), &command, 1);   \
  69 } G_STMT_END
  70
  71 #define DEFAULT_BLOCKSIZE       4096
  72
  73 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  74     GST_PAD_SRC,
  75     GST_PAD_ALWAYS,
  76     GST_STATIC_CAPS_ANY);
  77
  78 GST_DEBUG_CATEGORY_STATIC (gst_fd_src_debug);
  79 #define GST_CAT_DEFAULT gst_fd_src_debug
  80
  81 static const GstElementDetails gst_fd_src_details =
  82 GST_ELEMENT_DETAILS ("Disk Source",
  83     "Source/File",
  84     "Synchronous read from a file",
  85     "Erik Walthinsen <omega@cse.ogi.edu>");
  86
  87 enum
  88 {
  89   PROP_0,
  90   PROP_FD,
  91 };
  92
  93 static void gst_fd_src_uri_handler_init (gpointer g_iface, gpointer iface_data);
  94
  95 static void
  96 _do_init (GType fd_src_type)
  97 {
  98   static const GInterfaceInfo urihandler_info = {
  99     gst_fd_src_uri_handler_init,
 100     NULL,
 101     NULL
 102   };
 103
 104   g_type_add_interface_static (fd_src_type, GST_TYPE_URI_HANDLER,
 105       &urihandler_info);
 106
 107   GST_DEBUG_CATEGORY_INIT (gst_fd_src_debug, "fdsrc", 0, "fdsrc element");
 108 }
 109
 110 GST_BOILERPLATE_FULL (GstFdSrc, gst_fd_src, GstElement, GST_TYPE_PUSH_SRC,
 111     _do_init);
 112
 113 static void gst_fd_src_set_property (GObject * object, guint prop_id,
 114     const GValue * value, GParamSpec * pspec);
 115 static void gst_fd_src_get_property (GObject * object, guint prop_id,
 116     GValue * value, GParamSpec * pspec);
 117 static void gst_fd_src_dispose (GObject * obj);
 118
 119 static gboolean gst_fd_src_start (GstBaseSrc * bsrc);
 120 static gboolean gst_fd_src_stop (GstBaseSrc * bsrc);
 121 static gboolean gst_fd_src_unlock (GstBaseSrc * bsrc);
 122 static gboolean gst_fd_src_is_seekable (GstBaseSrc * bsrc);
 123 static gboolean gst_fd_src_get_size (GstBaseSrc * src, guint64 * size);
 124
 125 static GstFlowReturn gst_fd_src_create (GstPushSrc * psrc, GstBuffer ** outbuf);
 126
 127 static void
 128 gst_fd_src_base_init (gpointer g_class)
 129 {
 130   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 131
 132   gst_element_class_add_pad_template (gstelement_class,
 133       gst_static_pad_template_get (&srctemplate));
 134   gst_element_class_set_details (gstelement_class, &gst_fd_src_details);
 135 }
 136
 137 static void
 138 gst_fd_src_class_init (GstFdSrcClass * klass)
 139 {
 140   GObjectClass *gobject_class;
 141   GstBaseSrcClass *gstbasesrc_class;
 142   GstElementClass *gstelement_class;
 143   GstPushSrcClass *gstpush_src_class;
 144
 145   gobject_class = G_OBJECT_CLASS (klass);
 146   gstelement_class = GST_ELEMENT_CLASS (klass);
 147   gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
 148   gstpush_src_class = GST_PUSH_SRC_CLASS (klass);
 149
 150   parent_class = g_type_class_peek_parent (klass);
 151
 152   gobject_class->set_property = gst_fd_src_set_property;
 153   gobject_class->get_property = gst_fd_src_get_property;
 154   gobject_class->dispose = gst_fd_src_dispose;
 155
 156   g_object_class_install_property (gobject_class, PROP_FD,
 157       g_param_spec_int ("fd", "fd", "An open file descriptor to read from",
 158           0, G_MAXINT, 0, G_PARAM_READWRITE));
 159
 160   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_fd_src_start);
 161   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_fd_src_stop);
 162   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_fd_src_unlock);
 163   gstbasesrc_class->is_seekable = GST_DEBUG_FUNCPTR (gst_fd_src_is_seekable);
 164   gstbasesrc_class->get_size = GST_DEBUG_FUNCPTR (gst_fd_src_get_size);
 165
 166   gstpush_src_class->create = GST_DEBUG_FUNCPTR (gst_fd_src_create);
 167 }
 168
 169 static void
 170 gst_fd_src_init (GstFdSrc * fdsrc, GstFdSrcClass * klass)
 171 {
 172   fdsrc->fd = 0;
 173   fdsrc->new_fd = 0;
 174   fdsrc->seekable_fd = FALSE;
 175   fdsrc->uri = g_strdup_printf ("fd://%d", fdsrc->fd);
 176   fdsrc->curoffset = 0;
 177 }
 178
 179 static void
 180 gst_fd_src_dispose (GObject * obj)
 181 {
 182   GstFdSrc *src = GST_FD_SRC (obj);
 183
 184   g_free (src->uri);
 185   src->uri = NULL;
 186
 187   G_OBJECT_CLASS (parent_class)->dispose (obj);
 188 }
 189
 190 static void
 191 gst_fd_src_update_fd (GstFdSrc * src)
 192 {
 193   struct stat stat_results;
 194
 195   src->fd = src->new_fd;
 196   GST_INFO_OBJECT (src, "Updating to fd %d", src->fd);
 197   g_free (src->uri);
 198   src->uri = g_strdup_printf ("fd://%d", src->fd);
 199
 200   if (fstat (src->fd, &stat_results) < 0)
 201     goto not_seekable;
 202
 203   if (!S_ISREG (stat_results.st_mode))
 204     goto not_seekable;
 205
 206   /* Try a seek of 0 bytes offset to check for seekability */
 207   if (lseek (src->fd, 0, SEEK_CUR) < 0)
 208     goto not_seekable;
 209
 210   src->seekable_fd = TRUE;
 211   return;
 212
 213 not_seekable:
 214   src->seekable_fd = FALSE;
 215 }
 216
 217 static gboolean
 218 gst_fd_src_start (GstBaseSrc * bsrc)
 219 {
 220   GstFdSrc *src = GST_FD_SRC (bsrc);
 221   gint control_sock[2];
 222
 223   src->curoffset = 0;
 224
 225   gst_fd_src_update_fd (src);
 226
 227   if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
 228     goto socket_pair;
 229
 230   READ_SOCKET (src) = control_sock[0];
 231   WRITE_SOCKET (src) = control_sock[1];
 232
 233   fcntl (READ_SOCKET (src), F_SETFL, O_NONBLOCK);
 234   fcntl (WRITE_SOCKET (src), F_SETFL, O_NONBLOCK);
 235
 236   return TRUE;
 237
 238   /* ERRORS */
 239 socket_pair:
 240   {
 241     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ_WRITE, (NULL),
 242         GST_ERROR_SYSTEM);
 243     return FALSE;
 244   }
 245 }
 246
 247 static gboolean
 248 gst_fd_src_stop (GstBaseSrc * bsrc)
 249 {
 250   GstFdSrc *src = GST_FD_SRC (bsrc);
 251
 252   close (READ_SOCKET (src));
 253   close (WRITE_SOCKET (src));
 254
 255   return TRUE;
 256 }
 257
 258 static gboolean
 259 gst_fd_src_unlock (GstBaseSrc * bsrc)
 260 {
 261   GstFdSrc *src = GST_FD_SRC (bsrc);
 262
 263   SEND_COMMAND (src, CONTROL_STOP);
 264
 265   return TRUE;
 266 }
 267
 268 static void
 269 gst_fd_src_set_property (GObject * object, guint prop_id, const GValue * value,
 270     GParamSpec * pspec)
 271 {
 272   GstFdSrc *src = GST_FD_SRC (object);
 273
 274   switch (prop_id) {
 275     case PROP_FD:
 276       src->new_fd = g_value_get_int (value);
 277
 278       /* If state is ready or below, update the current fd immediately
 279        * so it is reflected in get_properties and uri */
 280       GST_OBJECT_LOCK (object);
 281       if (GST_STATE (GST_ELEMENT (src)) <= GST_STATE_READY) {
 282         GST_DEBUG_OBJECT (src, "state ready or lower, updating to use new fd");
 283         gst_fd_src_update_fd (src);
 284       } else {
 285         GST_DEBUG_OBJECT (src, "state above ready, not updating to new fd yet");
 286       }
 287       GST_OBJECT_UNLOCK (object);
 288       break;
 289     default:
 290       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 291       break;
 292   }
 293 }
 294
 295 static void
 296 gst_fd_src_get_property (GObject * object, guint prop_id, GValue * value,
 297     GParamSpec * pspec)
 298 {
 299   GstFdSrc *src = GST_FD_SRC (object);
 300
 301   switch (prop_id) {
 302     case PROP_FD:
 303       g_value_set_int (value, src->fd);
 304       break;
 305     default:
 306       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 307       break;
 308   }
 309 }
 310
 311 static GstFlowReturn
 312 gst_fd_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 313 {
 314   GstFdSrc *src;
 315   GstBuffer *buf;
 316   glong readbytes;
 317   guint blocksize;
 318
 319 #ifndef HAVE_WIN32
 320   fd_set readfds;
 321   gint retval;
 322 #endif
 323
 324   src = GST_FD_SRC (psrc);
 325
 326 #ifndef HAVE_WIN32
 327   FD_ZERO (&readfds);
 328   FD_SET (src->fd, &readfds);
 329   FD_SET (READ_SOCKET (src), &readfds);
 330
 331   do {
 332     retval = select (FD_SETSIZE, &readfds, NULL, NULL, NULL);
 333   } while ((retval == -1 && errno == EINTR));
 334
 335   if (retval == -1)
 336     goto select_error;
 337
 338   if (FD_ISSET (READ_SOCKET (src), &readfds)) {
 339     /* read all stop commands */
 340     while (TRUE) {
 341       gchar command;
 342       int res;
 343
 344       READ_COMMAND (src, command, res);
 345       if (res < 0) {
 346         GST_LOG_OBJECT (src, "no more commands");
 347         /* no more commands */
 348         break;
 349       }
 350     }
 351     goto stopped;
 352   }
 353 #endif
 354
 355   blocksize = GST_BASE_SRC (src)->blocksize;
 356
 357   /* create the buffer */
 358   buf = gst_buffer_new_and_alloc (blocksize);
 359
 360   do {
 361     readbytes = read (src->fd, GST_BUFFER_DATA (buf), blocksize);
 362   } while (readbytes == -1 && errno == EINTR);  /* retry if interrupted */
 363
 364   if (readbytes < 0)
 365     goto read_error;
 366
 367   if (readbytes == 0)
 368     goto eos;
 369
 370   GST_BUFFER_OFFSET (buf) = src->curoffset;
 371   GST_BUFFER_SIZE (buf) = readbytes;
 372   GST_BUFFER_TIMESTAMP (buf) = GST_CLOCK_TIME_NONE;
 373   src->curoffset += readbytes;
 374
 375   GST_LOG_OBJECT (psrc, "Read buffer of size %ld", readbytes);
 376
 377   /* we're done, return the buffer */
 378   *outbuf = buf;
 379
 380   return GST_FLOW_OK;
 381
 382   /* ERRORS */
 383 #ifndef HAVE_WIN32
 384 select_error:
 385   {
 386     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 387         ("select on file descriptor: %s.", g_strerror (errno)));
 388     GST_DEBUG_OBJECT (psrc, "Error during select");
 389     return GST_FLOW_ERROR;
 390   }
 391 stopped:
 392   {
 393     GST_DEBUG_OBJECT (psrc, "Select stopped");
 394     return GST_FLOW_WRONG_STATE;
 395   }
 396 #endif
 397 eos:
 398   {
 399     GST_DEBUG_OBJECT (psrc, "Read 0 bytes. EOS.");
 400     gst_buffer_unref (buf);
 401     return GST_FLOW_UNEXPECTED;
 402   }
 403 read_error:
 404   {
 405     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 406         ("read on file descriptor: %s.", g_strerror (errno)));
 407     GST_DEBUG_OBJECT (psrc, "Error reading from fd");
 408     gst_buffer_unref (buf);
 409     return GST_FLOW_ERROR;
 410   }
 411 }
 412
 413 gboolean
 414 gst_fd_src_is_seekable (GstBaseSrc * bsrc)
 415 {
 416   GstFdSrc *src = GST_FD_SRC (bsrc);
 417
 418   return src->seekable_fd;
 419 }
 420
 421 gboolean
 422 gst_fd_src_get_size (GstBaseSrc * bsrc, guint64 * size)
 423 {
 424   GstFdSrc *src = GST_FD_SRC (bsrc);
 425   struct stat stat_results;
 426
 427   if (!src->seekable_fd) {
 428     /* If it isn't seekable, we won't know the length (but fstat will still
 429      * succeed, and wrongly say our length is zero. */
 430     return FALSE;
 431   }
 432
 433   if (fstat (src->fd, &stat_results) < 0)
 434     goto could_not_stat;
 435
 436   *size = stat_results.st_size;
 437
 438   return TRUE;
 439
 440   /* ERROR */
 441 could_not_stat:
 442   {
 443     return FALSE;
 444   }
 445
 446 }
 447
 448 /*** GSTURIHANDLER INTERFACE *************************************************/
 449
 450 static GstURIType
 451 gst_fd_src_uri_get_type (void)
 452 {
 453   return GST_URI_SRC;
 454 }
 455 static gchar **
 456 gst_fd_src_uri_get_protocols (void)
 457 {
 458   static gchar *protocols[] = { "fd", NULL };
 459
 460   return protocols;
 461 }
 462 static const gchar *
 463 gst_fd_src_uri_get_uri (GstURIHandler * handler)
 464 {
 465   GstFdSrc *src = GST_FD_SRC (handler);
 466
 467   return src->uri;
 468 }
 469
 470 static gboolean
 471 gst_fd_src_uri_set_uri (GstURIHandler * handler, const gchar * uri)
 472 {
 473   gchar *protocol;
 474   GstFdSrc *src = GST_FD_SRC (handler);
 475   gint fd;
 476
 477   protocol = gst_uri_get_protocol (uri);
 478   if (strcmp (protocol, "fd") != 0) {
 479     g_free (protocol);
 480     return FALSE;
 481   }
 482   g_free (protocol);
 483
 484   if (sscanf (uri, "fd://%d", &fd) != 1)
 485     return FALSE;
 486
 487   src->new_fd = fd;
 488
 489   GST_OBJECT_LOCK (src);
 490   if (GST_STATE (GST_ELEMENT (src)) <= GST_STATE_READY) {
 491     gst_fd_src_update_fd (src);
 492   }
 493   GST_OBJECT_UNLOCK (src);
 494
 495   return TRUE;
 496 }
 497
 498 static void
 499 gst_fd_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
 500 {
 501   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 502
 503   iface->get_type = gst_fd_src_uri_get_type;
 504   iface->get_protocols = gst_fd_src_uri_get_protocols;
 505   iface->get_uri = gst_fd_src_uri_get_uri;
 506   iface->set_uri = gst_fd_src_uri_set_uri;
 507 }