src/polypcore/pstream.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of polypaudio.
   5
   6   polypaudio is free software; you can redistribute it and/or modify
   7   it under the terms of the GNU Lesser General Public License as
   8   published by the Free Software Foundation; either version 2.1 of the
   9   License, or (at your option) any later version.
  10
  11   polypaudio is distributed in the hope that it will be useful, but
  12   WITHOUT ANY WARRANTY; without even the implied warranty of
  13   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14   Lesser General Public License for more details.
  15
  16   You should have received a copy of the GNU Lesser General Public
  17   License along with polypaudio; if not, write to the Free Software
  18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <stdio.h>
  27 #include <stdlib.h>
  28 #include <assert.h>
  29 #include <unistd.h>
  30
  31 #ifdef HAVE_NETINET_IN_H
  32 #include <netinet/in.h>
  33 #endif
  34
  35 #include "winsock.h"
  36
  37 #include <polypcore/queue.h>
  38 #include <polypcore/xmalloc.h>
  39 #include <polypcore/log.h>
  40
  41 #include "pstream.h"
  42
  43 enum {
  44     PA_PSTREAM_DESCRIPTOR_LENGTH,
  45     PA_PSTREAM_DESCRIPTOR_CHANNEL,
  46     PA_PSTREAM_DESCRIPTOR_OFFSET_HI,
  47     PA_PSTREAM_DESCRIPTOR_OFFSET_LO,
  48     PA_PSTREAM_DESCRIPTOR_SEEK,
  49     PA_PSTREAM_DESCRIPTOR_MAX
  50 };
  51
  52 typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX];
  53
  54 #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t))
  55 #define FRAME_SIZE_MAX (1024*500) /* half a megabyte */
  56
  57 struct item_info {
  58     enum { PA_PSTREAM_ITEM_PACKET, PA_PSTREAM_ITEM_MEMBLOCK } type;
  59
  60     /* memblock info */
  61     pa_memchunk chunk;
  62     uint32_t channel;
  63     int64_t offset;
  64     pa_seek_mode_t seek_mode;
  65
  66     /* packet info */
  67     pa_packet *packet;
  68 };
  69
  70 struct pa_pstream {
  71     int ref;
  72
  73     pa_mainloop_api *mainloop;
  74     pa_defer_event *defer_event;
  75     pa_iochannel *io;
  76     pa_queue *send_queue;
  77
  78     int dead;
  79     void (*die_callback) (pa_pstream *p, void *userdata);
  80     void *die_callback_userdata;
  81
  82     struct {
  83         struct item_info* current;
  84         pa_pstream_descriptor descriptor;
  85         void *data;
  86         size_t index;
  87     } write;
  88
  89     struct {
  90         pa_memblock *memblock;
  91         pa_packet *packet;
  92         pa_pstream_descriptor descriptor;
  93         void *data;
  94         size_t index;
  95     } read;
  96
  97     void (*recieve_packet_callback) (pa_pstream *p, pa_packet *packet, void *userdata);
  98     void *recieve_packet_callback_userdata;
  99
 100     void (*recieve_memblock_callback) (pa_pstream *p, uint32_t channel, int64_t offset, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata);
 101     void *recieve_memblock_callback_userdata;
 102
 103     void (*drain_callback)(pa_pstream *p, void *userdata);
 104     void *drain_userdata;
 105
 106     pa_memblock_stat *memblock_stat;
 107 };
 108
 109 static int do_write(pa_pstream *p);
 110 static int do_read(pa_pstream *p);
 111
 112 static void do_something(pa_pstream *p) {
 113     assert(p);
 114
 115     p->mainloop->defer_enable(p->defer_event, 0);
 116
 117     pa_pstream_ref(p);
 118
 119     if (!p->dead && pa_iochannel_is_readable(p->io)) {
 120         if (do_read(p) < 0)
 121             goto fail;
 122     } else if (!p->dead && pa_iochannel_is_hungup(p->io))
 123         goto fail;
 124
 125     if (!p->dead && pa_iochannel_is_writable(p->io)) {
 126         if (do_write(p) < 0)
 127             goto fail;
 128     }
 129
 130     pa_pstream_unref(p);
 131     return;
 132
 133 fail:
 134
 135     p->dead = 1;
 136
 137     if (p->die_callback)
 138         p->die_callback(p, p->die_callback_userdata);
 139
 140     pa_pstream_unref(p);
 141 }
 142
 143 static void io_callback(pa_iochannel*io, void *userdata) {
 144     pa_pstream *p = userdata;
 145
 146     assert(p);
 147     assert(p->io == io);
 148
 149     do_something(p);
 150 }
 151
 152 static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
 153     pa_pstream *p = userdata;
 154
 155     assert(p);
 156     assert(p->defer_event == e);
 157     assert(p->mainloop == m);
 158
 159     do_something(p);
 160 }
 161
 162 pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_memblock_stat *s) {
 163     pa_pstream *p;
 164     assert(io);
 165
 166     p = pa_xnew(pa_pstream, 1);
 167
 168     p->ref = 1;
 169     p->io = io;
 170     pa_iochannel_set_callback(io, io_callback, p);
 171
 172     p->dead = 0;
 173     p->die_callback = NULL;
 174     p->die_callback_userdata = NULL;
 175
 176     p->mainloop = m;
 177     p->defer_event = m->defer_new(m, defer_callback, p);
 178     m->defer_enable(p->defer_event, 0);
 179
 180     p->send_queue = pa_queue_new();
 181     assert(p->send_queue);
 182
 183     p->write.current = NULL;
 184     p->write.index = 0;
 185
 186     p->read.memblock = NULL;
 187     p->read.packet = NULL;
 188     p->read.index = 0;
 189
 190     p->recieve_packet_callback = NULL;
 191     p->recieve_packet_callback_userdata = NULL;
 192
 193     p->recieve_memblock_callback = NULL;
 194     p->recieve_memblock_callback_userdata = NULL;
 195
 196     p->drain_callback = NULL;
 197     p->drain_userdata = NULL;
 198
 199     p->memblock_stat = s;
 200
 201     pa_iochannel_socket_set_rcvbuf(io, 1024*8);
 202     pa_iochannel_socket_set_sndbuf(io, 1024*8);
 203
 204     return p;
 205 }
 206
 207 static void item_free(void *item, PA_GCC_UNUSED void *p) {
 208     struct item_info *i = item;
 209     assert(i);
 210
 211     if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) {
 212         assert(i->chunk.memblock);
 213         pa_memblock_unref(i->chunk.memblock);
 214     } else {
 215         assert(i->type == PA_PSTREAM_ITEM_PACKET);
 216         assert(i->packet);
 217         pa_packet_unref(i->packet);
 218     }
 219
 220     pa_xfree(i);
 221 }
 222
 223 static void pstream_free(pa_pstream *p) {
 224     assert(p);
 225
 226     pa_pstream_close(p);
 227
 228     pa_queue_free(p->send_queue, item_free, NULL);
 229
 230     if (p->write.current)
 231         item_free(p->write.current, NULL);
 232
 233     if (p->read.memblock)
 234         pa_memblock_unref(p->read.memblock);
 235
 236     if (p->read.packet)
 237         pa_packet_unref(p->read.packet);
 238
 239     pa_xfree(p);
 240 }
 241
 242 void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet) {
 243     struct item_info *i;
 244     assert(p && packet && p->ref >= 1);
 245
 246     if (p->dead)
 247         return;
 248
 249 /*     pa_log(__FILE__": push-packet %p", packet); */
 250
 251     i = pa_xnew(struct item_info, 1);
 252     i->type = PA_PSTREAM_ITEM_PACKET;
 253     i->packet = pa_packet_ref(packet);
 254
 255     pa_queue_push(p->send_queue, i);
 256     p->mainloop->defer_enable(p->defer_event, 1);
 257 }
 258
 259 void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) {
 260     struct item_info *i;
 261     assert(p && channel != (uint32_t) -1 && chunk && p->ref >= 1);
 262
 263     if (p->dead)
 264         return;
 265
 266 /*     pa_log(__FILE__": push-memblock %p", chunk); */
 267
 268     i = pa_xnew(struct item_info, 1);
 269     i->type = PA_PSTREAM_ITEM_MEMBLOCK;
 270     i->chunk = *chunk;
 271     i->channel = channel;
 272     i->offset = offset;
 273     i->seek_mode = seek_mode;
 274
 275     pa_memblock_ref(i->chunk.memblock);
 276
 277     pa_queue_push(p->send_queue, i);
 278     p->mainloop->defer_enable(p->defer_event, 1);
 279 }
 280
 281 void pa_pstream_set_recieve_packet_callback(pa_pstream *p, void (*callback) (pa_pstream *p, pa_packet *packet, void *userdata), void *userdata) {
 282     assert(p && callback);
 283
 284     p->recieve_packet_callback = callback;
 285     p->recieve_packet_callback_userdata = userdata;
 286 }
 287
 288 void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, void (*callback) (pa_pstream *p, uint32_t channel, int64_t delta, pa_seek_mode_t seek, const pa_memchunk *chunk, void *userdata), void *userdata) {
 289     assert(p && callback);
 290
 291     p->recieve_memblock_callback = callback;
 292     p->recieve_memblock_callback_userdata = userdata;
 293 }
 294
 295 static void prepare_next_write_item(pa_pstream *p) {
 296     assert(p);
 297
 298     if (!(p->write.current = pa_queue_pop(p->send_queue)))
 299         return;
 300
 301     p->write.index = 0;
 302
 303     if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
 304         /*pa_log(__FILE__": pop-packet %p", p->write.current->packet);*/
 305
 306         assert(p->write.current->packet);
 307         p->write.data = p->write.current->packet->data;
 308         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length);
 309         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
 310         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = 0;
 311         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
 312         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] = 0;
 313     } else {
 314         assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK && p->write.current->chunk.memblock);
 315         p->write.data = (uint8_t*) p->write.current->chunk.memblock->data + p->write.current->chunk.index;
 316         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
 317         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel);
 318         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl((uint32_t) (((uint64_t) p->write.current->offset) >> 32));
 319         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = htonl((uint32_t) ((uint64_t) p->write.current->offset));
 320         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] = htonl(p->write.current->seek_mode);
 321     }
 322 }
 323
 324 static int do_write(pa_pstream *p) {
 325     void *d;
 326     size_t l;
 327     ssize_t r;
 328     assert(p);
 329
 330     if (!p->write.current)
 331         prepare_next_write_item(p);
 332
 333     if (!p->write.current)
 334         return 0;
 335
 336     assert(p->write.data);
 337
 338     if (p->write.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
 339         d = (uint8_t*) p->write.descriptor + p->write.index;
 340         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
 341     } else {
 342         d = (uint8_t*) p->write.data + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
 343         l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
 344     }
 345
 346     if ((r = pa_iochannel_write(p->io, d, l)) < 0)
 347         return -1;
 348
 349     p->write.index += r;
 350
 351     if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE+ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
 352         assert(p->write.current);
 353         item_free(p->write.current, (void *) 1);
 354         p->write.current = NULL;
 355
 356         if (p->drain_callback && !pa_pstream_is_pending(p))
 357             p->drain_callback(p, p->drain_userdata);
 358     }
 359
 360     return 0;
 361 }
 362
 363 static int do_read(pa_pstream *p) {
 364     void *d;
 365     size_t l;
 366     ssize_t r;
 367     assert(p);
 368
 369     if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
 370         d = (uint8_t*) p->read.descriptor + p->read.index;
 371         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
 372     } else {
 373         assert(p->read.data);
 374         d = (uint8_t*) p->read.data + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
 375         l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
 376     }
 377
 378     if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
 379         return -1;
 380
 381     p->read.index += r;
 382
 383     if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
 384         /* Reading of frame descriptor complete */
 385
 386         /* Frame size too large */
 387         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) > FRAME_SIZE_MAX) {
 388             pa_log_warn(__FILE__": Frame size too large");
 389             return -1;
 390         }
 391
 392         assert(!p->read.packet && !p->read.memblock);
 393
 394         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]) == (uint32_t) -1) {
 395             /* Frame is a packet frame */
 396             p->read.packet = pa_packet_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]));
 397             p->read.data = p->read.packet->data;
 398         } else {
 399             /* Frame is a memblock frame */
 400             p->read.memblock = pa_memblock_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]), p->memblock_stat);
 401             p->read.data = p->read.memblock->data;
 402
 403             if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK]) > PA_SEEK_RELATIVE_END) {
 404                 pa_log_warn(__FILE__": Invalid seek mode");
 405                 return -1;
 406             }
 407         }
 408
 409     } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) {
 410         /* Frame payload available */
 411
 412         if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblock data? Than pass it to the user */
 413             l = (p->read.index - r) < PA_PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE : (size_t) r;
 414
 415             if (l > 0) {
 416                 pa_memchunk chunk;
 417
 418                 chunk.memblock = p->read.memblock;
 419                 chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l;
 420                 chunk.length = l;
 421
 422                 if (p->recieve_memblock_callback) {
 423                     int64_t offset;
 424
 425                     offset = (int64_t) (
 426                             (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
 427                             (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO]))));
 428
 429                     p->recieve_memblock_callback(
 430                         p,
 431                         ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
 432                         offset,
 433                         ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK]),
 434                         &chunk,
 435                         p->recieve_memblock_callback_userdata);
 436                 }
 437
 438                 /* Drop seek info for following callbacks */
 439                 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_SEEK] =
 440                     p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] =
 441                     p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
 442             }
 443         }
 444
 445         /* Frame complete */
 446         if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) {
 447             if (p->read.memblock) {
 448                 assert(!p->read.packet);
 449
 450                 pa_memblock_unref(p->read.memblock);
 451                 p->read.memblock = NULL;
 452             } else {
 453                 assert(p->read.packet);
 454
 455                 if (p->recieve_packet_callback)
 456                     p->recieve_packet_callback(p, p->read.packet, p->recieve_packet_callback_userdata);
 457
 458                 pa_packet_unref(p->read.packet);
 459                 p->read.packet = NULL;
 460             }
 461
 462             p->read.index = 0;
 463         }
 464     }
 465
 466     return 0;
 467 }
 468
 469 void pa_pstream_set_die_callback(pa_pstream *p, void (*callback)(pa_pstream *p, void *userdata), void *userdata) {
 470     assert(p && callback);
 471     p->die_callback = callback;
 472     p->die_callback_userdata = userdata;
 473 }
 474
 475 int pa_pstream_is_pending(pa_pstream *p) {
 476     assert(p);
 477
 478     if (p->dead)
 479         return 0;
 480
 481     return p->write.current || !pa_queue_is_empty(p->send_queue);
 482 }
 483
 484 void pa_pstream_set_drain_callback(pa_pstream *p, void (*cb)(pa_pstream *p, void *userdata), void *userdata) {
 485     assert(p);
 486     assert(p->ref >= 1);
 487
 488     p->drain_callback = cb;
 489     p->drain_userdata = userdata;
 490 }
 491
 492 void pa_pstream_unref(pa_pstream*p) {
 493     assert(p);
 494     assert(p->ref >= 1);
 495
 496     if (--p->ref == 0)
 497         pstream_free(p);
 498 }
 499
 500 pa_pstream* pa_pstream_ref(pa_pstream*p) {
 501     assert(p);
 502     assert(p->ref >= 1);
 503
 504     p->ref++;
 505     return p;
 506 }
 507
 508 void pa_pstream_close(pa_pstream *p) {
 509     assert(p);
 510
 511     p->dead = 1;
 512
 513     if (p->io) {
 514         pa_iochannel_free(p->io);
 515         p->io = NULL;
 516     }
 517
 518     if (p->defer_event) {
 519         p->mainloop->defer_free(p->defer_event);
 520         p->defer_event = NULL;
 521     }
 522
 523     p->die_callback = NULL;
 524     p->drain_callback = NULL;
 525     p->recieve_packet_callback = NULL;
 526     p->recieve_memblock_callback = NULL;
 527 }