libs/gst/bytestream/bytestream.c

   1 /* GStreamer
   2  * Copyright (C) 2001 Erik Walthinsen <omega@temple-baptist.com>
   3  *
   4  * gstbytestream.c: adds a convenient bytestream based API to a pad.
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  */
  21
  22 #include <stdio.h>
  23 #include <string.h>
  24 #include <stdlib.h>
  25
  26 #include <gst/gstinfo.h>
  27 #include "bytestream.h"
  28
  29 /* #define BS_DEBUG */
  30
  31 #ifdef BS_DEBUG
  32 # define bs_print(format,args...)       GST_DEBUG (GST_CAT_BUFFER,  format, ## args)
  33 # define bs_status(bs)                  gst_bytestream_print_status(bs)
  34 #else
  35 # define bs_print(format,args...)
  36 # define bs_status(bs)
  37 #endif
  38
  39 guint8 *gst_bytestream_assemble (GstByteStream * bs, guint32 len);
  40
  41 /**
  42  * gst_bytestream_new:
  43  * @pad: the pad to attach the bytestream to
  44  *
  45  * creates a bytestream from the given pad
  46  *
  47  * Returns: a new #GstByteStream object
  48  */
  49 GstByteStream *
  50 gst_bytestream_new (GstPad * pad)
  51 {
  52   GstByteStream *bs = g_new (GstByteStream, 1);
  53
  54   bs->pad = pad;
  55   bs->event = NULL;
  56   bs->buflist = NULL;
  57   bs->headbufavail = 0;
  58   bs->listavail = 0;
  59   bs->assembled = NULL;
  60
  61   return bs;
  62 }
  63
  64 void
  65 gst_bytestream_destroy (GstByteStream * bs)
  66 {
  67   GSList *walk;
  68
  69   if (bs->event)
  70     gst_event_free (bs->event);
  71
  72   walk = bs->buflist;
  73   while (walk) {
  74     gst_buffer_unref (GST_BUFFER (walk->data));
  75     walk = g_slist_next (walk);
  76   }
  77   g_slist_free (bs->buflist);
  78   if (bs->assembled)
  79     g_free (bs->assembled);
  80   g_free (bs);
  81 }
  82
  83 /* HOW THIS WORKS:
  84  *
  85  * The fundamental structure is a singly-linked list of buffers.  The
  86  * buffer on the front is the oldest, and thus the first to read data
  87  * from.  The number of bytes left to be read in this buffer is stored
  88  * in bs->headbufavail.  The number of bytes available in the entire
  89  * list (including the head buffer) is in bs->listavail.
  90  *
  91  * When a request is made for data (peek), _fill_bytes is called with
  92  * the number of bytes needed, but only if the listavail indicates
  93  * that there aren't already enough.  This calls _get_next_buf until
  94  * the listavail is sufficient to satisfy the demand.
  95  *
  96  * _get_next_buf pulls a buffer from the pad the bytestream is attached
  97  * to, and shoves it in the list.  There are actually two things it can
  98  * do.  If there's already a buffer in the list, and the _is_span_fast()
  99  * test returns true, it will merge it with that last buffer.  Otherwise
 100  * it will simply tack it onto the end of the list.
 101  *
 102  * The _peek itself first checks the simple case of the request fitting
 103  * within the head buffer, and if so creates a subbuffer and returns.
 104  * Otherwise, it creates a new buffer and allocates space for the request
 105  * and calls _assemble to fill it.  We know we have to copy because this
 106  * case only happens when the _merge wasn't feasible during _get_next_buf.
 107  *
 108  * The _flush method repeatedly inspects the head buffer and flushes as
 109  * much data from it as it needs to, up to the size of the buffer.  If
 110  * the flush decimates the buffer, it's stripped, unref'd, and removed.
 111  */
 112
 113
 114 /* get the next buffer
 115  * if the buffer can be merged with the head buffer, do so
 116  * else add it onto the head of the
 117  */
 118 static gboolean
 119 gst_bytestream_get_next_buf (GstByteStream * bs)
 120 {
 121   GstBuffer *nextbuf, *lastbuf;
 122   GSList *end;
 123
 124   g_assert (!bs->event);
 125
 126   bs_print ("get_next_buf: pulling buffer\n");
 127   nextbuf = gst_pad_pull (bs->pad);
 128
 129   if (GST_IS_EVENT (nextbuf))
 130     {
 131       bs->event = GST_EVENT (nextbuf);
 132       return FALSE;
 133     }
 134
 135   if (!nextbuf)
 136     return FALSE;
 137
 138   bs_print ("get_next_buf: got buffer of %d bytes\n", GST_BUFFER_SIZE (nextbuf));
 139
 140   /* first see if there are any buffers in the list at all */
 141   if (bs->buflist) {
 142     bs_print ("gst_next_buf: there is at least one buffer in the list\n");
 143     /* now find the end of the list */
 144     end = g_slist_last (bs->buflist);
 145     /* get the buffer that's there */
 146     lastbuf = GST_BUFFER (end->data);
 147
 148     /* see if we can marge cheaply */
 149     if (gst_buffer_is_span_fast (lastbuf, nextbuf)) {
 150       bs_print ("get_next_buf: merging new buffer with last buf on list\n");
 151       /* it is, let's merge them (this is really an append, but...) */
 152       end->data = gst_buffer_merge (lastbuf, nextbuf);
 153       /* add to the length of the list */
 154       bs->listavail += GST_BUFFER_SIZE (nextbuf);
 155
 156       /* have to check to see if we merged with the head buffer */
 157       if (end == bs->buflist) {
 158         bs->headbufavail += GST_BUFFER_SIZE (nextbuf);
 159       }
 160
 161       gst_buffer_unref (lastbuf);
 162       gst_buffer_unref (nextbuf);
 163
 164       /* if we can't, we just append this buffer */
 165     }
 166     else {
 167       bs_print ("get_next_buf: adding new buffer to the end of the list\n");
 168       end = g_slist_append (end, nextbuf);
 169       /* also need to increment length of list and buffer count */
 170       bs->listavail += GST_BUFFER_SIZE (nextbuf);
 171     }
 172
 173     /* if there are no buffers in the list */
 174   }
 175   else {
 176     bs_print ("get_next_buf: buflist is empty, adding new buffer to list\n");
 177     /* put this on the end of the list */
 178     bs->buflist = g_slist_append (bs->buflist, nextbuf);
 179     /* and increment the number of bytes in the list */
 180     bs->listavail = GST_BUFFER_SIZE (nextbuf);
 181     /* set the head buffer avail to the size */
 182     bs->headbufavail = GST_BUFFER_SIZE (nextbuf);
 183   }
 184
 185   return TRUE;
 186 }
 187
 188 static gboolean
 189 gst_bytestream_fill_bytes (GstByteStream * bs, guint32 len)
 190 {
 191   /* as long as we don't have enough, we get more buffers */
 192   while (bs->listavail < len) {
 193     bs_print ("fill_bytes: there are %d bytes in the list, we need %d\n", bs->listavail, len);
 194     if (!gst_bytestream_get_next_buf (bs))
 195       return FALSE;
 196   }
 197
 198   return TRUE;
 199 }
 200
 201
 202 GstBuffer *
 203 gst_bytestream_peek (GstByteStream * bs, guint32 len)
 204 {
 205   GstBuffer *headbuf, *retbuf = NULL;
 206
 207   g_return_val_if_fail (bs != NULL, NULL);
 208   g_return_val_if_fail (len > 0, NULL);
 209
 210   bs_print ("peek: asking for %d bytes\n", len);
 211
 212   /* make sure we have enough */
 213   bs_print ("peek: there are %d bytes in the list\n", bs->listavail);
 214   if (len > bs->listavail) {
 215     if (!gst_bytestream_fill_bytes (bs, len))
 216       return NULL;
 217     bs_print ("peek: there are now %d bytes in the list\n", bs->listavail);
 218   }
 219   bs_status (bs);
 220
 221   /* extract the head buffer */
 222   headbuf = GST_BUFFER (bs->buflist->data);
 223
 224   /* if the requested bytes are in the current buffer */
 225   bs_print ("peek: headbufavail is %d\n", bs->headbufavail);
 226   if (len <= bs->headbufavail) {
 227     bs_print ("peek: there are enough bytes in headbuf (need %d, have %d)\n", len, bs->headbufavail);
 228     /* create a sub-buffer of the headbuf */
 229     retbuf = gst_buffer_create_sub (headbuf, GST_BUFFER_SIZE (headbuf) - bs->headbufavail, len);
 230
 231     /* otherwise we need to figure out how to assemble one */
 232   }
 233   else {
 234     bs_print ("peek: current buffer is not big enough for len %d\n", len);
 235
 236     retbuf = gst_buffer_new ();
 237     GST_BUFFER_SIZE (retbuf) = len;
 238     GST_BUFFER_DATA (retbuf) = gst_bytestream_assemble (bs, len);
 239     if (GST_BUFFER_OFFSET (headbuf) != -1)
 240       GST_BUFFER_OFFSET (retbuf) = GST_BUFFER_OFFSET (headbuf) + (GST_BUFFER_SIZE (headbuf) - bs->headbufavail);
 241   }
 242
 243   return retbuf;
 244 }
 245
 246 guint8 *
 247 gst_bytestream_peek_bytes (GstByteStream * bs, guint32 len)
 248 {
 249   GstBuffer *headbuf;
 250   guint8 *data = NULL;
 251
 252   g_return_val_if_fail (bs != NULL, NULL);
 253   g_return_val_if_fail (len > 0, NULL);
 254
 255   bs_print ("peek_bytes: asking for %d bytes\n", len);
 256   if (bs->assembled) {
 257     g_free (bs->assembled);
 258     bs->assembled = NULL;
 259   }
 260
 261   /* make sure we have enough */
 262   bs_print ("peek_bytes: there are %d bytes in the list\n", bs->listavail);
 263   if (len > bs->listavail) {
 264     if (!gst_bytestream_fill_bytes (bs, len))
 265       return NULL;
 266     bs_print ("peek_bytes: there are now %d bytes in the list\n", bs->listavail);
 267   }
 268   bs_status (bs);
 269
 270   /* extract the head buffer */
 271   headbuf = GST_BUFFER (bs->buflist->data);
 272
 273   /* if the requested bytes are in the current buffer */
 274   bs_print ("peek_bytes: headbufavail is %d\n", bs->headbufavail);
 275   if (len <= bs->headbufavail) {
 276     bs_print ("peek_bytes: there are enough bytes in headbuf (need %d, have %d)\n", len, bs->headbufavail);
 277     /* create a sub-buffer of the headbuf */
 278     data = GST_BUFFER_DATA (headbuf) + (GST_BUFFER_SIZE (headbuf) - bs->headbufavail);
 279
 280     /* otherwise we need to figure out how to assemble one */
 281   }
 282   else {
 283     bs_print ("peek_bytes: current buffer is not big enough for len %d\n", len);
 284
 285     data = gst_bytestream_assemble (bs, len);
 286     bs->assembled = data;
 287     bs->assembled_len = len;
 288   }
 289
 290   return data;
 291 }
 292
 293 guint8 *
 294 gst_bytestream_assemble (GstByteStream * bs, guint32 len)
 295 {
 296   guint8 *data = g_malloc (len);
 297   GSList *walk;
 298   guint32 copied = 0;
 299   GstBuffer *buf;
 300
 301   /* copy the data from the curbuf */
 302   buf = GST_BUFFER (bs->buflist->data);
 303   bs_print ("assemble: copying %d bytes from curbuf at %d to *data\n", bs->headbufavail,
 304             GST_BUFFER_SIZE (buf) - bs->headbufavail);
 305   memcpy (data, GST_BUFFER_DATA (buf) + GST_BUFFER_SIZE (buf) - bs->headbufavail, bs->headbufavail);
 306   copied += bs->headbufavail;
 307
 308   /* asumption is made that the buffers all exist in the list */
 309   walk = g_slist_next (bs->buflist);
 310   while (copied < len) {
 311     buf = GST_BUFFER (walk->data);
 312     if (GST_BUFFER_SIZE (buf) < (len - copied)) {
 313       bs_print ("assemble: copying %d bytes from buf to output offset %d\n", GST_BUFFER_SIZE (buf), copied);
 314       memcpy (data + copied, GST_BUFFER_DATA (buf), GST_BUFFER_SIZE (buf));
 315       copied += GST_BUFFER_SIZE (buf);
 316     }
 317     else {
 318       bs_print ("assemble: copying %d bytes from buf to output offset %d\n", len - copied, copied);
 319       memcpy (data + copied, GST_BUFFER_DATA (buf), len - copied);
 320       copied = len;
 321     }
 322     walk = g_slist_next (walk);
 323   }
 324
 325   return data;
 326 }
 327
 328 gboolean
 329 gst_bytestream_flush (GstByteStream * bs, guint32 len)
 330 {
 331   bs_print ("flush: flushing %d bytes\n", len);
 332
 333   /* make sure we have enough */
 334   bs_print ("flush: there are %d bytes in the list\n", bs->listavail);
 335   if (len > bs->listavail) {
 336     if (!gst_bytestream_fill_bytes (bs, len))
 337       return FALSE;
 338     bs_print ("flush: there are now %d bytes in the list\n", bs->listavail);
 339   }
 340
 341   gst_bytestream_flush_fast (bs, len);
 342
 343   return TRUE;
 344 }
 345
 346 void
 347 gst_bytestream_flush_fast (GstByteStream * bs, guint32 len)
 348 {
 349   GstBuffer *headbuf;
 350
 351   g_assert (len <= bs->listavail);
 352
 353   if (bs->assembled) {
 354     g_free (bs->assembled);
 355     bs->assembled = NULL;
 356   }
 357
 358   /* repeat until we've flushed enough data */
 359   while (len > 0) {
 360     headbuf = GST_BUFFER (bs->buflist->data);
 361
 362     bs_print ("flush: analyzing buffer that's %d bytes long, offset %d\n", GST_BUFFER_SIZE (headbuf),
 363               GST_BUFFER_OFFSET (headbuf));
 364
 365     /* if there's enough to complete the flush */
 366     if (bs->headbufavail > len) {
 367       /* just trim it off */
 368       bs_print ("flush: trimming %d bytes off end of headbuf\n", len);
 369       bs->headbufavail -= len;
 370       bs->listavail -= len;
 371       len = 0;
 372
 373       /* otherwise we have to trim the whole buffer */
 374     }
 375     else {
 376       bs_print ("flush: removing head buffer completely\n");
 377       /* remove it from the list */
 378       bs->buflist = g_slist_delete_link (bs->buflist, bs->buflist);
 379       /* trim it from the avail size */
 380       bs->listavail -= bs->headbufavail;
 381       /* record that we've trimmed this many bytes */
 382       len -= bs->headbufavail;
 383       /* unref it */
 384       gst_buffer_unref (headbuf);
 385
 386       /* record the new headbufavail */
 387       if (bs->buflist) {
 388         bs->headbufavail = GST_BUFFER_SIZE (GST_BUFFER (bs->buflist->data));
 389         bs_print ("flush: next headbuf is %d bytes\n", bs->headbufavail);
 390       }
 391       else {
 392         bs_print ("flush: no more bytes at all\n");
 393       }
 394     }
 395
 396     bs_print ("flush: bottom of while(), len is now %d\n", len);
 397   }
 398 }
 399
 400 gboolean
 401 gst_bytestream_seek (GstByteStream *bs, GstSeekType type, gint64 offset)
 402 {
 403   GstRealPad *peer = GST_RPAD_PEER (bs->pad);
 404
 405   if (gst_pad_send_event (GST_PAD (peer), gst_event_new_seek (type, offset, TRUE))) {
 406     GstBuffer *nextbuf;
 407
 408     gst_bytestream_flush_fast (bs, bs->listavail);
 409
 410     do {
 411       nextbuf = gst_pad_pull (bs->pad);
 412     }
 413     while (!GST_IS_EVENT (nextbuf));
 414
 415     return TRUE;
 416   }
 417   return FALSE;
 418 }
 419
 420 guint64
 421 gst_bytestream_tell (GstByteStream *bs)
 422 {
 423   return 0;
 424 }
 425
 426 GstBuffer *
 427 gst_bytestream_read (GstByteStream * bs, guint32 len)
 428 {
 429   GstBuffer *buf = gst_bytestream_peek (bs, len);
 430   if (!buf)
 431     return NULL;
 432
 433   gst_bytestream_flush_fast (bs, len);
 434
 435   return buf;
 436 }
 437
 438
 439 /**
 440  * gst_bytestream_get_status
 441  * @bs: a bytestream
 442  * @avail_out: total number of bytes buffered
 443  * @event_out: an event
 444  *
 445  * When an event occurs, the bytestream will return NULL.  You must
 446  * retrieve the event using this API before reading more bytes from
 447  * the stream.
 448  *
 449  * It is possible for the bytestream to return NULL due to running
 450  * out of buffers, however, this indicates a bug because an EOS
 451  * event should have been sent.
 452  */
 453 void
 454 gst_bytestream_get_status (GstByteStream *bs,
 455                            guint32 *avail_out,
 456                            GstEvent **event_out)
 457 {
 458   if (avail_out)
 459     *avail_out = bs->listavail;
 460
 461   if (event_out)
 462     {
 463       *event_out = bs->event;
 464       bs->event = NULL;
 465     }
 466 }
 467
 468 void
 469 gst_bytestream_print_status (GstByteStream * bs)
 470 {
 471   GSList *walk;
 472   GstBuffer *buf;
 473
 474   bs_print ("STATUS: head buffer has %d bytes available\n", bs->headbufavail);
 475   bs_print ("STATUS: list has %d bytes available\n", bs->listavail);
 476   walk = bs->buflist;
 477   while (walk) {
 478     buf = GST_BUFFER (walk->data);
 479     walk = g_slist_next (walk);
 480
 481     bs_print ("STATUS: buffer starts at %d and is %d bytes long\n", GST_BUFFER_OFFSET (buf), GST_BUFFER_SIZE (buf));
 482   }
 483 }
 484
 485
 486 static gboolean
 487 plugin_init (GModule *module, GstPlugin *plugin)
 488 {
 489   gst_plugin_set_longname (plugin, "GstByteStream: a byte-oriented layer on top of buffer-passing");
 490   return TRUE;
 491 }
 492
 493 GstPluginDesc plugin_desc = {
 494   GST_VERSION_MAJOR,
 495   GST_VERSION_MINOR,
 496   "gstbytestream",
 497   plugin_init
 498 };