code.delx.au - pulseaudio/blob - polyp/polyplib-stream.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 <assert.h>
  27 #include <string.h>
  28 #include <stdio.h>
  29
  30 #include "polyplib-internal.h"
  31 #include "xmalloc.h"
  32 #include "pstream-util.h"
  33
  34 struct pa_stream *pa_stream_new(struct pa_context *c, const char *name, const struct pa_sample_spec *ss) {
  35     struct pa_stream *s;
  36     assert(c && ss);
  37
  38     s = pa_xmalloc(sizeof(struct pa_stream));
  39     s->ref = 1;
  40     s->context = c;
  41
  42     s->read_callback = NULL;
  43     s->read_userdata = NULL;
  44     s->write_callback = NULL;
  45     s->write_userdata = NULL;
  46     s->state_callback = NULL;
  47     s->state_userdata = NULL;
  48
  49     s->state = PA_STREAM_NODIRECTION;
  50     s->name = pa_xstrdup(name);
  51     s->sample_spec = *ss;
  52     s->channel = 0;
  53     s->channel_valid = 0;
  54     s->device_index = PA_INVALID_INDEX;
  55     s->requested_bytes = 0;
  56     s->state = PA_STREAM_DISCONNECTED;
  57     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
  58
  59     PA_LLIST_PREPEND(struct pa_stream, c->streams, s);
  60
  61     return pa_stream_ref(s);
  62 }
  63
  64 static void stream_free(struct pa_stream *s) {
  65     assert(s);
  66     pa_xfree(s->name);
  67     pa_xfree(s);
  68 }
  69
  70 void pa_stream_unref(struct pa_stream *s) {
  71     assert(s && s->ref >= 1);
  72
  73     if (--(s->ref) == 0)
  74         stream_free(s);
  75 }
  76
  77 struct pa_stream* pa_stream_ref(struct pa_stream *s) {
  78     assert(s && s->ref >= 1);
  79     s->ref++;
  80     return s;
  81 }
  82
  83 enum pa_stream_state pa_stream_get_state(struct pa_stream *s) {
  84     assert(s && s->ref >= 1);
  85     return s->state;
  86 }
  87
  88 struct pa_context* pa_stream_get_context(struct pa_stream *s) {
  89     assert(s && s->ref >= 1);
  90     return s->context;
  91 }
  92
  93 uint32_t pa_stream_get_index(struct pa_stream *s) {
  94     assert(s && s->ref >= 1);
  95     return s->device_index;
  96 }
  97
  98 void pa_stream_set_state(struct pa_stream *s, enum pa_stream_state st) {
  99     assert(s && s->ref >= 1);
 100
 101     if (s->state == st)
 102         return;
 103
 104     pa_stream_ref(s);
 105
 106     s->state = st;
 107
 108     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
 109         if (s->channel_valid)
 110             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 111
 112         PA_LLIST_REMOVE(struct pa_stream, s->context->streams, s);
 113         pa_stream_unref(s);
 114     }
 115
 116     if (s->state_callback)
 117         s->state_callback(s, s->state_userdata);
 118
 119     pa_stream_unref(s);
 120 }
 121
 122 void pa_command_stream_killed(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 123     struct pa_context *c = userdata;
 124     struct pa_stream *s;
 125     uint32_t channel;
 126     assert(pd && (command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED) && t && c);
 127
 128     pa_context_ref(c);
 129
 130     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 131         !pa_tagstruct_eof(t)) {
 132         pa_context_fail(c, PA_ERROR_PROTOCOL);
 133         goto finish;
 134     }
 135
 136     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 137         goto finish;
 138
 139     c->error = PA_ERROR_KILLED;
 140     pa_stream_set_state(s, PA_STREAM_FAILED);
 141
 142 finish:
 143     pa_context_unref(c);
 144 }
 145
 146 void pa_command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 147     struct pa_stream *s;
 148     struct pa_context *c = userdata;
 149     uint32_t bytes, channel;
 150     assert(pd && command == PA_COMMAND_REQUEST && t && c);
 151
 152     pa_context_ref(c);
 153
 154     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 155         pa_tagstruct_getu32(t, &bytes) < 0 ||
 156         !pa_tagstruct_eof(t)) {
 157         pa_context_fail(c, PA_ERROR_PROTOCOL);
 158         goto finish;
 159     }
 160
 161     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 162         goto finish;
 163
 164     if (s->state != PA_STREAM_READY)
 165         goto finish;
 166
 167     pa_stream_ref(s);
 168
 169     s->requested_bytes += bytes;
 170
 171     fprintf(stderr, "total req: %u (%u)\n", s->requested_bytes, bytes);
 172
 173     if (s->requested_bytes && s->write_callback)
 174         s->write_callback(s, s->requested_bytes, s->write_userdata);
 175
 176     pa_stream_unref(s);
 177
 178 finish:
 179     pa_context_unref(c);
 180 }
 181
 182 void pa_create_stream_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 183     struct pa_stream *s = userdata;
 184     assert(pd && s && s->state == PA_STREAM_CREATING);
 185
 186     pa_stream_ref(s);
 187
 188     if (command != PA_COMMAND_REPLY) {
 189         if (pa_context_handle_error(s->context, command, t) < 0)
 190             goto finish;
 191
 192         pa_stream_set_state(s, PA_STREAM_FAILED);
 193         goto finish;
 194     }
 195
 196     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 197         ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
 198         !pa_tagstruct_eof(t)) {
 199         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 200         goto finish;
 201     }
 202
 203     s->channel_valid = 1;
 204     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 205     pa_stream_set_state(s, PA_STREAM_READY);
 206
 207 finish:
 208     pa_stream_unref(s);
 209 }
 210
 211 static void create_stream(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 212     struct pa_tagstruct *t;
 213     uint32_t tag;
 214     assert(s && s->ref >= 1 && s->state == PA_STREAM_DISCONNECTED);
 215
 216     pa_stream_ref(s);
 217
 218     if (attr)
 219         s->buffer_attr = *attr;
 220     else {
 221         s->buffer_attr.maxlength = DEFAULT_MAXLENGTH;
 222         s->buffer_attr.tlength = DEFAULT_TLENGTH;
 223         s->buffer_attr.prebuf = DEFAULT_PREBUF;
 224         s->buffer_attr.minreq = DEFAULT_MINREQ;
 225         s->buffer_attr.fragsize = DEFAULT_FRAGSIZE;
 226     }
 227
 228     pa_stream_set_state(s, PA_STREAM_CREATING);
 229
 230     t = pa_tagstruct_new(NULL, 0);
 231     assert(t);
 232
 233     if (!dev) {
 234         if (s->direction == PA_STREAM_PLAYBACK)
 235             dev = getenv(ENV_DEFAULT_SINK);
 236         else
 237             dev = getenv(ENV_DEFAULT_SOURCE);
 238     }
 239
 240     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM);
 241     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 242     pa_tagstruct_puts(t, s->name);
 243     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
 244     pa_tagstruct_putu32(t, (uint32_t) -1);
 245     pa_tagstruct_puts(t, dev ? dev : "");
 246     pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
 247     if (s->direction == PA_STREAM_PLAYBACK) {
 248         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
 249         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
 250         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
 251     } else
 252         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 253
 254     pa_pstream_send_tagstruct(s->context->pstream, t);
 255     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
 256
 257     pa_stream_unref(s);
 258 }
 259
 260 void pa_stream_connect_playback(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 261     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 262     s->direction = PA_STREAM_PLAYBACK;
 263     create_stream(s, dev, attr);
 264 }
 265
 266 void pa_stream_connect_record(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 267     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 268     s->direction = PA_STREAM_RECORD;
 269     create_stream(s, dev, attr);
 270 }
 271
 272 void pa_stream_write(struct pa_stream *s, const void *data, size_t length, void (*free_cb)(void *p), size_t delta) {
 273     struct pa_memchunk chunk;
 274     assert(s && s->context && data && length && s->state == PA_STREAM_READY && s->ref >= 1);
 275
 276     if (free_cb) {
 277         chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, s->context->memblock_stat);
 278         assert(chunk.memblock && chunk.memblock->data);
 279     } else {
 280         chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
 281         assert(chunk.memblock && chunk.memblock->data);
 282         memcpy(chunk.memblock->data, data, length);
 283     }
 284     chunk.index = 0;
 285     chunk.length = length;
 286
 287     pa_pstream_send_memblock(s->context->pstream, s->channel, delta, &chunk);
 288     pa_memblock_unref(chunk.memblock);
 289
 290     if (length < s->requested_bytes)
 291         s->requested_bytes -= length;
 292     else
 293         s->requested_bytes = 0;
 294 }
 295
 296 size_t pa_stream_writable_size(struct pa_stream *s) {
 297     assert(s && s->state == PA_STREAM_READY && s->ref >= 1);
 298     return s->requested_bytes;
 299 }
 300
 301 struct pa_operation * pa_stream_drain(struct pa_stream *s, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata) {
 302     struct pa_operation *o;
 303     struct pa_tagstruct *t;
 304     uint32_t tag;
 305     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 306
 307     o = pa_operation_new(s->context, s);
 308     assert(o);
 309     o->callback = cb;
 310     o->userdata = userdata;
 311
 312     t = pa_tagstruct_new(NULL, 0);
 313     assert(t);
 314     pa_tagstruct_putu32(t, PA_COMMAND_DRAIN_PLAYBACK_STREAM);
 315     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 316     pa_tagstruct_putu32(t, s->channel);
 317     pa_pstream_send_tagstruct(s->context->pstream, t);
 318     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 319
 320     return pa_operation_ref(o);
 321 }
 322
 323 static void stream_get_latency_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 324     struct pa_operation *o = userdata;
 325     struct pa_latency_info i, *p = NULL;
 326     assert(pd && o && o->stream && o->context);
 327
 328     if (command != PA_COMMAND_REPLY) {
 329         if (pa_context_handle_error(o->context, command, t) < 0)
 330             goto finish;
 331
 332     } else if (pa_tagstruct_getu32(t, &i.buffer_usec) < 0 ||
 333                pa_tagstruct_getu32(t, &i.sink_usec) < 0 ||
 334                pa_tagstruct_getu32(t, &i.playing) < 0 ||
 335                pa_tagstruct_getu32(t, &i.queue_length) < 0 ||
 336                !pa_tagstruct_eof(t)) {
 337         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 338         goto finish;
 339     } else
 340         p = &i;
 341
 342     if (o->callback) {
 343         void (*cb)(struct pa_stream *s, const struct pa_latency_info *i, void *userdata) = o->callback;
 344         cb(o->stream, p, o->userdata);
 345     }
 346
 347 finish:
 348     pa_operation_done(o);
 349     pa_operation_unref(o);
 350 }
 351
 352 struct pa_operation* pa_stream_get_latency(struct pa_stream *s, void (*cb)(struct pa_stream *p, const struct pa_latency_info*i, void *userdata), void *userdata) {
 353     uint32_t tag;
 354     struct pa_operation *o;
 355     struct pa_tagstruct *t;
 356
 357     o = pa_operation_new(s->context, s);
 358     assert(o);
 359     o->callback = cb;
 360     o->userdata = userdata;
 361
 362     t = pa_tagstruct_new(NULL, 0);
 363     assert(t);
 364     pa_tagstruct_putu32(t, PA_COMMAND_GET_PLAYBACK_LATENCY);
 365     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 366     pa_tagstruct_putu32(t, s->channel);
 367     pa_pstream_send_tagstruct(s->context->pstream, t);
 368     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, o);
 369
 370     return pa_operation_ref(o);
 371 }
 372
 373 void pa_stream_disconnect_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 374     struct pa_stream *s = userdata;
 375     assert(pd && s && s->ref >= 1);
 376
 377     pa_stream_ref(s);
 378
 379     if (command != PA_COMMAND_REPLY) {
 380         if (pa_context_handle_error(s->context, command, t) < 0)
 381             goto finish;
 382
 383         pa_stream_set_state(s, PA_STREAM_FAILED);
 384         goto finish;
 385     } else if (!pa_tagstruct_eof(t)) {
 386         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 387         goto finish;
 388     }
 389
 390     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 391
 392 finish:
 393     pa_stream_unref(s);
 394 }
 395
 396 void pa_stream_disconnect(struct pa_stream *s) {
 397     struct pa_tagstruct *t;
 398     uint32_t tag;
 399     assert(s && s->ref >= 1);
 400
 401     if (!s->channel_valid || !s->context->state == PA_CONTEXT_READY)
 402         return;
 403
 404     pa_stream_ref(s);
 405
 406     t = pa_tagstruct_new(NULL, 0);
 407     assert(t);
 408
 409     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 410                         (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM));
 411     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 412     pa_tagstruct_putu32(t, s->channel);
 413     pa_pstream_send_tagstruct(s->context->pstream, t);
 414     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
 415
 416     pa_stream_unref(s);
 417 }
 418
 419 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) {
 420     assert(s && s->ref >= 1);
 421     s->read_callback = cb;
 422     s->read_userdata = userdata;
 423 }
 424
 425 void pa_stream_set_write_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, size_t length, void *userdata), void *userdata) {
 426     assert(s && s->ref >= 1);
 427     s->write_callback = cb;
 428     s->write_userdata = userdata;
 429 }
 430
 431 void pa_stream_set_state_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata) {
 432     assert(s && s->ref >= 1);
 433     s->state_callback = cb;
 434     s->state_userdata = userdata;
 435 }
 436
 437 void pa_stream_simple_ack_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 438     struct pa_operation *o = userdata;
 439     int success = 1;
 440     assert(pd && o && o->context && o->ref >= 1);
 441
 442     if (command != PA_COMMAND_REPLY) {
 443         if (pa_context_handle_error(o->context, command, t) < 0)
 444             goto finish;
 445
 446         success = 0;
 447     } else if (!pa_tagstruct_eof(t)) {
 448         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 449         goto finish;
 450     }
 451
 452     if (o->callback) {
 453         void (*cb)(struct pa_stream *s, int success, void *userdata) = o->callback;
 454         cb(o->stream, success, o->userdata);
 455     }
 456
 457 finish:
 458     pa_operation_done(o);
 459     pa_operation_unref(o);
 460 }
 461
 462 struct pa_operation* pa_stream_cork(struct pa_stream *s, int b, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata) {
 463     struct pa_operation *o;
 464     struct pa_tagstruct *t;
 465     uint32_t tag;
 466     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 467
 468     o = pa_operation_new(s->context, s);
 469     assert(o);
 470     o->callback = cb;
 471     o->userdata = userdata;
 472
 473     t = pa_tagstruct_new(NULL, 0);
 474     assert(t);
 475     pa_tagstruct_putu32(t, PA_COMMAND_CORK_PLAYBACK_STREAM);
 476     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 477     pa_tagstruct_putu32(t, s->channel);
 478     pa_tagstruct_putu32(t, !!b);
 479     pa_pstream_send_tagstruct(s->context->pstream, t);
 480     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 481
 482     return pa_operation_ref(o);
 483 }
 484
 485 struct pa_operation* pa_stream_send_simple_command(struct pa_stream *s, uint32_t command, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 486     struct pa_tagstruct *t;
 487     struct pa_operation *o;
 488     uint32_t tag;
 489     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 490
 491     o = pa_operation_new(s->context, s);
 492     o->callback = cb;
 493     o->userdata = userdata;
 494
 495     t = pa_tagstruct_new(NULL, 0);
 496     pa_tagstruct_putu32(t, command);
 497     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 498     pa_tagstruct_putu32(t, s->channel);
 499     pa_pstream_send_tagstruct(s->context->pstream, t);
 500     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 501
 502     return pa_operation_ref(o);
 503 }
 504
 505
 506 struct pa_operation* pa_stream_flush(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 507     return pa_stream_send_simple_command(s, PA_COMMAND_FLUSH_PLAYBACK_STREAM, cb, userdata);
 508 }
 509
 510 struct pa_operation* pa_stream_trigger(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 511     return pa_stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata);
 512 }