code.delx.au - pulseaudio/blob - polyp/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 General Public License as published
   8   by the Free Software Foundation; either version 2 of the License,
   9   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   General Public License for more details.
  15
  16   You should have received a copy of the GNU General Public License
  17   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 <stdlib.h>
  27 #include <assert.h>
  28 #include <netinet/in.h>
  29
  30 #include "pstream.h"
  31 #include "queue.h"
  32
  33 enum pa_pstream_descriptor_index {
  34     PA_PSTREAM_DESCRIPTOR_LENGTH,
  35     PA_PSTREAM_DESCRIPTOR_CHANNEL,
  36     PA_PSTREAM_DESCRIPTOR_DELTA,
  37     PA_PSTREAM_DESCRIPTOR_MAX
  38 };
  39
  40 typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX];
  41
  42 #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t))
  43 #define FRAME_SIZE_MAX (1024*64)
  44
  45 struct item_info {
  46     enum { PA_PSTREAM_ITEM_PACKET, PA_PSTREAM_ITEM_MEMBLOCK } type;
  47
  48     /* memblock info */
  49     struct pa_memchunk chunk;
  50     uint32_t channel;
  51     int32_t delta;
  52
  53     /* packet info */
  54     struct pa_packet *packet;
  55 };
  56
  57 struct pa_pstream {
  58     struct pa_mainloop_api *mainloop;
  59     struct mainloop_source *mainloop_source;
  60     struct pa_iochannel *io;
  61     struct pa_queue *send_queue;
  62
  63     int in_use, shall_free;
  64
  65     int dead;
  66     void (*die_callback) (struct pa_pstream *p, void *userdata);
  67     void *die_callback_userdata;
  68
  69     struct {
  70         struct item_info* current;
  71         pa_pstream_descriptor descriptor;
  72         void *data;
  73         size_t index;
  74     } write;
  75
  76     struct {
  77         struct pa_memblock *memblock;
  78         struct pa_packet *packet;
  79         pa_pstream_descriptor descriptor;
  80         void *data;
  81         size_t index;
  82     } read;
  83
  84     void (*recieve_packet_callback) (struct pa_pstream *p, struct pa_packet *packet, void *userdata);
  85     void *recieve_packet_callback_userdata;
  86
  87     void (*recieve_memblock_callback) (struct pa_pstream *p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk, void *userdata);
  88     void *recieve_memblock_callback_userdata;
  89
  90     void (*drain_callback)(struct pa_pstream *p, void *userdata);
  91     void *drain_userdata;
  92 };
  93
  94 static void do_write(struct pa_pstream *p);
  95 static void do_read(struct pa_pstream *p);
  96
  97 static void do_something(struct pa_pstream *p) {
  98     assert(p && !p->shall_free);
  99     p->mainloop->enable_fixed(p->mainloop, p->mainloop_source, 0);
 100
 101     if (p->dead)
 102         return;
 103
 104     if (pa_iochannel_is_hungup(p->io)) {
 105         p->dead = 1;
 106         if (p->die_callback)
 107             p->die_callback(p, p->die_callback_userdata);
 108
 109         return;
 110     }
 111
 112     if (pa_iochannel_is_writable(p->io)) {
 113         p->in_use = 1;
 114         do_write(p);
 115         p->in_use = 0;
 116
 117         if (p->shall_free) {
 118             pa_pstream_free(p);
 119             return;
 120         }
 121     }
 122
 123     if (pa_iochannel_is_readable(p->io)) {
 124         p->in_use = 1;
 125         do_read(p);
 126         p->in_use = 0;
 127         if (p->shall_free) {
 128             pa_pstream_free(p);
 129             return;
 130         }
 131     }
 132 }
 133
 134 static void io_callback(struct pa_iochannel*io, void *userdata) {
 135     struct pa_pstream *p = userdata;
 136     assert(p && p->io == io);
 137     do_something(p);
 138 }
 139
 140 static void fixed_callback(struct pa_mainloop_api *m, void *id, void*userdata) {
 141     struct pa_pstream *p = userdata;
 142     assert(p && p->mainloop_source == id && p->mainloop == m);
 143     do_something(p);
 144 }
 145
 146 struct pa_pstream *pa_pstream_new(struct pa_mainloop_api *m, struct pa_iochannel *io) {
 147     struct pa_pstream *p;
 148     assert(io);
 149
 150     p = malloc(sizeof(struct pa_pstream));
 151     assert(p);
 152
 153     p->io = io;
 154     pa_iochannel_set_callback(io, io_callback, p);
 155
 156     p->dead = 0;
 157     p->die_callback = NULL;
 158     p->die_callback_userdata = NULL;
 159
 160     p->mainloop = m;
 161     p->mainloop_source = m->source_fixed(m, fixed_callback, p);
 162     m->enable_fixed(m, p->mainloop_source, 0);
 163
 164     p->send_queue = pa_queue_new();
 165     assert(p->send_queue);
 166
 167     p->write.current = NULL;
 168     p->write.index = 0;
 169
 170     p->read.memblock = NULL;
 171     p->read.packet = NULL;
 172     p->read.index = 0;
 173
 174     p->recieve_packet_callback = NULL;
 175     p->recieve_packet_callback_userdata = NULL;
 176
 177     p->recieve_memblock_callback = NULL;
 178     p->recieve_memblock_callback_userdata = NULL;
 179
 180     p->drain_callback = NULL;
 181     p->drain_userdata = NULL;
 182
 183     p->in_use = p->shall_free = 0;
 184
 185     return p;
 186 }
 187
 188 static void item_free(void *item, void *p) {
 189     struct item_info *i = item;
 190     assert(i);
 191
 192     if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) {
 193         assert(i->chunk.memblock);
 194         pa_memblock_unref(i->chunk.memblock);
 195     } else {
 196         assert(i->type == PA_PSTREAM_ITEM_PACKET);
 197         assert(i->packet);
 198         pa_packet_unref(i->packet);
 199     }
 200
 201     free(i);
 202 }
 203
 204 void pa_pstream_free(struct pa_pstream *p) {
 205     assert(p);
 206
 207     if (p->in_use) {
 208         /* If this pstream object is used by someone else on the call stack, we have to postpone the freeing */
 209         p->dead = p->shall_free = 1;
 210         return;
 211     }
 212
 213     pa_iochannel_free(p->io);
 214     pa_queue_free(p->send_queue, item_free, NULL);
 215
 216     if (p->write.current)
 217         item_free(p->write.current, NULL);
 218
 219     if (p->read.memblock)
 220         pa_memblock_unref(p->read.memblock);
 221
 222     if (p->read.packet)
 223         pa_packet_unref(p->read.packet);
 224
 225     p->mainloop->cancel_fixed(p->mainloop, p->mainloop_source);
 226     free(p);
 227 }
 228
 229 void pa_pstream_send_packet(struct pa_pstream*p, struct pa_packet *packet) {
 230     struct item_info *i;
 231     assert(p && packet);
 232
 233     i = malloc(sizeof(struct item_info));
 234     assert(i);
 235     i->type = PA_PSTREAM_ITEM_PACKET;
 236     i->packet = pa_packet_ref(packet);
 237
 238     pa_queue_push(p->send_queue, i);
 239     p->mainloop->enable_fixed(p->mainloop, p->mainloop_source, 1);
 240 }
 241
 242 void pa_pstream_send_memblock(struct pa_pstream*p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk) {
 243     struct item_info *i;
 244     assert(p && channel != (uint32_t) -1 && chunk);
 245
 246     i = malloc(sizeof(struct item_info));
 247     assert(i);
 248     i->type = PA_PSTREAM_ITEM_MEMBLOCK;
 249     i->chunk = *chunk;
 250     i->channel = channel;
 251     i->delta = delta;
 252
 253     pa_memblock_ref(i->chunk.memblock);
 254
 255     pa_queue_push(p->send_queue, i);
 256     p->mainloop->enable_fixed(p->mainloop, p->mainloop_source, 1);
 257 }
 258
 259 void pa_pstream_set_recieve_packet_callback(struct pa_pstream *p, void (*callback) (struct pa_pstream *p, struct pa_packet *packet, void *userdata), void *userdata) {
 260     assert(p && callback);
 261
 262     p->recieve_packet_callback = callback;
 263     p->recieve_packet_callback_userdata = userdata;
 264 }
 265
 266 void pa_pstream_set_recieve_memblock_callback(struct pa_pstream *p, void (*callback) (struct pa_pstream *p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk, void *userdata), void *userdata) {
 267     assert(p && callback);
 268
 269     p->recieve_memblock_callback = callback;
 270     p->recieve_memblock_callback_userdata = userdata;
 271 }
 272
 273 static void prepare_next_write_item(struct pa_pstream *p) {
 274     assert(p);
 275
 276     if (!(p->write.current = pa_queue_pop(p->send_queue)))
 277         return;
 278
 279     p->write.index = 0;
 280
 281     if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
 282         assert(p->write.current->packet);
 283         p->write.data = p->write.current->packet->data;
 284         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length);
 285         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
 286         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = 0;
 287     } else {
 288         assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK && p->write.current->chunk.memblock);
 289         p->write.data = p->write.current->chunk.memblock->data + p->write.current->chunk.index;
 290         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
 291         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel);
 292         p->write.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA] = htonl(p->write.current->delta);
 293     }
 294 }
 295
 296 static void do_write(struct pa_pstream *p) {
 297     void *d;
 298     size_t l;
 299     ssize_t r;
 300     assert(p);
 301
 302     if (!p->write.current)
 303         prepare_next_write_item(p);
 304
 305     if (!p->write.current)
 306         return;
 307
 308     assert(p->write.data);
 309
 310     if (p->write.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
 311         d = (void*) p->write.descriptor + p->write.index;
 312         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
 313     } else {
 314         d = (void*) p->write.data + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
 315         l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
 316     }
 317
 318     if ((r = pa_iochannel_write(p->io, d, l)) < 0)
 319         goto die;
 320
 321     p->write.index += r;
 322
 323     if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE+ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
 324         assert(p->write.current);
 325         item_free(p->write.current, (void *) 1);
 326         p->write.current = NULL;
 327
 328         if (p->drain_callback && !pa_pstream_is_pending(p))
 329             p->drain_callback(p, p->drain_userdata);
 330     }
 331
 332     return;
 333
 334 die:
 335     p->dead = 1;
 336     if (p->die_callback)
 337         p->die_callback(p, p->die_callback_userdata);
 338 }
 339
 340 static void do_read(struct pa_pstream *p) {
 341     void *d;
 342     size_t l;
 343     ssize_t r;
 344     assert(p);
 345
 346     if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
 347         d = (void*) p->read.descriptor + p->read.index;
 348         l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
 349     } else {
 350         assert(p->read.data);
 351         d = (void*) p->read.data + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
 352         l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
 353     }
 354
 355     if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
 356         goto die;
 357
 358     p->read.index += r;
 359
 360     if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
 361         /* Reading of frame descriptor complete */
 362
 363         /* Frame size too large */
 364         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) > FRAME_SIZE_MAX)
 365             goto die;
 366
 367         assert(!p->read.packet && !p->read.memblock);
 368
 369         if (ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]) == (uint32_t) -1) {
 370             /* Frame is a packet frame */
 371             p->read.packet = pa_packet_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]));
 372             assert(p->read.packet);
 373             p->read.data = p->read.packet->data;
 374         } else {
 375             /* Frame is a memblock frame */
 376             p->read.memblock = pa_memblock_new(ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]));
 377             assert(p->read.memblock);
 378             p->read.data = p->read.memblock->data;
 379         }
 380
 381     } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) {
 382         /* Frame payload available */
 383
 384         if (p->read.memblock && p->recieve_memblock_callback) { /* Is this memblock data? Than pass it to the user */
 385             size_t l;
 386
 387             l = (p->read.index - r) < PA_PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE : (size_t) r;
 388
 389             if (l > 0) {
 390                 struct pa_memchunk chunk;
 391
 392                 chunk.memblock = p->read.memblock;
 393                 chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l;
 394                 chunk.length = l;
 395
 396                 if (p->recieve_memblock_callback)
 397                     p->recieve_memblock_callback(
 398                         p,
 399                         ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
 400                         (int32_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_DELTA]),
 401                         &chunk,
 402                         p->recieve_memblock_callback_userdata);
 403             }
 404         }
 405
 406         /* Frame complete */
 407         if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) {
 408             if (p->read.memblock) {
 409                 assert(!p->read.packet);
 410
 411                 pa_memblock_unref(p->read.memblock);
 412                 p->read.memblock = NULL;
 413             } else {
 414                 assert(p->read.packet);
 415
 416                 if (p->recieve_packet_callback)
 417                     p->recieve_packet_callback(p, p->read.packet, p->recieve_packet_callback_userdata);
 418
 419                 pa_packet_unref(p->read.packet);
 420                 p->read.packet = NULL;
 421             }
 422
 423             p->read.index = 0;
 424         }
 425     }
 426
 427     return;
 428
 429 die:
 430     p->dead = 1;
 431     if (p->die_callback)
 432         p->die_callback(p, p->die_callback_userdata);
 433
 434 }
 435
 436 void pa_pstream_set_die_callback(struct pa_pstream *p, void (*callback)(struct pa_pstream *p, void *userdata), void *userdata) {
 437     assert(p && callback);
 438     p->die_callback = callback;
 439     p->die_callback_userdata = userdata;
 440 }
 441
 442 int pa_pstream_is_pending(struct pa_pstream *p) {
 443     assert(p);
 444
 445     if (p->dead)
 446         return 0;
 447
 448     return p->write.current || !pa_queue_is_empty(p->send_queue);
 449 }
 450
 451 void pa_pstream_set_drain_callback(struct pa_pstream *p, void (*cb)(struct pa_pstream *p, void *userdata), void *userdata) {
 452     assert(p);
 453
 454     p->drain_callback = cb;
 455     p->drain_userdata = userdata;
 456 }
 457