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     assert(c && ss);
  36     struct pa_stream *s;
  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     if (s->requested_bytes && s->write_callback)
 172         s->write_callback(s, s->requested_bytes, s->write_userdata);
 173
 174     pa_stream_unref(s);
 175
 176 finish:
 177     pa_context_unref(c);
 178 }
 179
 180 void pa_create_stream_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 181     struct pa_stream *s = userdata;
 182     assert(pd && s && s->state == PA_STREAM_CREATING);
 183
 184     pa_stream_ref(s);
 185
 186     if (command != PA_COMMAND_REPLY) {
 187         if (pa_context_handle_error(s->context, command, t) < 0)
 188             goto finish;
 189
 190         pa_stream_set_state(s, PA_STREAM_FAILED);
 191         goto finish;
 192     }
 193
 194     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 195         ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
 196         !pa_tagstruct_eof(t)) {
 197         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 198         goto finish;
 199     }
 200
 201     s->channel_valid = 1;
 202     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 203     pa_stream_set_state(s, PA_STREAM_READY);
 204
 205 finish:
 206     pa_stream_unref(s);
 207 }
 208
 209 static void create_stream(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 210     struct pa_tagstruct *t;
 211     uint32_t tag;
 212     assert(s && s->ref >= 1 && s->state == PA_STREAM_DISCONNECTED);
 213
 214     pa_stream_ref(s);
 215
 216     if (attr)
 217         s->buffer_attr = *attr;
 218     else {
 219         s->buffer_attr.maxlength = DEFAULT_MAXLENGTH;
 220         s->buffer_attr.tlength = DEFAULT_TLENGTH;
 221         s->buffer_attr.prebuf = DEFAULT_PREBUF;
 222         s->buffer_attr.minreq = DEFAULT_MINREQ;
 223         s->buffer_attr.fragsize = DEFAULT_FRAGSIZE;
 224     }
 225
 226     pa_stream_set_state(s, PA_STREAM_CREATING);
 227
 228     t = pa_tagstruct_new(NULL, 0);
 229     assert(t);
 230
 231     if (!dev) {
 232         if (s->direction == PA_STREAM_PLAYBACK)
 233             dev = getenv(ENV_DEFAULT_SINK);
 234         else
 235             dev = getenv(ENV_DEFAULT_SOURCE);
 236     }
 237
 238     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM);
 239     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 240     pa_tagstruct_puts(t, s->name);
 241     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
 242     pa_tagstruct_putu32(t, (uint32_t) -1);
 243     pa_tagstruct_puts(t, dev ? dev : "");
 244     pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
 245     if (s->direction == PA_STREAM_PLAYBACK) {
 246         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
 247         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
 248         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
 249     } else
 250         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 251
 252     pa_pstream_send_tagstruct(s->context->pstream, t);
 253     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
 254
 255     pa_stream_unref(s);
 256 }
 257
 258 void pa_stream_connect_playback(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 259     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 260     s->direction = PA_STREAM_PLAYBACK;
 261     create_stream(s, dev, attr);
 262 }
 263
 264 void pa_stream_connect_record(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr) {
 265     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 266     s->direction = PA_STREAM_RECORD;
 267     create_stream(s, dev, attr);
 268 }
 269
 270 void pa_stream_write(struct pa_stream *s, const void *data, size_t length, void (*free_cb)(void *p)) {
 271     struct pa_memchunk chunk;
 272     assert(s && s->context && data && length && s->state == PA_STREAM_READY && s->ref >= 1);
 273
 274     if (free_cb) {
 275         chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, s->context->memblock_stat);
 276         assert(chunk.memblock && chunk.memblock->data);
 277     } else {
 278         chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
 279         assert(chunk.memblock && chunk.memblock->data);
 280         memcpy(chunk.memblock->data, data, length);
 281     }
 282     chunk.index = 0;
 283     chunk.length = length;
 284
 285     pa_pstream_send_memblock(s->context->pstream, s->channel, 0, &chunk);
 286     pa_memblock_unref(chunk.memblock);
 287
 288     if (length < s->requested_bytes)
 289         s->requested_bytes -= length;
 290     else
 291         s->requested_bytes = 0;
 292 }
 293
 294 size_t pa_stream_writable_size(struct pa_stream *s) {
 295     assert(s && s->state == PA_STREAM_READY && s->ref >= 1);
 296     return s->requested_bytes;
 297 }
 298
 299 struct pa_operation * pa_stream_drain(struct pa_stream *s, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata) {
 300     struct pa_operation *o;
 301     struct pa_tagstruct *t;
 302     uint32_t tag;
 303     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 304
 305     o = pa_operation_new(s->context, s);
 306     assert(o);
 307     o->callback = cb;
 308     o->userdata = userdata;
 309
 310     t = pa_tagstruct_new(NULL, 0);
 311     assert(t);
 312     pa_tagstruct_putu32(t, PA_COMMAND_DRAIN_PLAYBACK_STREAM);
 313     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 314     pa_tagstruct_putu32(t, s->channel);
 315     pa_pstream_send_tagstruct(s->context->pstream, t);
 316     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 317
 318     return pa_operation_ref(o);
 319 }
 320
 321 static void stream_get_latency_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 322     struct pa_operation *o = userdata;
 323     uint32_t latency;
 324     assert(pd && o && o->stream && o->context);
 325
 326     if (command != PA_COMMAND_REPLY) {
 327         if (pa_context_handle_error(o->context, command, t) < 0)
 328             goto finish;
 329
 330         latency = (uint32_t) -1;
 331     } else if (pa_tagstruct_getu32(t, &latency) < 0 || !pa_tagstruct_eof(t)) {
 332         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 333         goto finish;
 334     }
 335
 336     if (o->callback) {
 337         void (*cb)(struct pa_stream *s, uint32_t latency, void *userdata) = o->callback;
 338         cb(o->stream, latency, o->userdata);
 339     }
 340
 341 finish:
 342     pa_operation_done(o);
 343     pa_operation_unref(o);
 344 }
 345
 346 struct pa_operation* pa_stream_get_latency(struct pa_stream *s, void (*cb)(struct pa_stream *p, uint32_t latency, void *userdata), void *userdata) {
 347     uint32_t tag;
 348     struct pa_operation *o;
 349     struct pa_tagstruct *t;
 350
 351     o = pa_operation_new(s->context, s);
 352     assert(o);
 353     o->callback = cb;
 354     o->userdata = userdata;
 355
 356     t = pa_tagstruct_new(NULL, 0);
 357     assert(t);
 358     pa_tagstruct_putu32(t, PA_COMMAND_GET_PLAYBACK_LATENCY);
 359     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 360     pa_tagstruct_putu32(t, s->channel);
 361     pa_pstream_send_tagstruct(s->context->pstream, t);
 362     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, o);
 363
 364     return pa_operation_ref(o);
 365 }
 366
 367 void pa_stream_disconnect_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 368     struct pa_stream *s = userdata;
 369     assert(pd && s && s->ref >= 1);
 370
 371     pa_stream_ref(s);
 372
 373     if (command != PA_COMMAND_REPLY) {
 374         if (pa_context_handle_error(s->context, command, t) < 0)
 375             goto finish;
 376
 377         pa_stream_set_state(s, PA_STREAM_FAILED);
 378         goto finish;
 379     } else if (!pa_tagstruct_eof(t)) {
 380         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 381         goto finish;
 382     }
 383
 384     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 385
 386 finish:
 387     pa_stream_unref(s);
 388 }
 389
 390 void pa_stream_disconnect(struct pa_stream *s) {
 391     struct pa_tagstruct *t;
 392     uint32_t tag;
 393     assert(s && s->ref >= 1);
 394
 395     if (!s->channel_valid || !s->context->state == PA_CONTEXT_READY)
 396         return;
 397
 398     pa_stream_ref(s);
 399
 400     t = pa_tagstruct_new(NULL, 0);
 401     assert(t);
 402
 403     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 404                         (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM));
 405     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 406     pa_tagstruct_putu32(t, s->channel);
 407     pa_pstream_send_tagstruct(s->context->pstream, t);
 408     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
 409
 410     pa_stream_unref(s);
 411 }
 412
 413 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) {
 414     assert(s && s->ref >= 1);
 415     s->read_callback = cb;
 416     s->read_userdata = userdata;
 417 }
 418
 419 void pa_stream_set_write_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, size_t length, void *userdata), void *userdata) {
 420     assert(s && s->ref >= 1);
 421     s->write_callback = cb;
 422     s->write_userdata = userdata;
 423 }
 424
 425 void pa_stream_set_state_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata) {
 426     assert(s && s->ref >= 1);
 427     s->state_callback = cb;
 428     s->state_userdata = userdata;
 429 }
 430
 431 void pa_stream_simple_ack_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 432     struct pa_operation *o = userdata;
 433     int success = 1;
 434     assert(pd && o && o->context && o->ref >= 1);
 435
 436     if (command != PA_COMMAND_REPLY) {
 437         if (pa_context_handle_error(o->context, command, t) < 0)
 438             goto finish;
 439
 440         success = 0;
 441     } else if (!pa_tagstruct_eof(t)) {
 442         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 443         goto finish;
 444     }
 445
 446     if (o->callback) {
 447         void (*cb)(struct pa_stream *s, int success, void *userdata) = o->callback;
 448         cb(o->stream, success, o->userdata);
 449     }
 450
 451 finish:
 452     pa_operation_done(o);
 453     pa_operation_unref(o);
 454 }