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 #ifdef HAVE_CONFIG_H
  25 #  include "config.h"
  26 #endif
  27 #include "gst/gst_private.h"
  28
  29 #include <sys/types.h>
  30 #include <sys/stat.h>
  31 #include <sys/socket.h>
  32 #include <fcntl.h>
  33 #include <stdio.h>
  34 #ifdef HAVE_UNISTD_H
  35 #include <unistd.h>
  36 #endif
  37 #include <stdlib.h>
  38 #include <errno.h>
  39
  40 #include "gstfdsrc.h"
  41
  42 /* the select call is also performed on the control sockets, that way
  43  * we can send special commands to unblock the select call */
  44 #define CONTROL_STOP            'S'     /* stop the select call */
  45 #define CONTROL_SOCKETS(src)   src->control_sock
  46 #define WRITE_SOCKET(src)      src->control_sock[1]
  47 #define READ_SOCKET(src)       src->control_sock[0]
  48
  49 #define SEND_COMMAND(src, command)          \
  50 G_STMT_START {                              \
  51   unsigned char c; c = command;             \
  52   write (WRITE_SOCKET(src), &c, 1);         \
  53 } G_STMT_END
  54
  55 #define READ_COMMAND(src, command, res)        \
  56 G_STMT_START {                                 \
  57   res = read(READ_SOCKET(src), &command, 1);   \
  58 } G_STMT_END
  59
  60 #define DEFAULT_BLOCKSIZE       4096
  61
  62 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  63     GST_PAD_SRC,
  64     GST_PAD_ALWAYS,
  65     GST_STATIC_CAPS_ANY);
  66
  67 GST_DEBUG_CATEGORY_STATIC (gst_fdsrc_debug);
  68 #define GST_CAT_DEFAULT gst_fdsrc_debug
  69
  70 static GstElementDetails gst_fdsrc_details = GST_ELEMENT_DETAILS ("Disk Source",
  71     "Source/File",
  72     "Synchronous read from a file",
  73     "Erik Walthinsen <omega@cse.ogi.edu>");
  74
  75
  76 /* FdSrc signals and args */
  77 enum
  78 {
  79   SIGNAL_TIMEOUT,
  80   LAST_SIGNAL
  81 };
  82
  83 enum
  84 {
  85   ARG_0,
  86   ARG_FD,
  87   ARG_BLOCKSIZE,
  88   ARG_TIMEOUT
  89 };
  90
  91 #define _do_init(bla) \
  92     GST_DEBUG_CATEGORY_INIT (gst_fdsrc_debug, "fdsrc", 0, "fdsrc element");
  93
  94 GST_BOILERPLATE_FULL (GstFdSrc, gst_fdsrc, GstElement, GST_TYPE_PUSH_SRC,
  95     _do_init);
  96
  97 static void gst_fdsrc_set_property (GObject * object, guint prop_id,
  98     const GValue * value, GParamSpec * pspec);
  99 static void gst_fdsrc_get_property (GObject * object, guint prop_id,
 100     GValue * value, GParamSpec * pspec);
 101
 102 static gboolean gst_fdsrc_start (GstBaseSrc * bsrc);
 103 static gboolean gst_fdsrc_stop (GstBaseSrc * bsrc);
 104 static gboolean gst_fdsrc_unlock (GstBaseSrc * bsrc);
 105
 106 static GstFlowReturn gst_fdsrc_create (GstPushSrc * psrc, GstBuffer ** outbuf);
 107
 108 static void
 109 gst_fdsrc_base_init (gpointer g_class)
 110 {
 111   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (g_class);
 112
 113   gst_element_class_add_pad_template (gstelement_class,
 114       gst_static_pad_template_get (&srctemplate));
 115   gst_element_class_set_details (gstelement_class, &gst_fdsrc_details);
 116 }
 117 static void
 118 gst_fdsrc_class_init (GstFdSrcClass * klass)
 119 {
 120   GObjectClass *gobject_class;
 121   GstBaseSrcClass *gstbasesrc_class;
 122   GstElementClass *gstelement_class;
 123   GstPushSrcClass *gstpush_src_class;
 124
 125   gobject_class = G_OBJECT_CLASS (klass);
 126   gstelement_class = GST_ELEMENT_CLASS (klass);
 127   gstbasesrc_class = (GstBaseSrcClass *) klass;
 128   gstpush_src_class = (GstPushSrcClass *) klass;
 129
 130   parent_class = g_type_class_ref (GST_TYPE_PUSH_SRC);
 131
 132   gobject_class->set_property = gst_fdsrc_set_property;
 133   gobject_class->get_property = gst_fdsrc_get_property;
 134
 135   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_FD,
 136       g_param_spec_int ("fd", "fd", "An open file descriptor to read from",
 137           0, G_MAXINT, 0, G_PARAM_READWRITE));
 138
 139   gstbasesrc_class->start = gst_fdsrc_start;
 140   gstbasesrc_class->stop = gst_fdsrc_stop;
 141   gstbasesrc_class->unlock = gst_fdsrc_unlock;
 142
 143   gstpush_src_class->create = gst_fdsrc_create;
 144 }
 145
 146 static void
 147 gst_fdsrc_init (GstFdSrc * fdsrc, GstFdSrcClass * klass)
 148 {
 149   /* TODO set live only if it's actually a live source (check
 150    * for seekable fd) */
 151   gst_base_src_set_live (GST_BASE_SRC (fdsrc), TRUE);
 152
 153   fdsrc->fd = 0;
 154   fdsrc->curoffset = 0;
 155 }
 156
 157
 158 static gboolean
 159 gst_fdsrc_start (GstBaseSrc * bsrc)
 160 {
 161   GstFdSrc *src = GST_FDSRC (bsrc);
 162   gint control_sock[2];
 163
 164   src->curoffset = 0;
 165
 166   if (socketpair (PF_UNIX, SOCK_STREAM, 0, control_sock) < 0)
 167     goto socket_pair;
 168
 169   READ_SOCKET (src) = control_sock[0];
 170   WRITE_SOCKET (src) = control_sock[1];
 171
 172   fcntl (READ_SOCKET (src), F_SETFL, O_NONBLOCK);
 173   fcntl (WRITE_SOCKET (src), F_SETFL, O_NONBLOCK);
 174
 175   return TRUE;
 176
 177   /* ERRORS */
 178 socket_pair:
 179   {
 180     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ_WRITE, (NULL),
 181         GST_ERROR_SYSTEM);
 182     return FALSE;
 183   }
 184 }
 185
 186 static gboolean
 187 gst_fdsrc_stop (GstBaseSrc * bsrc)
 188 {
 189   GstFdSrc *src = GST_FDSRC (bsrc);
 190
 191   close (READ_SOCKET (src));
 192   close (WRITE_SOCKET (src));
 193
 194   return TRUE;
 195 }
 196
 197 static gboolean
 198 gst_fdsrc_unlock (GstBaseSrc * bsrc)
 199 {
 200   GstFdSrc *src = GST_FDSRC (bsrc);
 201
 202   SEND_COMMAND (src, CONTROL_STOP);
 203
 204   return TRUE;
 205 }
 206
 207 static void
 208 gst_fdsrc_set_property (GObject * object, guint prop_id, const GValue * value,
 209     GParamSpec * pspec)
 210 {
 211   GstFdSrc *src;
 212
 213   g_return_if_fail (GST_IS_FDSRC (object));
 214
 215   src = GST_FDSRC (object);
 216
 217   switch (prop_id) {
 218     case ARG_FD:
 219       src->fd = g_value_get_int (value);
 220       break;
 221     default:
 222       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 223       break;
 224   }
 225 }
 226
 227 static void
 228 gst_fdsrc_get_property (GObject * object, guint prop_id, GValue * value,
 229     GParamSpec * pspec)
 230 {
 231   GstFdSrc *src;
 232
 233   g_return_if_fail (GST_IS_FDSRC (object));
 234
 235   src = GST_FDSRC (object);
 236
 237   switch (prop_id) {
 238     case ARG_FD:
 239       g_value_set_int (value, src->fd);
 240       break;
 241     default:
 242       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 243       break;
 244   }
 245 }
 246
 247 static GstFlowReturn
 248 gst_fdsrc_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 249 {
 250   GstFdSrc *src;
 251   GstBuffer *buf;
 252   glong readbytes;
 253   guint blocksize;
 254
 255 #ifndef HAVE_WIN32
 256   fd_set readfds;
 257   gint retval;
 258 #endif
 259
 260   src = GST_FDSRC (psrc);
 261
 262 #ifndef HAVE_WIN32
 263   FD_ZERO (&readfds);
 264   FD_SET (src->fd, &readfds);
 265   FD_SET (READ_SOCKET (src), &readfds);
 266
 267   do {
 268     retval = select (FD_SETSIZE, &readfds, NULL, NULL, NULL);
 269   } while ((retval == -1 && errno == EINTR));
 270
 271   if (retval == -1)
 272     goto select_error;
 273
 274   if (FD_ISSET (READ_SOCKET (src), &readfds)) {
 275     /* read all stop commands */
 276     while (TRUE) {
 277       gchar command;
 278       int res;
 279
 280       READ_COMMAND (src, command, res);
 281       if (res < 0) {
 282         GST_LOG_OBJECT (src, "no more commands");
 283         /* no more commands */
 284         break;
 285       }
 286     }
 287     goto stopped;
 288   }
 289 #endif
 290
 291   blocksize = GST_BASE_SRC (src)->blocksize;
 292
 293   /* create the buffer */
 294   buf = gst_buffer_new_and_alloc (blocksize);
 295
 296   do {
 297     readbytes = read (src->fd, GST_BUFFER_DATA (buf), blocksize);
 298   } while (readbytes == -1 && errno == EINTR);  /* retry if interrupted */
 299
 300   if (readbytes < 0)
 301     goto read_error;
 302
 303   if (readbytes == 0)
 304     goto eos;
 305
 306   GST_BUFFER_OFFSET (buf) = src->curoffset;
 307   GST_BUFFER_SIZE (buf) = readbytes;
 308   GST_BUFFER_TIMESTAMP (buf) = GST_CLOCK_TIME_NONE;
 309   src->curoffset += readbytes;
 310
 311   GST_DEBUG_OBJECT (psrc, "Read buffer of size %u.", readbytes);
 312
 313   /* we're done, return the buffer */
 314   *outbuf = buf;
 315
 316   return GST_FLOW_OK;
 317
 318   /* ERRORS */
 319 #ifndef HAVE_WIN32
 320 select_error:
 321   {
 322     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 323         ("select on file descriptor: %s.", g_strerror (errno)));
 324     GST_DEBUG_OBJECT (psrc, "Error during select");
 325     return GST_FLOW_ERROR;
 326   }
 327 stopped:
 328   {
 329     GST_DEBUG_OBJECT (psrc, "Select stopped");
 330     return GST_FLOW_WRONG_STATE;
 331   }
 332 #endif
 333 eos:
 334   {
 335     GST_DEBUG_OBJECT (psrc, "Read 0 bytes. EOS.");
 336     gst_buffer_unref (buf);
 337     return GST_FLOW_UNEXPECTED;
 338   }
 339 read_error:
 340   {
 341     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 342         ("read on file descriptor: %s.", g_strerror (errno)));
 343     GST_DEBUG_OBJECT (psrc, "Error reading from fd");
 344     gst_buffer_unref (buf);
 345     return GST_FLOW_ERROR;
 346   }
 347 }