src/polyp.c

   1 #include <stdio.h>
   2 #include <assert.h>
   3 #include <stdlib.h>
   4 #include <string.h>
   5
   6 #include "polyp.h"
   7 #include "protocol-native-spec.h"
   8 #include "pdispatch.h"
   9 #include "pstream.h"
  10 #include "dynarray.h"
  11 #include "socket-client.h"
  12 #include "pstream-util.h"
  13 #include "authkey.h"
  14 #include "util.h"
  15
  16 #define DEFAULT_MAXLENGTH 20480
  17 #define DEFAULT_TLENGTH 10240
  18 #define DEFAULT_PREBUF 4096
  19 #define DEFAULT_MINREQ 1024
  20
  21 #define DEFAULT_TIMEOUT (5*60)
  22 #define DEFAULT_SERVER "/tmp/polypaudio/native"
  23
  24 struct pa_context {
  25     char *name;
  26     struct pa_mainloop_api* mainloop;
  27     struct pa_socket_client *client;
  28     struct pa_pstream *pstream;
  29     struct pa_pdispatch *pdispatch;
  30     struct pa_dynarray *streams;
  31     struct pa_stream *first_stream;
  32     uint32_t ctag;
  33     uint32_t error;
  34     enum {
  35         CONTEXT_UNCONNECTED,
  36         CONTEXT_CONNECTING,
  37         CONTEXT_AUTHORIZING,
  38         CONTEXT_SETTING_NAME,
  39         CONTEXT_READY,
  40         CONTEXT_DEAD
  41     } state;
  42
  43     void (*connect_complete_callback)(struct pa_context*c, int success, void *userdata);
  44     void *connect_complete_userdata;
  45
  46     void (*drain_complete_callback)(struct pa_context*c, void *userdata);
  47     void *drain_complete_userdata;
  48
  49     void (*die_callback)(struct pa_context*c, void *userdata);
  50     void *die_userdata;
  51
  52     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
  53 };
  54
  55 struct pa_stream {
  56     struct pa_context *context;
  57     struct pa_stream *next, *previous;
  58
  59     char *name;
  60     struct pa_buffer_attr buffer_attr;
  61     struct pa_sample_spec sample_spec;
  62     uint32_t device_index;
  63     uint32_t channel;
  64     int channel_valid;
  65     enum pa_stream_direction direction;
  66
  67     enum { STREAM_LOOKING_UP, STREAM_CREATING, STREAM_READY, STREAM_DEAD} state;
  68     uint32_t requested_bytes;
  69
  70     void (*read_callback)(struct pa_stream *p, const void*data, size_t length, void *userdata);
  71     void *read_userdata;
  72
  73     void (*write_callback)(struct pa_stream *p, size_t length, void *userdata);
  74     void *write_userdata;
  75
  76     void (*create_complete_callback)(struct pa_stream *s, int success, void *userdata);
  77     void *create_complete_userdata;
  78
  79     void (*die_callback)(struct pa_stream*c, void *userdata);
  80     void *die_userdata;
  81 };
  82
  83 static void command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
  84
  85 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
  86     [PA_COMMAND_ERROR] = { NULL },
  87     [PA_COMMAND_REPLY] = { NULL },
  88     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { NULL },
  89     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { NULL },
  90     [PA_COMMAND_CREATE_RECORD_STREAM] = { NULL },
  91     [PA_COMMAND_DELETE_RECORD_STREAM] = { NULL },
  92     [PA_COMMAND_EXIT] = { NULL },
  93     [PA_COMMAND_REQUEST] = { command_request },
  94 };
  95
  96 struct pa_context *pa_context_new(struct pa_mainloop_api *mainloop, const char *name) {
  97     struct pa_context *c;
  98     assert(mainloop && name);
  99
 100     c = malloc(sizeof(struct pa_context));
 101     assert(c);
 102     c->name = strdup(name);
 103     c->mainloop = mainloop;
 104     c->client = NULL;
 105     c->pstream = NULL;
 106     c->pdispatch = NULL;
 107     c->streams = pa_dynarray_new();
 108     assert(c->streams);
 109     c->first_stream = NULL;
 110     c->error = PA_ERROR_OK;
 111     c->state = CONTEXT_UNCONNECTED;
 112     c->ctag = 0;
 113
 114     c->connect_complete_callback = NULL;
 115     c->connect_complete_userdata = NULL;
 116
 117     c->drain_complete_callback = NULL;
 118     c->drain_complete_userdata = NULL;
 119
 120     c->die_callback = NULL;
 121     c->die_userdata = NULL;
 122
 123     pa_check_for_sigpipe();
 124     return c;
 125 }
 126
 127 void pa_context_free(struct pa_context *c) {
 128     assert(c);
 129
 130     while (c->first_stream)
 131         pa_stream_free(c->first_stream);
 132
 133     if (c->client)
 134         pa_socket_client_free(c->client);
 135     if (c->pdispatch)
 136         pa_pdispatch_free(c->pdispatch);
 137     if (c->pstream)
 138         pa_pstream_free(c->pstream);
 139     if (c->streams)
 140         pa_dynarray_free(c->streams, NULL, NULL);
 141
 142     free(c->name);
 143     free(c);
 144 }
 145
 146 static void stream_dead(struct pa_stream *s) {
 147     assert(s);
 148
 149     if (s->state == STREAM_DEAD)
 150         return;
 151
 152     if (s->state == STREAM_READY) {
 153         s->state = STREAM_DEAD;
 154         if (s->die_callback)
 155             s->die_callback(s, s->die_userdata);
 156     } else
 157         s->state = STREAM_DEAD;
 158 }
 159
 160 static void context_dead(struct pa_context *c) {
 161     struct pa_stream *s;
 162     assert(c);
 163
 164     if (c->state == CONTEXT_DEAD)
 165         return;
 166
 167     if (c->pdispatch)
 168         pa_pdispatch_free(c->pdispatch);
 169     c->pdispatch = NULL;
 170
 171     if (c->pstream)
 172         pa_pstream_free(c->pstream);
 173     c->pstream = NULL;
 174
 175     if (c->client)
 176         pa_socket_client_free(c->client);
 177     c->client = NULL;
 178
 179     for (s = c->first_stream; s; s = s->next)
 180         stream_dead(s);
 181
 182     if (c->state == CONTEXT_READY) {
 183         c->state = CONTEXT_DEAD;
 184         if (c->die_callback)
 185             c->die_callback(c, c->die_userdata);
 186     } else
 187         s->state = CONTEXT_DEAD;
 188 }
 189
 190 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
 191     struct pa_context *c = userdata;
 192     assert(p && c);
 193     context_dead(c);
 194 }
 195
 196 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
 197     struct pa_context *c = userdata;
 198     assert(p && packet && c);
 199
 200     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
 201         fprintf(stderr, "polyp.c: invalid packet.\n");
 202         context_dead(c);
 203     }
 204 }
 205
 206 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, int32_t delta, struct pa_memchunk *chunk, void *userdata) {
 207     struct pa_context *c = userdata;
 208     struct pa_stream *s;
 209     assert(p && chunk && c && chunk->memblock && chunk->memblock->data);
 210
 211     if (!(s = pa_dynarray_get(c->streams, channel)))
 212         return;
 213
 214     if (s->read_callback)
 215         s->read_callback(s, chunk->memblock->data + chunk->index, chunk->length, s->read_userdata);
 216 }
 217
 218 static int handle_error(struct pa_context *c, uint32_t command, struct pa_tagstruct *t) {
 219     assert(c && t);
 220
 221     if (command == PA_COMMAND_ERROR) {
 222         if (pa_tagstruct_getu32(t, &c->error) < 0) {
 223             c->error = PA_ERROR_PROTOCOL;
 224             return -1;
 225         }
 226
 227         return 0;
 228     }
 229
 230     c->error = (command == PA_COMMAND_TIMEOUT) ? PA_ERROR_TIMEOUT : PA_ERROR_INTERNAL;
 231     return -1;
 232 }
 233
 234 static void setup_complete_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 235     struct pa_context *c = userdata;
 236     assert(pd && c && (c->state == CONTEXT_AUTHORIZING || c->state == CONTEXT_SETTING_NAME));
 237
 238     if (command != PA_COMMAND_REPLY) {
 239         handle_error(c, command, t);
 240         context_dead(c);
 241
 242         if (c->connect_complete_callback)
 243             c->connect_complete_callback(c, 0, c->connect_complete_userdata);
 244
 245         return;
 246     }
 247
 248     if (c->state == CONTEXT_AUTHORIZING) {
 249         struct pa_tagstruct *t;
 250         c->state = CONTEXT_SETTING_NAME;
 251         t = pa_tagstruct_new(NULL, 0);
 252         assert(t);
 253         pa_tagstruct_putu32(t, PA_COMMAND_SET_NAME);
 254         pa_tagstruct_putu32(t, tag = c->ctag++);
 255         pa_tagstruct_puts(t, c->name);
 256         pa_pstream_send_tagstruct(c->pstream, t);
 257         pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
 258     } else {
 259         assert(c->state == CONTEXT_SETTING_NAME);
 260
 261         c->state = CONTEXT_READY;
 262
 263         if (c->connect_complete_callback)
 264             c->connect_complete_callback(c, 1, c->connect_complete_userdata);
 265     }
 266
 267     return;
 268 }
 269
 270 static void on_connection(struct pa_socket_client *client, struct pa_iochannel*io, void *userdata) {
 271     struct pa_context *c = userdata;
 272     struct pa_tagstruct *t;
 273     uint32_t tag;
 274     assert(client && io && c && c->state == CONTEXT_CONNECTING);
 275
 276     pa_socket_client_free(client);
 277     c->client = NULL;
 278
 279     if (!io) {
 280         c->error = PA_ERROR_CONNECTIONREFUSED;
 281         context_dead(c);
 282
 283         if (c->connect_complete_callback)
 284             c->connect_complete_callback(c, 0, c->connect_complete_userdata);
 285
 286         return;
 287     }
 288
 289     c->pstream = pa_pstream_new(c->mainloop, io);
 290     assert(c->pstream);
 291     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
 292     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
 293     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
 294
 295     c->pdispatch = pa_pdispatch_new(c->mainloop, command_table, PA_COMMAND_MAX);
 296     assert(c->pdispatch);
 297
 298     t = pa_tagstruct_new(NULL, 0);
 299     assert(t);
 300     pa_tagstruct_putu32(t, PA_COMMAND_AUTH);
 301     pa_tagstruct_putu32(t, tag = c->ctag++);
 302     pa_tagstruct_put_arbitrary(t, c->auth_cookie, sizeof(c->auth_cookie));
 303     pa_pstream_send_tagstruct(c->pstream, t);
 304     pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
 305     c->state = CONTEXT_AUTHORIZING;
 306 }
 307
 308 int pa_context_connect(struct pa_context *c, const char *server, void (*complete) (struct pa_context*c, int success, void *userdata), void *userdata) {
 309     assert(c && c->state == CONTEXT_UNCONNECTED);
 310
 311     if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE, c->auth_cookie, sizeof(c->auth_cookie)) < 0) {
 312         c->error = PA_ERROR_AUTHKEY;
 313         return -1;
 314     }
 315
 316     assert(!c->client);
 317     if (!(c->client = pa_socket_client_new_unix(c->mainloop, server ? server : DEFAULT_SERVER))) {
 318         c->error = PA_ERROR_CONNECTIONREFUSED;
 319         return -1;
 320     }
 321
 322     c->connect_complete_callback = complete;
 323     c->connect_complete_userdata = userdata;
 324
 325     pa_socket_client_set_callback(c->client, on_connection, c);
 326     c->state = CONTEXT_CONNECTING;
 327
 328     return 0;
 329 }
 330
 331 int pa_context_is_dead(struct pa_context *c) {
 332     assert(c);
 333     return c->state == CONTEXT_DEAD;
 334 }
 335
 336 int pa_context_is_ready(struct pa_context *c) {
 337     assert(c);
 338     return c->state == CONTEXT_READY;
 339 }
 340
 341 int pa_context_errno(struct pa_context *c) {
 342     assert(c);
 343     return c->error;
 344 }
 345
 346 void pa_context_set_die_callback(struct pa_context *c, void (*cb)(struct pa_context *c, void *userdata), void *userdata) {
 347     assert(c);
 348     c->die_callback = cb;
 349     c->die_userdata = userdata;
 350 }
 351
 352 static void command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 353     struct pa_stream *s;
 354     struct pa_context *c = userdata;
 355     uint32_t bytes, channel;
 356     assert(pd && command == PA_COMMAND_REQUEST && t && c);
 357
 358     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 359         pa_tagstruct_getu32(t, &bytes) < 0 ||
 360         !pa_tagstruct_eof(t)) {
 361         c->error = PA_ERROR_PROTOCOL;
 362         context_dead(c);
 363         return;
 364     }
 365
 366     if (!(s = pa_dynarray_get(c->streams, channel)))
 367         return;
 368
 369     if (s->state != STREAM_READY)
 370         return;
 371
 372     s->requested_bytes += bytes;
 373
 374     if (s->requested_bytes && s->write_callback)
 375         s->write_callback(s, s->requested_bytes, s->write_userdata);
 376 }
 377
 378 static void create_playback_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 379     struct pa_stream *s = userdata;
 380     assert(pd && s && s->state == STREAM_CREATING);
 381
 382     if (command != PA_COMMAND_REPLY) {
 383         if (handle_error(s->context, command, t) < 0) {
 384             context_dead(s->context);
 385             return;
 386         }
 387
 388         stream_dead(s);
 389         if (s->create_complete_callback)
 390             s->create_complete_callback(s, 0, s->create_complete_userdata);
 391
 392         return;
 393     }
 394
 395     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 396         pa_tagstruct_getu32(t, &s->device_index) < 0 ||
 397         !pa_tagstruct_eof(t)) {
 398         s->context->error = PA_ERROR_PROTOCOL;
 399         context_dead(s->context);
 400         return;
 401     }
 402
 403     s->channel_valid = 1;
 404     pa_dynarray_put(s->context->streams, s->channel, s);
 405
 406     s->state = STREAM_READY;
 407     if (s->create_complete_callback)
 408         s->create_complete_callback(s, 1, s->create_complete_userdata);
 409 }
 410
 411 static void create_stream(struct pa_stream *s, uint32_t tdev_index) {
 412     struct pa_tagstruct *t;
 413     uint32_t tag;
 414     assert(s);
 415
 416     s->state = STREAM_CREATING;
 417
 418     t = pa_tagstruct_new(NULL, 0);
 419     assert(t);
 420
 421     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM);
 422     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 423     pa_tagstruct_puts(t, s->name);
 424     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
 425     pa_tagstruct_putu32(t, tdev_index);
 426     pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
 427     pa_tagstruct_putu32(t, s->buffer_attr.tlength);
 428     pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
 429     pa_tagstruct_putu32(t, s->buffer_attr.minreq);
 430
 431     pa_pstream_send_tagstruct(s->context->pstream, t);
 432     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, create_playback_callback, s);
 433 }
 434
 435 static void lookup_device_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 436     struct pa_stream *s = userdata;
 437     uint32_t tdev;
 438     assert(pd && s && s->state == STREAM_LOOKING_UP);
 439
 440     if (command != PA_COMMAND_REPLY) {
 441         if (handle_error(s->context, command, t) < 0) {
 442             context_dead(s->context);
 443             return;
 444         }
 445
 446         stream_dead(s);
 447         if (s->create_complete_callback)
 448             s->create_complete_callback(s, 0, s->create_complete_userdata);
 449         return;
 450     }
 451
 452     if (pa_tagstruct_getu32(t, &tdev) < 0 ||
 453         !pa_tagstruct_eof(t)) {
 454         s->context->error = PA_ERROR_PROTOCOL;
 455         context_dead(s->context);
 456         return;
 457     }
 458
 459     create_stream(s, tdev);
 460 }
 461
 462 static void lookup_device(struct pa_stream *s, const char *tdev) {
 463     struct pa_tagstruct *t;
 464     uint32_t tag;
 465     assert(s);
 466
 467     s->state = STREAM_LOOKING_UP;
 468
 469     t = pa_tagstruct_new(NULL, 0);
 470     assert(t);
 471
 472     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_LOOKUP_SINK : PA_COMMAND_LOOKUP_SOURCE);
 473     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 474     pa_tagstruct_puts(t, tdev);
 475
 476     pa_pstream_send_tagstruct(s->context->pstream, t);
 477     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, lookup_device_callback, s);
 478 }
 479
 480 struct pa_stream* pa_stream_new(
 481     struct pa_context *c,
 482     enum pa_stream_direction dir,
 483     const char *dev,
 484     const char *name,
 485     const struct pa_sample_spec *ss,
 486     const struct pa_buffer_attr *attr,
 487     void (*complete) (struct pa_stream*s, int success, void *userdata),
 488     void *userdata) {
 489
 490     struct pa_stream *s;
 491
 492     assert(c && name && ss && c->state == CONTEXT_READY);
 493
 494     s = malloc(sizeof(struct pa_stream));
 495     assert(s);
 496     s->context = c;
 497
 498     s->read_callback = NULL;
 499     s->read_userdata = NULL;
 500     s->write_callback = NULL;
 501     s->write_userdata = NULL;
 502     s->die_callback = NULL;
 503     s->die_userdata = NULL;
 504     s->create_complete_callback = complete;
 505     s->create_complete_userdata = NULL;
 506
 507     s->name = strdup(name);
 508     s->state = STREAM_CREATING;
 509     s->requested_bytes = 0;
 510     s->channel = 0;
 511     s->channel_valid = 0;
 512     s->device_index = (uint32_t) -1;
 513     s->direction = dir;
 514     s->sample_spec = *ss;
 515     if (attr)
 516         s->buffer_attr = *attr;
 517     else {
 518         s->buffer_attr.maxlength = DEFAULT_MAXLENGTH;
 519         s->buffer_attr.tlength = DEFAULT_TLENGTH;
 520         s->buffer_attr.prebuf = DEFAULT_PREBUF;
 521         s->buffer_attr.minreq = DEFAULT_MINREQ;
 522     }
 523
 524     s->next = c->first_stream;
 525     if (s->next)
 526         s->next->previous = s;
 527     s->previous = NULL;
 528     c->first_stream = s;
 529
 530     if (dev)
 531         lookup_device(s, dev);
 532     else
 533         create_stream(s, (uint32_t) -1);
 534
 535     return s;
 536 }
 537
 538 void pa_stream_free(struct pa_stream *s) {
 539     assert(s && s->context);
 540
 541     free(s->name);
 542
 543     if (s->channel_valid && s->context->state == CONTEXT_READY) {
 544         struct pa_tagstruct *t = pa_tagstruct_new(NULL, 0);
 545         assert(t);
 546
 547         pa_tagstruct_putu32(t, PA_COMMAND_DELETE_PLAYBACK_STREAM);
 548         pa_tagstruct_putu32(t, s->context->ctag++);
 549         pa_tagstruct_putu32(t, s->channel);
 550         pa_pstream_send_tagstruct(s->context->pstream, t);
 551     }
 552
 553     if (s->channel_valid)
 554         pa_dynarray_put(s->context->streams, s->channel, NULL);
 555
 556     if (s->next)
 557         s->next->previous = s->previous;
 558     if (s->previous)
 559         s->previous->next = s->next;
 560     else
 561         s->context->first_stream = s->next;
 562
 563     free(s);
 564 }
 565
 566 void pa_stream_set_write_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, size_t length, void *userdata), void *userdata) {
 567     assert(s && cb);
 568     s->write_callback = cb;
 569     s->write_userdata = userdata;
 570 }
 571
 572 void pa_stream_write(struct pa_stream *s, const void *data, size_t length) {
 573     struct pa_memchunk chunk;
 574     assert(s && s->context && data && length && s->state == STREAM_READY);
 575
 576     chunk.memblock = pa_memblock_new(length);
 577     assert(chunk.memblock && chunk.memblock->data);
 578     memcpy(chunk.memblock->data, data, length);
 579     chunk.index = 0;
 580     chunk.length = length;
 581
 582     pa_pstream_send_memblock(s->context->pstream, s->channel, 0, &chunk);
 583     pa_memblock_unref(chunk.memblock);
 584
 585     /*fprintf(stderr, "Sent %u bytes\n", length);*/
 586
 587     if (length < s->requested_bytes)
 588         s->requested_bytes -= length;
 589     else
 590         s->requested_bytes = 0;
 591 }
 592
 593 size_t pa_stream_writable_size(struct pa_stream *s) {
 594     assert(s && s->state == STREAM_READY);
 595     return s->requested_bytes;
 596 }
 597
 598 void pa_stream_set_read_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, const void*data, size_t length, void *userdata), void *userdata) {
 599     assert(s && cb);
 600     s->read_callback = cb;
 601     s->read_userdata = userdata;
 602 }
 603
 604 int pa_stream_is_dead(struct pa_stream *s) {
 605     return s->state == STREAM_DEAD;
 606 }
 607
 608 int pa_stream_is_ready(struct pa_stream*s) {
 609     return s->state == STREAM_READY;
 610 }
 611
 612 void pa_stream_set_die_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata) {
 613     assert(s);
 614     s->die_callback = cb;
 615     s->die_userdata = userdata;
 616 }
 617
 618 int pa_context_is_pending(struct pa_context *c) {
 619     assert(c);
 620
 621     if (c->state != CONTEXT_READY)
 622         return 0;
 623
 624     return pa_pstream_is_pending(c->pstream) || pa_pdispatch_is_pending(c->pdispatch);
 625 }
 626
 627 struct pa_context* pa_stream_get_context(struct pa_stream *p) {
 628     assert(p);
 629     return p->context;
 630 }
 631
 632 static void set_dispatch_callbacks(struct pa_context *c);
 633
 634 static void pdispatch_drain_callback(struct pa_pdispatch*pd, void *userdata) {
 635     set_dispatch_callbacks(userdata);
 636 }
 637
 638 static void pstream_drain_callback(struct pa_pstream *s, void *userdata) {
 639     set_dispatch_callbacks(userdata);
 640 }
 641
 642 static void set_dispatch_callbacks(struct pa_context *c) {
 643     assert(c && c->state == CONTEXT_READY);
 644
 645     pa_pstream_set_drain_callback(c->pstream, NULL, NULL);
 646     pa_pdispatch_set_drain_callback(c->pdispatch, NULL, NULL);
 647
 648     if (pa_pdispatch_is_pending(c->pdispatch)) {
 649         pa_pdispatch_set_drain_callback(c->pdispatch, pdispatch_drain_callback, c);
 650         return;
 651     }
 652
 653     if (pa_pstream_is_pending(c->pstream)) {
 654         pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
 655         return;
 656     }
 657
 658     assert(c->drain_complete_callback);
 659     c->drain_complete_callback(c, c->drain_complete_userdata);
 660 }
 661
 662 int pa_context_drain(
 663     struct pa_context *c,
 664     void (*complete) (struct pa_context*c, void *userdata),
 665     void *userdata) {
 666
 667     assert(c && c->state == CONTEXT_READY);
 668
 669     if (complete == NULL) {
 670         c->drain_complete_callback = NULL;
 671         pa_pstream_set_drain_callback(c->pstream, NULL, NULL);
 672         pa_pdispatch_set_drain_callback(c->pdispatch, NULL, NULL);
 673         return 0;
 674     }
 675
 676     if (!pa_context_is_pending(c))
 677         return -1;
 678
 679     c->drain_complete_callback = complete;
 680     c->drain_complete_userdata = userdata;
 681
 682     set_dispatch_callbacks(c);
 683
 684     return 0;
 685 }