gst/rtmp2/rtmp/rtmpchunkstream.c

   1 /* GStreamer RTMP Library
   2  * Copyright (C) 2017 Make.TV, Inc. <info@make.tv>
   3  *   Contact: Jan Alexander Steffens (heftig) <jsteffens@make.tv>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
  18  * Boston, MA 02110-1335, USA.
  19  */
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include "rtmpchunkstream.h"
  26 #include "rtmputils.h"
  27
  28 GST_DEBUG_CATEGORY_STATIC (gst_rtmp_chunk_stream_debug_category);
  29 #define GST_CAT_DEFAULT gst_rtmp_chunk_stream_debug_category
  30
  31 static void
  32 init_debug (void)
  33 {
  34   static volatile gsize done = 0;
  35   if (g_once_init_enter (&done)) {
  36     GST_DEBUG_CATEGORY_INIT (gst_rtmp_chunk_stream_debug_category,
  37         "rtmpchunkstream", 0, "debug category for rtmp chunk streams");
  38     g_once_init_leave (&done, 1);
  39   }
  40 }
  41
  42 enum
  43 {
  44   CHUNK_BYTE_TWOBYTE = 0,
  45   CHUNK_BYTE_THREEBYTE = 1,
  46   CHUNK_BYTE_MASK = 0x3f,
  47   CHUNK_STREAM_MIN_TWOBYTE = 0x40,
  48   CHUNK_STREAM_MIN_THREEBYTE = 0x140,
  49   CHUNK_STREAM_MAX_THREEBYTE = 0x1003f,
  50 };
  51
  52 typedef enum
  53 {
  54   CHUNK_TYPE_0 = 0,
  55   CHUNK_TYPE_1 = 1,
  56   CHUNK_TYPE_2 = 2,
  57   CHUNK_TYPE_3 = 3,
  58 } ChunkType;
  59
  60 static const gsize chunk_header_sizes[4] = { 11, 7, 3, 0 };
  61
  62 struct _GstRtmpChunkStream
  63 {
  64   GstBuffer *buffer;
  65   GstRtmpMeta *meta;
  66   GstMapInfo map;               /* Only used for parsing */
  67   guint32 id;
  68   guint32 offset;
  69   guint64 bytes;
  70 };
  71
  72 struct _GstRtmpChunkStreams
  73 {
  74   GArray *array;
  75 };
  76
  77 static inline gboolean
  78 chunk_stream_is_open (GstRtmpChunkStream * cstream)
  79 {
  80   return cstream->map.data != NULL;
  81 }
  82
  83 static void
  84 chunk_stream_take_buffer (GstRtmpChunkStream * cstream, GstBuffer * buffer)
  85 {
  86   GstRtmpMeta *meta = gst_buffer_get_rtmp_meta (buffer);
  87   g_assert (meta);
  88   g_assert (cstream->buffer == NULL);
  89   cstream->buffer = buffer;
  90   cstream->meta = meta;
  91 }
  92
  93 static void
  94 chunk_stream_clear (GstRtmpChunkStream * cstream)
  95 {
  96   if (chunk_stream_is_open (cstream)) {
  97     gst_buffer_unmap (cstream->buffer, &cstream->map);
  98     cstream->map.data = NULL;
  99   }
 100
 101   gst_buffer_replace (&cstream->buffer, NULL);
 102   cstream->meta = NULL;
 103   cstream->offset = 0;
 104 }
 105
 106 static guint32
 107 chunk_stream_next_size (GstRtmpChunkStream * cstream, guint32 chunk_size)
 108 {
 109   guint32 size, offset;
 110
 111   size = cstream->meta->size;
 112   offset = cstream->offset;
 113
 114   g_return_val_if_fail (offset <= size, 0);
 115   return MIN (size - offset, chunk_size);
 116 }
 117
 118 static inline gboolean
 119 needs_ext_ts (GstRtmpMeta * meta)
 120 {
 121   return meta->ts_delta >= 0xffffff;
 122 }
 123
 124
 125 static guint32
 126 dts_to_abs_ts (GstBuffer * buffer)
 127 {
 128   GstClockTime dts = GST_BUFFER_DTS (buffer);
 129   guint32 ret = 0;
 130
 131   if (GST_CLOCK_TIME_IS_VALID (dts)) {
 132     ret = gst_util_uint64_scale_round (dts, 1, GST_MSECOND);
 133   }
 134
 135   GST_TRACE ("Converted DTS %" GST_TIME_FORMAT " into abs TS %"
 136       G_GUINT32_FORMAT " ms", GST_TIME_ARGS (dts), ret);
 137   return ret;
 138 }
 139
 140 static gboolean
 141 dts_diff_to_delta_ts (GstBuffer * old_buffer, GstBuffer * buffer,
 142     guint32 * out_ts)
 143 {
 144   GstClockTime dts = GST_BUFFER_DTS (buffer),
 145       old_dts = GST_BUFFER_DTS (old_buffer);
 146   guint32 abs_ts, old_abs_ts, delta_32 = 0;
 147
 148   if (!GST_CLOCK_TIME_IS_VALID (dts) || !GST_CLOCK_TIME_IS_VALID (old_dts)) {
 149     GST_LOG ("Timestamps not valid; using delta TS 0");
 150     goto out;
 151   }
 152
 153   if (ABS (GST_CLOCK_DIFF (old_dts, dts)) > GST_MSECOND * G_MAXINT32) {
 154     GST_WARNING ("Timestamp delta too large: %" GST_TIME_FORMAT " -> %"
 155         GST_TIME_FORMAT, GST_TIME_ARGS (old_dts), GST_TIME_ARGS (dts));
 156     return FALSE;
 157   }
 158
 159   abs_ts = gst_util_uint64_scale_round (dts, 1, GST_MSECOND);
 160   old_abs_ts = gst_util_uint64_scale_round (old_dts, 1, GST_MSECOND);
 161
 162   /* underflow wraps around */
 163   delta_32 = abs_ts - old_abs_ts;
 164
 165   GST_TRACE ("Converted DTS %" GST_TIME_FORMAT " (%" G_GUINT32_FORMAT
 166       " ms) -> %" GST_TIME_FORMAT " (%" G_GUINT32_FORMAT " ms) into delta TS %"
 167       G_GUINT32_FORMAT " ms", GST_TIME_ARGS (old_dts), old_abs_ts,
 168       GST_TIME_ARGS (dts), abs_ts, delta_32);
 169
 170 out:
 171   *out_ts = delta_32;
 172   return TRUE;
 173 }
 174
 175 static ChunkType
 176 select_chunk_type (GstRtmpChunkStream * cstream, GstBuffer * buffer)
 177 {
 178   GstBuffer *old_buffer = cstream->buffer;
 179   GstRtmpMeta *meta, *old_meta;
 180
 181   g_return_val_if_fail (buffer, -1);
 182
 183   meta = gst_buffer_get_rtmp_meta (buffer);
 184
 185   g_return_val_if_fail (meta, -1);
 186   g_return_val_if_fail (gst_rtmp_message_type_is_valid (meta->type), -1);
 187
 188   meta->size = gst_buffer_get_size (buffer);
 189   meta->cstream = cstream->id;
 190
 191   if (!old_buffer) {
 192     GST_TRACE ("Picking header 0: no previous header");
 193     meta->ts_delta = dts_to_abs_ts (buffer);
 194     return CHUNK_TYPE_0;
 195   }
 196
 197   old_meta = gst_buffer_get_rtmp_meta (old_buffer);
 198
 199   if (old_meta->mstream != meta->mstream) {
 200     GST_TRACE ("Picking header 0: stream mismatch; "
 201         "want %" G_GUINT32_FORMAT " got %" G_GUINT32_FORMAT,
 202         old_meta->mstream, meta->mstream);
 203     meta->ts_delta = dts_to_abs_ts (buffer);
 204     return CHUNK_TYPE_0;
 205   }
 206
 207   if (!dts_diff_to_delta_ts (old_buffer, buffer, &meta->ts_delta)) {
 208     GST_TRACE ("Picking header 0: timestamp delta overflow");
 209     meta->ts_delta = dts_to_abs_ts (buffer);
 210     return CHUNK_TYPE_0;
 211   }
 212
 213   /* now at least type 1 */
 214
 215   if (old_meta->type != meta->type) {
 216     GST_TRACE ("Picking header 1: type mismatch; want %d got %d",
 217         old_meta->type, meta->type);
 218     return CHUNK_TYPE_1;
 219   }
 220
 221   if (old_meta->size != meta->size) {
 222     GST_TRACE ("Picking header 1: size mismatch; "
 223         "want %" G_GUINT32_FORMAT " got %" G_GUINT32_FORMAT,
 224         old_meta->size, meta->size);
 225     return CHUNK_TYPE_1;
 226   }
 227
 228   /* now at least type 2 */
 229
 230   if (old_meta->ts_delta != meta->ts_delta) {
 231     GST_TRACE ("Picking header 2: timestamp delta mismatch; "
 232         "want %" G_GUINT32_FORMAT " got %" G_GUINT32_FORMAT,
 233         old_meta->ts_delta, meta->ts_delta);
 234     return CHUNK_TYPE_2;
 235   }
 236
 237   /* now at least type 3 */
 238
 239   GST_TRACE ("Picking header 3");
 240   return CHUNK_TYPE_3;
 241 }
 242
 243 static GstBuffer *
 244 serialize_next (GstRtmpChunkStream * cstream, guint32 chunk_size,
 245     ChunkType type)
 246 {
 247   GstRtmpMeta *meta = cstream->meta;
 248   guint8 small_stream_id;
 249   gsize header_size = chunk_header_sizes[type], offset;
 250   gboolean ext_ts;
 251   GstBuffer *ret;
 252   GstMapInfo map;
 253
 254   GST_TRACE ("Serializing a chunk of type %d, offset %" G_GUINT32_FORMAT,
 255       type, cstream->offset);
 256
 257   if (cstream->id < CHUNK_STREAM_MIN_TWOBYTE) {
 258     small_stream_id = cstream->id;
 259     header_size += 1;
 260   } else if (cstream->id < CHUNK_STREAM_MIN_THREEBYTE) {
 261     small_stream_id = CHUNK_BYTE_TWOBYTE;
 262     header_size += 2;
 263   } else {
 264     small_stream_id = CHUNK_BYTE_THREEBYTE;
 265     header_size += 3;
 266   }
 267
 268   ext_ts = needs_ext_ts (meta);
 269   if (ext_ts) {
 270     header_size += 4;
 271   }
 272
 273   GST_TRACE ("Allocating buffer, header size %" G_GSIZE_FORMAT, header_size);
 274
 275   ret = gst_buffer_new_allocate (NULL, header_size, NULL);
 276   if (!ret) {
 277     GST_ERROR ("Failed to allocate chunk buffer");
 278     return NULL;
 279   }
 280
 281   if (!gst_buffer_map (ret, &map, GST_MAP_WRITE)) {
 282     GST_ERROR ("Failed to map %" GST_PTR_FORMAT, ret);
 283     gst_buffer_unref (ret);
 284     return NULL;
 285   }
 286
 287   /* Chunk Basic Header */
 288   GST_WRITE_UINT8 (map.data, (type << 6) | small_stream_id);
 289   offset = 1;
 290
 291   switch (small_stream_id) {
 292     case CHUNK_BYTE_TWOBYTE:
 293       GST_WRITE_UINT8 (map.data + 1, cstream->id - CHUNK_STREAM_MIN_TWOBYTE);
 294       offset += 1;
 295       break;
 296
 297     case CHUNK_BYTE_THREEBYTE:
 298       GST_WRITE_UINT16_LE (map.data + 1,
 299           cstream->id - CHUNK_STREAM_MIN_TWOBYTE);
 300       offset += 2;
 301       break;
 302   }
 303
 304   switch (type) {
 305     case CHUNK_TYPE_0:
 306       /* SRSLY:  "Message stream ID is stored in little-endian format." */
 307       GST_WRITE_UINT32_LE (map.data + offset + 7, meta->mstream);
 308       /* no break */
 309     case CHUNK_TYPE_1:
 310       GST_WRITE_UINT24_BE (map.data + offset + 3, meta->size);
 311       GST_WRITE_UINT8 (map.data + offset + 6, meta->type);
 312       /* no break */
 313     case CHUNK_TYPE_2:
 314       GST_WRITE_UINT24_BE (map.data + offset,
 315           ext_ts ? 0xffffff : meta->ts_delta);
 316       /* no break */
 317     case CHUNK_TYPE_3:
 318       offset += chunk_header_sizes[type];
 319
 320       if (ext_ts) {
 321         GST_WRITE_UINT32_BE (map.data + offset, meta->ts_delta);
 322         offset += 4;
 323       }
 324   }
 325
 326   g_assert (offset == header_size);
 327   GST_MEMDUMP (">>> chunk header", map.data, offset);
 328
 329   gst_buffer_unmap (ret, &map);
 330
 331   GST_BUFFER_OFFSET (ret) = GST_BUFFER_OFFSET_IS_VALID (cstream->buffer) ?
 332       GST_BUFFER_OFFSET (cstream->buffer) + cstream->offset : cstream->bytes;
 333   GST_BUFFER_OFFSET_END (ret) = GST_BUFFER_OFFSET (ret);
 334
 335   if (meta->size > 0) {
 336     guint32 payload_size = chunk_stream_next_size (cstream, chunk_size);
 337
 338     GST_TRACE ("Appending %" G_GUINT32_FORMAT " bytes of payload",
 339         payload_size);
 340
 341     gst_buffer_copy_into (ret, cstream->buffer, GST_BUFFER_COPY_MEMORY,
 342         cstream->offset, payload_size);
 343
 344     GST_BUFFER_OFFSET_END (ret) += payload_size;
 345     cstream->offset += payload_size;
 346     cstream->bytes += payload_size;
 347   } else {
 348     GST_TRACE ("Chunk has no payload");
 349   }
 350
 351   gst_rtmp_buffer_dump (ret, ">>> chunk");
 352
 353   return ret;
 354 }
 355
 356 void
 357 gst_rtmp_chunk_stream_clear (GstRtmpChunkStream * cstream)
 358 {
 359   g_return_if_fail (cstream);
 360   GST_LOG ("Clearing chunk stream %" G_GUINT32_FORMAT, cstream->id);
 361   chunk_stream_clear (cstream);
 362 }
 363
 364 guint32
 365 gst_rtmp_chunk_stream_parse_id (const guint8 * data, gsize size)
 366 {
 367   guint32 ret;
 368
 369   if (size < 1) {
 370     GST_TRACE ("Not enough bytes to read ID");
 371     return 0;
 372   }
 373
 374   ret = GST_READ_UINT8 (data) & CHUNK_BYTE_MASK;
 375
 376   switch (ret) {
 377     case CHUNK_BYTE_TWOBYTE:
 378       if (size < 2) {
 379         GST_TRACE ("Not enough bytes to read two-byte ID");
 380         return 0;
 381       }
 382
 383       ret = GST_READ_UINT8 (data + 1) + CHUNK_STREAM_MIN_TWOBYTE;
 384       break;
 385
 386     case CHUNK_BYTE_THREEBYTE:
 387       if (size < 3) {
 388         GST_TRACE ("Not enough bytes to read three-byte ID");
 389         return 0;
 390       }
 391
 392       ret = GST_READ_UINT16_LE (data + 1) + CHUNK_STREAM_MIN_TWOBYTE;
 393       break;
 394   }
 395
 396   GST_TRACE ("Parsed chunk stream ID %" G_GUINT32_FORMAT, ret);
 397   return ret;
 398 }
 399
 400 guint32
 401 gst_rtmp_chunk_stream_parse_header (GstRtmpChunkStream * cstream,
 402     const guint8 * data, gsize size)
 403 {
 404   GstBuffer *buffer;
 405   GstRtmpMeta *meta;
 406   const guint8 *message_header;
 407   guint32 header_size;
 408   ChunkType type;
 409   gboolean has_abs_timestamp = FALSE;
 410
 411   g_return_val_if_fail (cstream, 0);
 412   g_return_val_if_fail (cstream->id == gst_rtmp_chunk_stream_parse_id (data,
 413           size), 0);
 414
 415   type = GST_READ_UINT8 (data) >> 6;
 416   GST_TRACE ("Parsing chunk stream %" G_GUINT32_FORMAT " header type %d",
 417       cstream->id, type);
 418
 419   switch (GST_READ_UINT8 (data) & CHUNK_BYTE_MASK) {
 420     case CHUNK_BYTE_TWOBYTE:
 421       header_size = 2;
 422       break;
 423     case CHUNK_BYTE_THREEBYTE:
 424       header_size = 3;
 425       break;
 426     default:
 427       header_size = 1;
 428       break;
 429   }
 430
 431   message_header = data + header_size;
 432   header_size += chunk_header_sizes[type];
 433
 434   if (cstream->buffer) {
 435     buffer = cstream->buffer;
 436     meta = cstream->meta;
 437     g_assert (meta->cstream == cstream->id);
 438   } else {
 439     buffer = gst_buffer_new ();
 440     GST_BUFFER_DTS (buffer) = 0;
 441     GST_BUFFER_OFFSET (buffer) = cstream->bytes;
 442     GST_BUFFER_FLAG_SET (buffer, GST_BUFFER_FLAG_DISCONT);
 443
 444     meta = gst_buffer_add_rtmp_meta (buffer);
 445     meta->cstream = cstream->id;
 446
 447     chunk_stream_take_buffer (cstream, buffer);
 448     GST_DEBUG ("Starting parse with new %" GST_PTR_FORMAT, buffer);
 449   }
 450
 451   if (size < header_size) {
 452     GST_TRACE ("not enough bytes to read header");
 453     return header_size;
 454   }
 455
 456   switch (type) {
 457     case CHUNK_TYPE_0:
 458       has_abs_timestamp = TRUE;
 459       /* SRSLY:  "Message stream ID is stored in little-endian format." */
 460       meta->mstream = GST_READ_UINT32_LE (message_header + 7);
 461       /* no break */
 462     case CHUNK_TYPE_1:
 463       meta->type = GST_READ_UINT8 (message_header + 6);
 464       meta->size = GST_READ_UINT24_BE (message_header + 3);
 465       /* no break */
 466     case CHUNK_TYPE_2:
 467       meta->ts_delta = GST_READ_UINT24_BE (message_header);
 468       /* no break */
 469     case CHUNK_TYPE_3:
 470       if (needs_ext_ts (meta)) {
 471         guint32 timestamp;
 472
 473         if (size < header_size + 4) {
 474           GST_TRACE ("not enough bytes to read extended timestamp");
 475           return header_size + 4;
 476         }
 477
 478         GST_TRACE ("Reading extended timestamp");
 479         timestamp = GST_READ_UINT32_BE (data + header_size);
 480
 481         if (type == 3 && meta->ts_delta != timestamp) {
 482           GST_WARNING ("Type 3 extended timestamp does not match expected"
 483               " timestamp (want %" G_GUINT32_FORMAT " got %" G_GUINT32_FORMAT
 484               "); assuming it's not present", meta->ts_delta, timestamp);
 485         } else {
 486           meta->ts_delta = timestamp;
 487           header_size += 4;
 488         }
 489       }
 490   }
 491
 492   GST_MEMDUMP ("<<< chunk header", data, header_size);
 493
 494   if (!chunk_stream_is_open (cstream)) {
 495     GstClockTime dts = GST_BUFFER_DTS (buffer);
 496     guint32 delta_32, abs_32;
 497     gint64 delta_64;
 498
 499     if (has_abs_timestamp) {
 500       abs_32 = meta->ts_delta;
 501       delta_32 = abs_32 - dts / GST_MSECOND;
 502     } else {
 503       delta_32 = meta->ts_delta;
 504       abs_32 = delta_32 + dts / GST_MSECOND;
 505     }
 506
 507     GST_TRACE ("Timestamp delta is %" G_GUINT32_FORMAT " (absolute %"
 508         G_GUINT32_FORMAT ")", delta_32, abs_32);
 509
 510     /* emulate signed overflow */
 511     delta_64 = delta_32;
 512     if (delta_64 > G_MAXINT32) {
 513       delta_64 -= G_MAXUINT32;
 514       delta_64 -= 1;
 515     }
 516
 517     delta_64 *= GST_MSECOND;
 518
 519     if (G_LIKELY (delta_64 >= 0)) {
 520       /* Normal advancement */
 521     } else if (G_LIKELY ((guint64) (-delta_64) <= dts)) {
 522       /* In-bounds regression */
 523       GST_WARNING ("Timestamp regression: %" GST_STIME_FORMAT,
 524           GST_STIME_ARGS (delta_64));
 525     } else {
 526       /* Out-of-bounds regression */
 527       GST_WARNING ("Timestamp regression: %" GST_STIME_FORMAT ", offsetting",
 528           GST_STIME_ARGS (delta_64));
 529       delta_64 = delta_32 * GST_MSECOND;
 530     }
 531
 532     GST_BUFFER_DTS (buffer) += delta_64;
 533
 534     GST_TRACE ("Adjusted buffer DTS (%" GST_TIME_FORMAT ") by %"
 535         GST_STIME_FORMAT " to %" GST_TIME_FORMAT, GST_TIME_ARGS (dts),
 536         GST_STIME_ARGS (delta_64), GST_TIME_ARGS (GST_BUFFER_DTS (buffer)));
 537   } else {
 538     GST_TRACE ("Message payload already started; not touching timestamp");
 539   }
 540
 541   return header_size;
 542 }
 543
 544 guint32
 545 gst_rtmp_chunk_stream_parse_payload (GstRtmpChunkStream * cstream,
 546     guint32 chunk_size, guint8 ** data)
 547 {
 548   GstMemory *mem;
 549
 550   g_return_val_if_fail (cstream, 0);
 551   g_return_val_if_fail (cstream->buffer, 0);
 552
 553   if (!chunk_stream_is_open (cstream)) {
 554     guint32 size = cstream->meta->size;
 555
 556     GST_TRACE ("Allocating buffer, payload size %" G_GUINT32_FORMAT, size);
 557
 558     mem = gst_allocator_alloc (NULL, size, 0);
 559     if (!mem) {
 560       GST_ERROR ("Failed to allocate buffer for payload size %"
 561           G_GUINT32_FORMAT, size);
 562       return 0;
 563     }
 564
 565     gst_buffer_append_memory (cstream->buffer, mem);
 566     gst_buffer_map (cstream->buffer, &cstream->map, GST_MAP_WRITE);
 567   }
 568
 569   g_return_val_if_fail (cstream->map.size == cstream->meta->size, 0);
 570
 571   if (data) {
 572     *data = cstream->map.data + cstream->offset;
 573   }
 574
 575   return chunk_stream_next_size (cstream, chunk_size);
 576 }
 577
 578 guint32
 579 gst_rtmp_chunk_stream_wrote_payload (GstRtmpChunkStream * cstream,
 580     guint32 chunk_size)
 581 {
 582   guint32 size;
 583
 584   g_return_val_if_fail (cstream, FALSE);
 585   g_return_val_if_fail (chunk_stream_is_open (cstream), FALSE);
 586
 587   size = chunk_stream_next_size (cstream, chunk_size);
 588   cstream->offset += size;
 589   cstream->bytes += size;
 590
 591   return chunk_stream_next_size (cstream, chunk_size);
 592 }
 593
 594 GstBuffer *
 595 gst_rtmp_chunk_stream_parse_finish (GstRtmpChunkStream * cstream)
 596 {
 597   GstBuffer *buffer, *empty;
 598
 599   g_return_val_if_fail (cstream, NULL);
 600   g_return_val_if_fail (cstream->buffer, NULL);
 601
 602   buffer = gst_buffer_ref (cstream->buffer);
 603   GST_BUFFER_OFFSET_END (buffer) = cstream->bytes;
 604
 605   gst_rtmp_buffer_dump (buffer, "<<< message");
 606
 607   chunk_stream_clear (cstream);
 608
 609   empty = gst_buffer_new ();
 610
 611   if (!gst_buffer_copy_into (empty, buffer, GST_BUFFER_COPY_META, 0, 0)) {
 612     GST_ERROR ("copy_into failed");
 613     return NULL;
 614   }
 615
 616   GST_BUFFER_DTS (empty) = GST_BUFFER_DTS (buffer);
 617   GST_BUFFER_OFFSET (empty) = GST_BUFFER_OFFSET_END (buffer);
 618
 619   chunk_stream_take_buffer (cstream, empty);
 620
 621   return buffer;
 622 }
 623
 624 GstBuffer *
 625 gst_rtmp_chunk_stream_serialize_start (GstRtmpChunkStream * cstream,
 626     GstBuffer * buffer, guint32 chunk_size)
 627 {
 628   ChunkType type;
 629
 630   g_return_val_if_fail (cstream, NULL);
 631   g_return_val_if_fail (GST_IS_BUFFER (buffer), NULL);
 632
 633   type = select_chunk_type (cstream, buffer);
 634   g_return_val_if_fail (type >= 0, NULL);
 635
 636   GST_TRACE ("Starting serialization of message %" GST_PTR_FORMAT
 637       " into stream %" G_GUINT32_FORMAT, buffer, cstream->id);
 638
 639   gst_rtmp_buffer_dump (buffer, ">>> message");
 640
 641   chunk_stream_clear (cstream);
 642   chunk_stream_take_buffer (cstream, buffer);
 643
 644   return serialize_next (cstream, chunk_size, type);
 645 }
 646
 647 GstBuffer *
 648 gst_rtmp_chunk_stream_serialize_next (GstRtmpChunkStream * cstream,
 649     guint32 chunk_size)
 650 {
 651   g_return_val_if_fail (cstream, NULL);
 652   g_return_val_if_fail (cstream->buffer, NULL);
 653
 654   if (chunk_stream_next_size (cstream, chunk_size) == 0) {
 655     GST_TRACE ("Message serialization finished");
 656     return NULL;
 657   }
 658
 659   GST_TRACE ("Continuing serialization of message %" GST_PTR_FORMAT
 660       " into stream %" G_GUINT32_FORMAT, cstream->buffer, cstream->id);
 661
 662   return serialize_next (cstream, chunk_size, CHUNK_TYPE_3);
 663 }
 664
 665 GstRtmpChunkStreams *
 666 gst_rtmp_chunk_streams_new (void)
 667 {
 668   GstRtmpChunkStreams *cstreams;
 669
 670   init_debug ();
 671
 672   cstreams = g_slice_new (GstRtmpChunkStreams);
 673   cstreams->array = g_array_new (FALSE, TRUE, sizeof (GstRtmpChunkStream));
 674   g_array_set_clear_func (cstreams->array,
 675       (GDestroyNotify) gst_rtmp_chunk_stream_clear);
 676   return cstreams;
 677 }
 678
 679 void
 680 gst_rtmp_chunk_streams_free (gpointer ptr)
 681 {
 682   GstRtmpChunkStreams *cstreams = ptr;
 683   g_clear_pointer (&cstreams->array, g_array_unref);
 684   g_slice_free (GstRtmpChunkStreams, cstreams);
 685 }
 686
 687 GstRtmpChunkStream *
 688 gst_rtmp_chunk_streams_get (GstRtmpChunkStreams * cstreams, guint32 id)
 689 {
 690   GArray *array;
 691   GstRtmpChunkStream *entry;
 692   guint i;
 693
 694   g_return_val_if_fail (cstreams, NULL);
 695   g_return_val_if_fail (id > CHUNK_BYTE_THREEBYTE, NULL);
 696   g_return_val_if_fail (id <= CHUNK_STREAM_MAX_THREEBYTE, NULL);
 697
 698   array = cstreams->array;
 699
 700   for (i = 0; i < array->len; i++) {
 701     entry = &g_array_index (array, GstRtmpChunkStream, i);
 702     if (entry->id == id) {
 703       GST_TRACE ("Obtaining chunk stream %" G_GUINT32_FORMAT, id);
 704       return entry;
 705     }
 706   }
 707
 708   GST_DEBUG ("Allocating chunk stream %" G_GUINT32_FORMAT, id);
 709
 710   g_array_set_size (array, i + 1);
 711   entry = &g_array_index (array, GstRtmpChunkStream, i);
 712   entry->id = id;
 713   return entry;
 714 }