code.delx.au - pulseaudio/blob - src/polyplib.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 <stdio.h>
  27 #include <assert.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30 #include <sys/types.h>
  31 #include <sys/socket.h>
  32 #include <netdb.h>
  33
  34 #include "polyplib.h"
  35 #include "native-common.h"
  36 #include "pdispatch.h"
  37 #include "pstream.h"
  38 #include "dynarray.h"
  39 #include "socket-client.h"
  40 #include "pstream-util.h"
  41 #include "authkey.h"
  42 #include "util.h"
  43
  44 #define DEFAULT_MAXLENGTH 204800
  45 #define DEFAULT_TLENGTH 10240
  46 #define DEFAULT_PREBUF 4096
  47 #define DEFAULT_MINREQ 1024
  48 #define DEFAULT_FRAGSIZE 1024
  49
  50 #define DEFAULT_TIMEOUT (5*60)
  51 #define DEFAULT_SERVER "/tmp/polypaudio/native"
  52 #define DEFAULT_PORT "4713"
  53
  54 struct pa_context {
  55     char *name;
  56     struct pa_mainloop_api* mainloop;
  57     struct pa_socket_client *client;
  58     struct pa_pstream *pstream;
  59     struct pa_pdispatch *pdispatch;
  60     struct pa_dynarray *record_streams, *playback_streams;
  61     struct pa_stream *first_stream;
  62     uint32_t ctag;
  63     uint32_t error;
  64     enum {
  65         CONTEXT_UNCONNECTED,
  66         CONTEXT_CONNECTING,
  67         CONTEXT_AUTHORIZING,
  68         CONTEXT_SETTING_NAME,
  69         CONTEXT_READY,
  70         CONTEXT_DEAD
  71     } state;
  72
  73     void (*connect_complete_callback)(struct pa_context*c, int success, void *userdata);
  74     void *connect_complete_userdata;
  75
  76     void (*drain_complete_callback)(struct pa_context*c, void *userdata);
  77     void *drain_complete_userdata;
  78
  79     void (*die_callback)(struct pa_context*c, void *userdata);
  80     void *die_userdata;
  81
  82     void (*stat_callback)(struct pa_context*c, uint32_t count, uint32_t total, void *userdata);
  83     void *stat_userdata;
  84
  85     uint8_t auth_cookie[PA_NATIVE_COOKIE_LENGTH];
  86 };
  87
  88 struct pa_stream {
  89     struct pa_context *context;
  90     struct pa_stream *next, *previous;
  91
  92     char *name;
  93     struct pa_buffer_attr buffer_attr;
  94     struct pa_sample_spec sample_spec;
  95     uint32_t device_index;
  96     uint32_t channel;
  97     int channel_valid;
  98     enum pa_stream_direction direction;
  99
 100     enum { STREAM_LOOKING_UP, STREAM_CREATING, STREAM_READY, STREAM_DEAD} state;
 101     uint32_t requested_bytes;
 102
 103     void (*read_callback)(struct pa_stream *p, const void*data, size_t length, void *userdata);
 104     void *read_userdata;
 105
 106     void (*write_callback)(struct pa_stream *p, size_t length, void *userdata);
 107     void *write_userdata;
 108
 109     void (*create_complete_callback)(struct pa_stream *s, int success, void *userdata);
 110     void *create_complete_userdata;
 111
 112     void (*drain_complete_callback)(struct pa_stream *s, void *userdata);
 113     void *drain_complete_userdata;
 114
 115     void (*die_callback)(struct pa_stream*c, void *userdata);
 116     void *die_userdata;
 117
 118     void (*get_latency_callback)(struct pa_stream*c, uint32_t latency, void *userdata);
 119     void *get_latency_userdata;
 120 };
 121
 122 static void command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 123 static void command_playback_stream_killed(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 124
 125 static const struct pa_pdispatch_command command_table[PA_COMMAND_MAX] = {
 126     [PA_COMMAND_ERROR] = { NULL },
 127     [PA_COMMAND_REPLY] = { NULL },
 128     [PA_COMMAND_CREATE_PLAYBACK_STREAM] = { NULL },
 129     [PA_COMMAND_DELETE_PLAYBACK_STREAM] = { NULL },
 130     [PA_COMMAND_CREATE_RECORD_STREAM] = { NULL },
 131     [PA_COMMAND_DELETE_RECORD_STREAM] = { NULL },
 132     [PA_COMMAND_EXIT] = { NULL },
 133     [PA_COMMAND_REQUEST] = { command_request },
 134     [PA_COMMAND_PLAYBACK_STREAM_KILLED] = { command_playback_stream_killed },
 135     [PA_COMMAND_RECORD_STREAM_KILLED] = { command_playback_stream_killed },
 136 };
 137
 138 struct pa_context *pa_context_new(struct pa_mainloop_api *mainloop, const char *name) {
 139     struct pa_context *c;
 140     assert(mainloop && name);
 141
 142     c = malloc(sizeof(struct pa_context));
 143     assert(c);
 144     c->name = strdup(name);
 145     c->mainloop = mainloop;
 146     c->client = NULL;
 147     c->pstream = NULL;
 148     c->pdispatch = NULL;
 149     c->playback_streams = pa_dynarray_new();
 150     assert(c->playback_streams);
 151     c->record_streams = pa_dynarray_new();
 152     assert(c->record_streams);
 153     c->first_stream = NULL;
 154     c->error = PA_ERROR_OK;
 155     c->state = CONTEXT_UNCONNECTED;
 156     c->ctag = 0;
 157
 158     c->connect_complete_callback = NULL;
 159     c->connect_complete_userdata = NULL;
 160
 161     c->drain_complete_callback = NULL;
 162     c->drain_complete_userdata = NULL;
 163
 164     c->die_callback = NULL;
 165     c->die_userdata = NULL;
 166
 167     c->stat_callback = NULL;
 168     c->stat_userdata = NULL;
 169
 170     pa_check_for_sigpipe();
 171     return c;
 172 }
 173
 174 void pa_context_free(struct pa_context *c) {
 175     assert(c);
 176
 177     while (c->first_stream)
 178         pa_stream_free(c->first_stream);
 179
 180     if (c->client)
 181         pa_socket_client_free(c->client);
 182     if (c->pdispatch)
 183         pa_pdispatch_free(c->pdispatch);
 184     if (c->pstream)
 185         pa_pstream_free(c->pstream);
 186     if (c->record_streams)
 187         pa_dynarray_free(c->record_streams, NULL, NULL);
 188     if (c->playback_streams)
 189         pa_dynarray_free(c->playback_streams, NULL, NULL);
 190
 191     free(c->name);
 192     free(c);
 193 }
 194
 195 static void stream_dead(struct pa_stream *s) {
 196     assert(s);
 197
 198     if (s->state == STREAM_DEAD)
 199         return;
 200
 201     if (s->state == STREAM_READY) {
 202         s->state = STREAM_DEAD;
 203         if (s->die_callback)
 204             s->die_callback(s, s->die_userdata);
 205     } else
 206         s->state = STREAM_DEAD;
 207 }
 208
 209 static void context_dead(struct pa_context *c) {
 210     struct pa_stream *s;
 211     assert(c);
 212
 213     if (c->state == CONTEXT_DEAD)
 214         return;
 215
 216     if (c->pdispatch)
 217         pa_pdispatch_free(c->pdispatch);
 218     c->pdispatch = NULL;
 219
 220     if (c->pstream)
 221         pa_pstream_free(c->pstream);
 222     c->pstream = NULL;
 223
 224     if (c->client)
 225         pa_socket_client_free(c->client);
 226     c->client = NULL;
 227
 228     for (s = c->first_stream; s; s = s->next)
 229         stream_dead(s);
 230
 231     if (c->state == CONTEXT_READY) {
 232         c->state = CONTEXT_DEAD;
 233         if (c->die_callback)
 234             c->die_callback(c, c->die_userdata);
 235     } else
 236         c->state = CONTEXT_DEAD;
 237 }
 238
 239 static void pstream_die_callback(struct pa_pstream *p, void *userdata) {
 240     struct pa_context *c = userdata;
 241     assert(p && c);
 242     c->error = PA_ERROR_CONNECTIONTERMINATED;
 243     context_dead(c);
 244 }
 245
 246 static void pstream_packet_callback(struct pa_pstream *p, struct pa_packet *packet, void *userdata) {
 247     struct pa_context *c = userdata;
 248     assert(p && packet && c);
 249
 250     if (pa_pdispatch_run(c->pdispatch, packet, c) < 0) {
 251         fprintf(stderr, "polyp.c: invalid packet.\n");
 252         c->error = PA_ERROR_PROTOCOL;
 253         context_dead(c);
 254     }
 255 }
 256
 257 static void pstream_memblock_callback(struct pa_pstream *p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk, void *userdata) {
 258     struct pa_context *c = userdata;
 259     struct pa_stream *s;
 260     assert(p && chunk && c && chunk->memblock && chunk->memblock->data);
 261
 262     if (!(s = pa_dynarray_get(c->record_streams, channel)))
 263         return;
 264
 265     if (s->read_callback)
 266         s->read_callback(s, chunk->memblock->data + chunk->index, chunk->length, s->read_userdata);
 267 }
 268
 269 static int handle_error(struct pa_context *c, uint32_t command, struct pa_tagstruct *t) {
 270     assert(c && t);
 271
 272     if (command == PA_COMMAND_ERROR) {
 273         if (pa_tagstruct_getu32(t, &c->error) < 0) {
 274             c->error = PA_ERROR_PROTOCOL;
 275             return -1;
 276         }
 277
 278         return 0;
 279     }
 280
 281     c->error = (command == PA_COMMAND_TIMEOUT) ? PA_ERROR_TIMEOUT : PA_ERROR_INTERNAL;
 282     return -1;
 283 }
 284
 285 static void setup_complete_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 286     struct pa_context *c = userdata;
 287     assert(pd && c && (c->state == CONTEXT_AUTHORIZING || c->state == CONTEXT_SETTING_NAME));
 288
 289     if (command != PA_COMMAND_REPLY) {
 290         handle_error(c, command, t);
 291         context_dead(c);
 292
 293         if (c->connect_complete_callback)
 294             c->connect_complete_callback(c, 0, c->connect_complete_userdata);
 295
 296         return;
 297     }
 298
 299     if (c->state == CONTEXT_AUTHORIZING) {
 300         struct pa_tagstruct *t;
 301         c->state = CONTEXT_SETTING_NAME;
 302         t = pa_tagstruct_new(NULL, 0);
 303         assert(t);
 304         pa_tagstruct_putu32(t, PA_COMMAND_SET_NAME);
 305         pa_tagstruct_putu32(t, tag = c->ctag++);
 306         pa_tagstruct_puts(t, c->name);
 307         pa_pstream_send_tagstruct(c->pstream, t);
 308         pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
 309     } else {
 310         assert(c->state == CONTEXT_SETTING_NAME);
 311
 312         c->state = CONTEXT_READY;
 313
 314         if (c->connect_complete_callback)
 315             c->connect_complete_callback(c, 1, c->connect_complete_userdata);
 316     }
 317
 318     return;
 319 }
 320
 321 static void on_connection(struct pa_socket_client *client, struct pa_iochannel*io, void *userdata) {
 322     struct pa_context *c = userdata;
 323     struct pa_tagstruct *t;
 324     uint32_t tag;
 325     assert(client && c && c->state == CONTEXT_CONNECTING);
 326
 327     pa_socket_client_free(client);
 328     c->client = NULL;
 329
 330     if (!io) {
 331         c->error = PA_ERROR_CONNECTIONREFUSED;
 332         context_dead(c);
 333
 334         if (c->connect_complete_callback)
 335             c->connect_complete_callback(c, 0, c->connect_complete_userdata);
 336
 337         return;
 338     }
 339
 340     c->pstream = pa_pstream_new(c->mainloop, io);
 341     assert(c->pstream);
 342     pa_pstream_set_die_callback(c->pstream, pstream_die_callback, c);
 343     pa_pstream_set_recieve_packet_callback(c->pstream, pstream_packet_callback, c);
 344     pa_pstream_set_recieve_memblock_callback(c->pstream, pstream_memblock_callback, c);
 345
 346     c->pdispatch = pa_pdispatch_new(c->mainloop, command_table, PA_COMMAND_MAX);
 347     assert(c->pdispatch);
 348
 349     t = pa_tagstruct_new(NULL, 0);
 350     assert(t);
 351     pa_tagstruct_putu32(t, PA_COMMAND_AUTH);
 352     pa_tagstruct_putu32(t, tag = c->ctag++);
 353     pa_tagstruct_put_arbitrary(t, c->auth_cookie, sizeof(c->auth_cookie));
 354     pa_pstream_send_tagstruct(c->pstream, t);
 355     pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
 356     c->state = CONTEXT_AUTHORIZING;
 357 }
 358
 359 static struct sockaddr *resolve_server(const char *server, size_t *len) {
 360     struct sockaddr *sa;
 361     struct addrinfo hints, *result = NULL;
 362     char *port;
 363     assert(server && len);
 364
 365     if ((port = strrchr(server, ':')))
 366         port++;
 367     if (!port)
 368         port = DEFAULT_PORT;
 369
 370     memset(&hints, 0, sizeof(hints));
 371     hints.ai_family = PF_UNSPEC;
 372     hints.ai_socktype = SOCK_STREAM;
 373     hints.ai_protocol = 0;
 374
 375     if (getaddrinfo(server, port, &hints, &result) != 0)
 376         return NULL;
 377     assert(result);
 378
 379     sa = malloc(*len = result->ai_addrlen);
 380     assert(sa);
 381     memcpy(sa, result->ai_addr, *len);
 382
 383     freeaddrinfo(result);
 384
 385     return sa;
 386
 387 }
 388
 389 int pa_context_connect(struct pa_context *c, const char *server, void (*complete) (struct pa_context*c, int success, void *userdata), void *userdata) {
 390     assert(c && c->state == CONTEXT_UNCONNECTED);
 391
 392     if (pa_authkey_load_from_home(PA_NATIVE_COOKIE_FILE, c->auth_cookie, sizeof(c->auth_cookie)) < 0) {
 393         c->error = PA_ERROR_AUTHKEY;
 394         return -1;
 395     }
 396
 397     if (!server)
 398         if (!(server = getenv("POLYP_SERVER")))
 399             server = DEFAULT_SERVER;
 400
 401     assert(!c->client);
 402
 403     if (*server == '/') {
 404         if (!(c->client = pa_socket_client_new_unix(c->mainloop, server))) {
 405             c->error = PA_ERROR_CONNECTIONREFUSED;
 406             return -1;
 407         }
 408     } else {
 409         struct sockaddr* sa;
 410         size_t sa_len;
 411
 412         if (!(sa = resolve_server(server, &sa_len))) {
 413             c->error = PA_ERROR_INVALIDSERVER;
 414             return -1;
 415         }
 416
 417         c->client = pa_socket_client_new_sockaddr(c->mainloop, sa, sa_len);
 418         free(sa);
 419
 420         if (!c->client) {
 421             c->error = PA_ERROR_CONNECTIONREFUSED;
 422             return -1;
 423         }
 424     }
 425
 426     c->connect_complete_callback = complete;
 427     c->connect_complete_userdata = userdata;
 428
 429     pa_socket_client_set_callback(c->client, on_connection, c);
 430     c->state = CONTEXT_CONNECTING;
 431
 432     return 0;
 433 }
 434
 435 int pa_context_is_dead(struct pa_context *c) {
 436     assert(c);
 437     return c->state == CONTEXT_DEAD;
 438 }
 439
 440 int pa_context_is_ready(struct pa_context *c) {
 441     assert(c);
 442     return c->state == CONTEXT_READY;
 443 }
 444
 445 int pa_context_errno(struct pa_context *c) {
 446     assert(c);
 447     return c->error;
 448 }
 449
 450 void pa_context_set_die_callback(struct pa_context *c, void (*cb)(struct pa_context *c, void *userdata), void *userdata) {
 451     assert(c);
 452     c->die_callback = cb;
 453     c->die_userdata = userdata;
 454 }
 455
 456 static void command_playback_stream_killed(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 457     struct pa_context *c = userdata;
 458     struct pa_stream *s;
 459     uint32_t channel;
 460     assert(pd && (command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED) && t && c);
 461
 462     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 463         !pa_tagstruct_eof(t)) {
 464         c->error = PA_ERROR_PROTOCOL;
 465         context_dead(c);
 466         return;
 467     }
 468
 469     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 470         return;
 471
 472     c->error = PA_ERROR_KILLED;
 473     stream_dead(s);
 474 }
 475
 476 static void command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 477     struct pa_stream *s;
 478     struct pa_context *c = userdata;
 479     uint32_t bytes, channel;
 480     assert(pd && command == PA_COMMAND_REQUEST && t && c);
 481
 482     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 483         pa_tagstruct_getu32(t, &bytes) < 0 ||
 484         !pa_tagstruct_eof(t)) {
 485         c->error = PA_ERROR_PROTOCOL;
 486         context_dead(c);
 487         return;
 488     }
 489
 490     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 491         return;
 492
 493     if (s->state != STREAM_READY)
 494         return;
 495
 496     s->requested_bytes += bytes;
 497
 498     if (s->requested_bytes && s->write_callback)
 499         s->write_callback(s, s->requested_bytes, s->write_userdata);
 500 }
 501
 502 static void create_stream_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 503     struct pa_stream *s = userdata;
 504     assert(pd && s && s->state == STREAM_CREATING);
 505
 506     if (command != PA_COMMAND_REPLY) {
 507         if (handle_error(s->context, command, t) < 0) {
 508             context_dead(s->context);
 509             return;
 510         }
 511
 512         stream_dead(s);
 513         if (s->create_complete_callback)
 514             s->create_complete_callback(s, 0, s->create_complete_userdata);
 515
 516         return;
 517     }
 518
 519     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 520         pa_tagstruct_getu32(t, &s->device_index) < 0 ||
 521         !pa_tagstruct_eof(t)) {
 522         s->context->error = PA_ERROR_PROTOCOL;
 523         context_dead(s->context);
 524         return;
 525     }
 526
 527     s->channel_valid = 1;
 528     pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams :  s->context->record_streams, s->channel, s);
 529
 530     s->state = STREAM_READY;
 531     if (s->create_complete_callback)
 532         s->create_complete_callback(s, 1, s->create_complete_userdata);
 533 }
 534
 535 static void create_stream(struct pa_stream *s, uint32_t tdev_index) {
 536     struct pa_tagstruct *t;
 537     uint32_t tag;
 538     assert(s);
 539
 540     s->state = STREAM_CREATING;
 541
 542     t = pa_tagstruct_new(NULL, 0);
 543     assert(t);
 544
 545     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM);
 546     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 547     pa_tagstruct_puts(t, s->name);
 548     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
 549     pa_tagstruct_putu32(t, tdev_index);
 550     pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
 551     if (s->direction == PA_STREAM_PLAYBACK) {
 552         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
 553         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
 554         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
 555     } else
 556         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 557
 558     pa_pstream_send_tagstruct(s->context->pstream, t);
 559     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, s);
 560 }
 561
 562 static void lookup_device_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 563     struct pa_stream *s = userdata;
 564     uint32_t tdev;
 565     assert(pd && s && s->state == STREAM_LOOKING_UP);
 566
 567     if (command != PA_COMMAND_REPLY) {
 568         if (handle_error(s->context, command, t) < 0) {
 569             context_dead(s->context);
 570             return;
 571         }
 572
 573         stream_dead(s);
 574         if (s->create_complete_callback)
 575             s->create_complete_callback(s, 0, s->create_complete_userdata);
 576         return;
 577     }
 578
 579     if (pa_tagstruct_getu32(t, &tdev) < 0 ||
 580         !pa_tagstruct_eof(t)) {
 581         s->context->error = PA_ERROR_PROTOCOL;
 582         context_dead(s->context);
 583         return;
 584     }
 585
 586     create_stream(s, tdev);
 587 }
 588
 589 static void lookup_device(struct pa_stream *s, const char *tdev) {
 590     struct pa_tagstruct *t;
 591     uint32_t tag;
 592     assert(s);
 593
 594     s->state = STREAM_LOOKING_UP;
 595
 596     t = pa_tagstruct_new(NULL, 0);
 597     assert(t);
 598
 599     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_LOOKUP_SINK : PA_COMMAND_LOOKUP_SOURCE);
 600     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 601     pa_tagstruct_puts(t, tdev);
 602
 603     pa_pstream_send_tagstruct(s->context->pstream, t);
 604     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, lookup_device_callback, s);
 605 }
 606
 607 struct pa_stream* pa_stream_new(
 608     struct pa_context *c,
 609     enum pa_stream_direction dir,
 610     const char *dev,
 611     const char *name,
 612     const struct pa_sample_spec *ss,
 613     const struct pa_buffer_attr *attr,
 614     void (*complete) (struct pa_stream*s, int success, void *userdata),
 615     void *userdata) {
 616
 617     struct pa_stream *s;
 618
 619     assert(c && name && ss && c->state == CONTEXT_READY);
 620
 621     s = malloc(sizeof(struct pa_stream));
 622     assert(s);
 623     s->context = c;
 624
 625     s->read_callback = NULL;
 626     s->read_userdata = NULL;
 627     s->write_callback = NULL;
 628     s->write_userdata = NULL;
 629     s->die_callback = NULL;
 630     s->die_userdata = NULL;
 631     s->create_complete_callback = complete;
 632     s->create_complete_userdata = NULL;
 633     s->get_latency_callback = NULL;
 634     s->get_latency_userdata = NULL;
 635
 636     s->name = strdup(name);
 637     s->state = STREAM_CREATING;
 638     s->requested_bytes = 0;
 639     s->channel = 0;
 640     s->channel_valid = 0;
 641     s->device_index = (uint32_t) -1;
 642     s->direction = dir;
 643     s->sample_spec = *ss;
 644     if (attr)
 645         s->buffer_attr = *attr;
 646     else {
 647         s->buffer_attr.maxlength = DEFAULT_MAXLENGTH;
 648         s->buffer_attr.tlength = DEFAULT_TLENGTH;
 649         s->buffer_attr.prebuf = DEFAULT_PREBUF;
 650         s->buffer_attr.minreq = DEFAULT_MINREQ;
 651         s->buffer_attr.fragsize = DEFAULT_FRAGSIZE;
 652     }
 653
 654     s->next = c->first_stream;
 655     if (s->next)
 656         s->next->previous = s;
 657     s->previous = NULL;
 658     c->first_stream = s;
 659
 660     if (dev)
 661         lookup_device(s, dev);
 662     else
 663         create_stream(s, (uint32_t) -1);
 664
 665     return s;
 666 }
 667
 668 void pa_stream_free(struct pa_stream *s) {
 669     assert(s && s->context);
 670
 671     if (s->context->pdispatch)
 672         pa_pdispatch_unregister_reply(s->context->pdispatch, s);
 673
 674     free(s->name);
 675
 676     if (s->channel_valid && s->context->state == CONTEXT_READY) {
 677         struct pa_tagstruct *t = pa_tagstruct_new(NULL, 0);
 678         assert(t);
 679
 680         pa_tagstruct_putu32(t, PA_COMMAND_DELETE_PLAYBACK_STREAM);
 681         pa_tagstruct_putu32(t, s->context->ctag++);
 682         pa_tagstruct_putu32(t, s->channel);
 683         pa_pstream_send_tagstruct(s->context->pstream, t);
 684     }
 685
 686     if (s->channel_valid)
 687         pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 688
 689     if (s->next)
 690         s->next->previous = s->previous;
 691     if (s->previous)
 692         s->previous->next = s->next;
 693     else
 694         s->context->first_stream = s->next;
 695
 696     free(s);
 697 }
 698
 699 void pa_stream_set_write_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, size_t length, void *userdata), void *userdata) {
 700     assert(s && cb);
 701     s->write_callback = cb;
 702     s->write_userdata = userdata;
 703 }
 704
 705 void pa_stream_write(struct pa_stream *s, const void *data, size_t length) {
 706     struct pa_memchunk chunk;
 707     assert(s && s->context && data && length && s->state == STREAM_READY);
 708
 709     chunk.memblock = pa_memblock_new(length);
 710     assert(chunk.memblock && chunk.memblock->data);
 711     memcpy(chunk.memblock->data, data, length);
 712     chunk.index = 0;
 713     chunk.length = length;
 714
 715     pa_pstream_send_memblock(s->context->pstream, s->channel, 0, &chunk);
 716     pa_memblock_unref(chunk.memblock);
 717
 718     /*fprintf(stderr, "Sent %u bytes\n", length);*/
 719
 720     if (length < s->requested_bytes)
 721         s->requested_bytes -= length;
 722     else
 723         s->requested_bytes = 0;
 724 }
 725
 726 size_t pa_stream_writable_size(struct pa_stream *s) {
 727     assert(s && s->state == STREAM_READY);
 728     return s->requested_bytes;
 729 }
 730
 731 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) {
 732     assert(s && cb);
 733     s->read_callback = cb;
 734     s->read_userdata = userdata;
 735 }
 736
 737 int pa_stream_is_dead(struct pa_stream *s) {
 738     return s->state == STREAM_DEAD;
 739 }
 740
 741 int pa_stream_is_ready(struct pa_stream*s) {
 742     return s->state == STREAM_READY;
 743 }
 744
 745 void pa_stream_set_die_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata) {
 746     assert(s);
 747     s->die_callback = cb;
 748     s->die_userdata = userdata;
 749 }
 750
 751 int pa_context_is_pending(struct pa_context *c) {
 752     assert(c);
 753
 754     if (c->state != CONTEXT_READY)
 755         return 0;
 756
 757     return pa_pstream_is_pending(c->pstream) || pa_pdispatch_is_pending(c->pdispatch);
 758 }
 759
 760 struct pa_context* pa_stream_get_context(struct pa_stream *p) {
 761     assert(p);
 762     return p->context;
 763 }
 764
 765 static void set_dispatch_callbacks(struct pa_context *c);
 766
 767 static void pdispatch_drain_callback(struct pa_pdispatch*pd, void *userdata) {
 768     set_dispatch_callbacks(userdata);
 769 }
 770
 771 static void pstream_drain_callback(struct pa_pstream *s, void *userdata) {
 772     set_dispatch_callbacks(userdata);
 773 }
 774
 775 static void set_dispatch_callbacks(struct pa_context *c) {
 776     assert(c && c->state == CONTEXT_READY);
 777
 778     pa_pstream_set_drain_callback(c->pstream, NULL, NULL);
 779     pa_pdispatch_set_drain_callback(c->pdispatch, NULL, NULL);
 780
 781     if (pa_pdispatch_is_pending(c->pdispatch)) {
 782         pa_pdispatch_set_drain_callback(c->pdispatch, pdispatch_drain_callback, c);
 783         return;
 784     }
 785
 786     if (pa_pstream_is_pending(c->pstream)) {
 787         pa_pstream_set_drain_callback(c->pstream, pstream_drain_callback, c);
 788         return;
 789     }
 790
 791     assert(c->drain_complete_callback);
 792     c->drain_complete_callback(c, c->drain_complete_userdata);
 793 }
 794
 795 int pa_context_drain(
 796     struct pa_context *c,
 797     void (*complete) (struct pa_context*c, void *userdata),
 798     void *userdata) {
 799
 800     assert(c && c->state == CONTEXT_READY);
 801
 802     if (complete == NULL) {
 803         c->drain_complete_callback = NULL;
 804         pa_pstream_set_drain_callback(c->pstream, NULL, NULL);
 805         pa_pdispatch_set_drain_callback(c->pdispatch, NULL, NULL);
 806         return 0;
 807     }
 808
 809     if (!pa_context_is_pending(c))
 810         return -1;
 811
 812     c->drain_complete_callback = complete;
 813     c->drain_complete_userdata = userdata;
 814
 815     set_dispatch_callbacks(c);
 816
 817     return 0;
 818 }
 819
 820 static void stream_drain_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 821     struct pa_stream *s = userdata;
 822     assert(pd && s);
 823
 824     if (command != PA_COMMAND_REPLY) {
 825         if (handle_error(s->context, command, t) < 0) {
 826             context_dead(s->context);
 827             return;
 828         }
 829
 830         stream_dead(s);
 831         return;
 832     }
 833
 834     if (s->state != STREAM_READY)
 835         return;
 836
 837     if (!pa_tagstruct_eof(t)) {
 838         s->context->error = PA_ERROR_PROTOCOL;
 839         context_dead(s->context);
 840         return;
 841     }
 842
 843     if (s->drain_complete_callback) {
 844         void (*temp) (struct pa_stream*s, void *userdata) = s->drain_complete_callback;
 845         s->drain_complete_callback = NULL;
 846         temp(s, s->drain_complete_userdata);
 847     }
 848 }
 849
 850
 851 void pa_stream_drain(struct pa_stream *s, void (*complete) (struct pa_stream*s, void *userdata), void *userdata) {
 852     struct pa_tagstruct *t;
 853     uint32_t tag;
 854     assert(s && s->state == STREAM_READY);
 855
 856     if (!complete) {
 857         s->drain_complete_callback = NULL;
 858         return;
 859     }
 860
 861     s->drain_complete_callback = complete;
 862     s->drain_complete_userdata = userdata;
 863
 864     t = pa_tagstruct_new(NULL, 0);
 865     assert(t);
 866
 867     pa_tagstruct_putu32(t, PA_COMMAND_DRAIN_PLAYBACK_STREAM);
 868     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 869     pa_tagstruct_putu32(t, s->channel);
 870     pa_pstream_send_tagstruct(s->context->pstream, t);
 871     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_drain_callback, s);
 872 }
 873
 874 void pa_context_exit(struct pa_context *c) {
 875     struct pa_tagstruct *t;
 876     t = pa_tagstruct_new(NULL, 0);
 877     assert(t);
 878     pa_tagstruct_putu32(t, PA_COMMAND_EXIT);
 879     pa_tagstruct_putu32(t, c->ctag++);
 880     pa_pstream_send_tagstruct(c->pstream, t);
 881 }
 882
 883 static void context_stat_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 884     struct pa_context *c = userdata;
 885     uint32_t total, count;
 886     assert(pd && c);
 887
 888     if (command != PA_COMMAND_REPLY) {
 889         if (handle_error(c, command, t) < 0) {
 890             context_dead(c);
 891             return;
 892         }
 893
 894         if (c->stat_callback)
 895             c->stat_callback(c, (uint32_t) -1, (uint32_t) -1, c->stat_userdata);
 896         return;
 897     }
 898
 899     if (pa_tagstruct_getu32(t, &count) < 0 ||
 900         pa_tagstruct_getu32(t, &total) < 0 ||
 901         !pa_tagstruct_eof(t)) {
 902         c->error = PA_ERROR_PROTOCOL;
 903         context_dead(c);
 904         return;
 905     }
 906
 907     if (c->stat_callback)
 908         c->stat_callback(c, count, total, c->stat_userdata);
 909 }
 910
 911 void pa_context_stat(struct pa_context *c, void (*cb)(struct pa_context *c, uint32_t count, uint32_t total, void *userdata), void *userdata) {
 912     uint32_t tag;
 913     struct pa_tagstruct *t;
 914
 915     c->stat_callback = cb;
 916     c->stat_userdata = userdata;
 917
 918     if (cb == NULL)
 919         return;
 920
 921     t = pa_tagstruct_new(NULL, 0);
 922     assert(t);
 923     pa_tagstruct_putu32(t, PA_COMMAND_STAT);
 924     pa_tagstruct_putu32(t, tag = c->ctag++);
 925     pa_pstream_send_tagstruct(c->pstream, t);
 926     pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_stat_callback, c);
 927 }
 928
 929 static void stream_get_latency_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 930     struct pa_stream *s = userdata;
 931     uint32_t latency;
 932     assert(pd && s);
 933
 934     if (command != PA_COMMAND_REPLY) {
 935         if (handle_error(s->context, command, t) < 0) {
 936             context_dead(s->context);
 937             return;
 938         }
 939
 940         if (s->get_latency_callback)
 941             s->get_latency_callback(s, (uint32_t) -1, s->get_latency_userdata);
 942         return;
 943     }
 944
 945     if (pa_tagstruct_getu32(t, &latency) < 0 ||
 946         !pa_tagstruct_eof(t)) {
 947         s->context->error = PA_ERROR_PROTOCOL;
 948         context_dead(s->context);
 949         return;
 950     }
 951
 952     if (s->get_latency_callback)
 953         s->get_latency_callback(s, latency, s->get_latency_userdata);
 954 }
 955
 956 void pa_stream_get_latency(struct pa_stream *p, void (*cb)(struct pa_stream *p, uint32_t latency, void *userdata), void *userdata) {
 957     uint32_t tag;
 958     struct pa_tagstruct *t;
 959
 960     p->get_latency_callback = cb;
 961     p->get_latency_userdata = userdata;
 962
 963     if (cb == NULL)
 964         return;
 965
 966     t = pa_tagstruct_new(NULL, 0);
 967     assert(t);
 968     pa_tagstruct_putu32(t, PA_COMMAND_GET_PLAYBACK_LATENCY);
 969     pa_tagstruct_putu32(t, tag = p->context->ctag++);
 970     pa_tagstruct_putu32(t, p->channel);
 971     pa_pstream_send_tagstruct(p->context->pstream, t);
 972     pa_pdispatch_register_reply(p->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, p);
 973 }