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 Lesser 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 Lesser 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 #include <string.h>
  30
  31 #include "polyplib-internal.h"
  32 #include "xmalloc.h"
  33 #include "pstream-util.h"
  34 #include "util.h"
  35 #include "log.h"
  36
  37 #define LATENCY_IPOL_INTERVAL_USEC (10000L)
  38
  39 struct pa_stream *pa_stream_new(struct pa_context *c, const char *name, const struct pa_sample_spec *ss) {
  40     struct pa_stream *s;
  41     assert(c && ss);
  42
  43     s = pa_xmalloc(sizeof(struct pa_stream));
  44     s->ref = 1;
  45     s->context = c;
  46     s->mainloop = c->mainloop;
  47
  48     s->read_callback = NULL;
  49     s->read_userdata = NULL;
  50     s->write_callback = NULL;
  51     s->write_userdata = NULL;
  52     s->state_callback = NULL;
  53     s->state_userdata = NULL;
  54
  55     s->direction = PA_STREAM_NODIRECTION;
  56     s->name = pa_xstrdup(name);
  57     s->sample_spec = *ss;
  58     s->channel = 0;
  59     s->channel_valid = 0;
  60     s->device_index = PA_INVALID_INDEX;
  61     s->requested_bytes = 0;
  62     s->state = PA_STREAM_DISCONNECTED;
  63     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
  64
  65     s->mcalign = pa_mcalign_new(pa_frame_size(ss), c->memblock_stat);
  66
  67     s->counter = 0;
  68     s->previous_time = 0;
  69
  70     s->corked = 0;
  71     s->interpolate = 0;
  72
  73     s->ipol_usec = 0;
  74     memset(&s->ipol_timestamp, 0, sizeof(s->ipol_timestamp));
  75     s->ipol_event = NULL;
  76     s->ipol_requested = 0;
  77
  78     PA_LLIST_PREPEND(struct pa_stream, c->streams, s);
  79
  80     return pa_stream_ref(s);
  81 }
  82
  83 static void stream_free(struct pa_stream *s) {
  84     assert(s);
  85
  86     if (s->ipol_event) {
  87         assert(s->mainloop);
  88         s->mainloop->time_free(s->ipol_event);
  89     }
  90
  91     pa_mcalign_free(s->mcalign);
  92
  93     pa_xfree(s->name);
  94     pa_xfree(s);
  95 }
  96
  97 void pa_stream_unref(struct pa_stream *s) {
  98     assert(s && s->ref >= 1);
  99
 100     if (--(s->ref) == 0)
 101         stream_free(s);
 102 }
 103
 104 struct pa_stream* pa_stream_ref(struct pa_stream *s) {
 105     assert(s && s->ref >= 1);
 106     s->ref++;
 107     return s;
 108 }
 109
 110 enum pa_stream_state pa_stream_get_state(struct pa_stream *s) {
 111     assert(s && s->ref >= 1);
 112     return s->state;
 113 }
 114
 115 struct pa_context* pa_stream_get_context(struct pa_stream *s) {
 116     assert(s && s->ref >= 1);
 117     return s->context;
 118 }
 119
 120 uint32_t pa_stream_get_index(struct pa_stream *s) {
 121     assert(s && s->ref >= 1);
 122     return s->device_index;
 123 }
 124
 125 void pa_stream_set_state(struct pa_stream *s, enum pa_stream_state st) {
 126     assert(s && s->ref >= 1);
 127
 128     if (s->state == st)
 129         return;
 130
 131     pa_stream_ref(s);
 132
 133     s->state = st;
 134
 135     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
 136         if (s->channel_valid)
 137             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 138
 139         PA_LLIST_REMOVE(struct pa_stream, s->context->streams, s);
 140         pa_stream_unref(s);
 141     }
 142
 143     if (s->state_callback)
 144         s->state_callback(s, s->state_userdata);
 145
 146     pa_stream_unref(s);
 147 }
 148
 149 void pa_command_stream_killed(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 150     struct pa_context *c = userdata;
 151     struct pa_stream *s;
 152     uint32_t channel;
 153     assert(pd && (command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED) && t && c);
 154
 155     pa_context_ref(c);
 156
 157     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 158         !pa_tagstruct_eof(t)) {
 159         pa_context_fail(c, PA_ERROR_PROTOCOL);
 160         goto finish;
 161     }
 162
 163     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 164         goto finish;
 165
 166     c->error = PA_ERROR_KILLED;
 167     pa_stream_set_state(s, PA_STREAM_FAILED);
 168
 169 finish:
 170     pa_context_unref(c);
 171 }
 172
 173 void pa_command_request(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 174     struct pa_stream *s;
 175     struct pa_context *c = userdata;
 176     uint32_t bytes, channel;
 177     assert(pd && command == PA_COMMAND_REQUEST && t && c);
 178
 179     pa_context_ref(c);
 180
 181     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 182         pa_tagstruct_getu32(t, &bytes) < 0 ||
 183         !pa_tagstruct_eof(t)) {
 184         pa_context_fail(c, PA_ERROR_PROTOCOL);
 185         goto finish;
 186     }
 187
 188     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 189         goto finish;
 190
 191     if (s->state != PA_STREAM_READY)
 192         goto finish;
 193
 194     pa_stream_ref(s);
 195
 196     s->requested_bytes += bytes;
 197
 198     if (s->requested_bytes && s->write_callback)
 199         s->write_callback(s, s->requested_bytes, s->write_userdata);
 200
 201     pa_stream_unref(s);
 202
 203 finish:
 204     pa_context_unref(c);
 205 }
 206
 207 static void ipol_callback(struct pa_mainloop_api *m, struct pa_time_event *e, const struct timeval *tv, void *userdata) {
 208     struct timeval tv2;
 209     struct pa_stream *s = userdata;
 210
 211     pa_stream_ref(s);
 212
 213 /*     pa_log("requesting new ipol data\n"); */
 214
 215     if (s->state == PA_STREAM_READY && !s->ipol_requested) {
 216         pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 217         s->ipol_requested = 1;
 218     }
 219
 220     gettimeofday(&tv2, NULL);
 221     pa_timeval_add(&tv2, LATENCY_IPOL_INTERVAL_USEC);
 222
 223     m->time_restart(e, &tv2);
 224
 225     pa_stream_unref(s);
 226 }
 227
 228
 229 void pa_create_stream_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 230     struct pa_stream *s = userdata;
 231     assert(pd && s && s->state == PA_STREAM_CREATING);
 232
 233     pa_stream_ref(s);
 234
 235     if (command != PA_COMMAND_REPLY) {
 236         if (pa_context_handle_error(s->context, command, t) < 0)
 237             goto finish;
 238
 239         pa_stream_set_state(s, PA_STREAM_FAILED);
 240         goto finish;
 241     }
 242
 243     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 244         ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
 245         ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0) ||
 246         !pa_tagstruct_eof(t)) {
 247         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 248         goto finish;
 249     }
 250
 251     s->channel_valid = 1;
 252     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 253     pa_stream_set_state(s, PA_STREAM_READY);
 254
 255     if (s->interpolate) {
 256         struct timeval tv;
 257         pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 258
 259         gettimeofday(&tv, NULL);
 260         tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
 261
 262         assert(!s->ipol_event);
 263         s->ipol_event = s->mainloop->time_new(s->mainloop, &tv, &ipol_callback, s);
 264     }
 265
 266     if (s->requested_bytes && s->ref > 1 && s->write_callback)
 267         s->write_callback(s, s->requested_bytes, s->write_userdata);
 268
 269 finish:
 270     pa_stream_unref(s);
 271 }
 272
 273 static void create_stream(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr, enum pa_stream_flags flags, pa_volume_t volume) {
 274     struct pa_tagstruct *t;
 275     uint32_t tag;
 276     assert(s && s->ref >= 1 && s->state == PA_STREAM_DISCONNECTED);
 277
 278     pa_stream_ref(s);
 279
 280     s->interpolate = !!(flags & PA_STREAM_INTERPOLATE_LATENCY);
 281     pa_stream_trash_ipol(s);
 282
 283     if (attr)
 284         s->buffer_attr = *attr;
 285     else {
 286         /* half a second */
 287         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
 288         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
 289         s->buffer_attr.minreq = s->buffer_attr.tlength/100;
 290         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
 291         s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
 292     }
 293
 294     pa_stream_set_state(s, PA_STREAM_CREATING);
 295
 296     t = pa_tagstruct_new(NULL, 0);
 297     assert(t);
 298
 299     if (!dev) {
 300         if (s->direction == PA_STREAM_PLAYBACK)
 301             dev = s->context->conf->default_sink;
 302         else
 303             dev = s->context->conf->default_source;
 304     }
 305
 306     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM);
 307     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 308     pa_tagstruct_puts(t, s->name);
 309     pa_tagstruct_put_sample_spec(t, &s->sample_spec);
 310     pa_tagstruct_putu32(t, PA_INVALID_INDEX);
 311     pa_tagstruct_puts(t, dev);
 312     pa_tagstruct_putu32(t, s->buffer_attr.maxlength);
 313     pa_tagstruct_put_boolean(t, !!(flags & PA_STREAM_START_CORKED));
 314     if (s->direction == PA_STREAM_PLAYBACK) {
 315         pa_tagstruct_putu32(t, s->buffer_attr.tlength);
 316         pa_tagstruct_putu32(t, s->buffer_attr.prebuf);
 317         pa_tagstruct_putu32(t, s->buffer_attr.minreq);
 318         pa_tagstruct_putu32(t, volume);
 319     } else
 320         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 321
 322     pa_pstream_send_tagstruct(s->context->pstream, t);
 323     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
 324
 325     pa_stream_unref(s);
 326 }
 327
 328 void pa_stream_connect_playback(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr, enum pa_stream_flags flags, pa_volume_t volume) {
 329     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 330     s->direction = PA_STREAM_PLAYBACK;
 331     create_stream(s, dev, attr, flags, volume);
 332 }
 333
 334 void pa_stream_connect_record(struct pa_stream *s, const char *dev, const struct pa_buffer_attr *attr, enum pa_stream_flags flags) {
 335     assert(s && s->context->state == PA_CONTEXT_READY && s->ref >= 1);
 336     s->direction = PA_STREAM_RECORD;
 337     create_stream(s, dev, attr, flags, 0);
 338 }
 339
 340 void pa_stream_write(struct pa_stream *s, const void *data, size_t length, void (*free_cb)(void *p), size_t delta) {
 341     struct pa_memchunk chunk;
 342     assert(s && s->context && data && length && s->state == PA_STREAM_READY && s->ref >= 1);
 343
 344     if (free_cb) {
 345         chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, 1, s->context->memblock_stat);
 346         assert(chunk.memblock && chunk.memblock->data);
 347     } else {
 348         chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
 349         assert(chunk.memblock && chunk.memblock->data);
 350         memcpy(chunk.memblock->data, data, length);
 351     }
 352     chunk.index = 0;
 353     chunk.length = length;
 354
 355     pa_pstream_send_memblock(s->context->pstream, s->channel, delta, &chunk);
 356     pa_memblock_unref(chunk.memblock);
 357
 358     if (length < s->requested_bytes)
 359         s->requested_bytes -= length;
 360     else
 361         s->requested_bytes = 0;
 362
 363     s->counter += length;
 364 }
 365
 366 size_t pa_stream_writable_size(struct pa_stream *s) {
 367     assert(s && s->ref >= 1);
 368     return s->state == PA_STREAM_READY ? s->requested_bytes : 0;
 369 }
 370
 371 struct pa_operation * pa_stream_drain(struct pa_stream *s, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata) {
 372     struct pa_operation *o;
 373     struct pa_tagstruct *t;
 374     uint32_t tag;
 375     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 376
 377     o = pa_operation_new(s->context, s);
 378     assert(o);
 379     o->callback = cb;
 380     o->userdata = userdata;
 381
 382     t = pa_tagstruct_new(NULL, 0);
 383     assert(t);
 384     pa_tagstruct_putu32(t, PA_COMMAND_DRAIN_PLAYBACK_STREAM);
 385     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 386     pa_tagstruct_putu32(t, s->channel);
 387     pa_pstream_send_tagstruct(s->context->pstream, t);
 388     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 389
 390     return pa_operation_ref(o);
 391 }
 392
 393 static void stream_get_latency_info_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 394     struct pa_operation *o = userdata;
 395     struct pa_latency_info i, *p = NULL;
 396     struct timeval local, remote, now;
 397     assert(pd && o && o->stream && o->context);
 398
 399     if (command != PA_COMMAND_REPLY) {
 400         if (pa_context_handle_error(o->context, command, t) < 0)
 401             goto finish;
 402
 403     } else if (pa_tagstruct_get_usec(t, &i.buffer_usec) < 0 ||
 404                pa_tagstruct_get_usec(t, &i.sink_usec) < 0 ||
 405                pa_tagstruct_get_usec(t, &i.source_usec) < 0 ||
 406                pa_tagstruct_get_boolean(t, &i.playing) < 0 ||
 407                pa_tagstruct_getu32(t, &i.queue_length) < 0 ||
 408                pa_tagstruct_get_timeval(t, &local) < 0 ||
 409                pa_tagstruct_get_timeval(t, &remote) < 0 ||
 410                pa_tagstruct_getu64(t, &i.counter) < 0 ||
 411                !pa_tagstruct_eof(t)) {
 412         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 413         goto finish;
 414     } else {
 415         gettimeofday(&now, NULL);
 416
 417         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
 418             /* local and remote seem to have synchronized clocks */
 419
 420             if (o->stream->direction == PA_STREAM_PLAYBACK)
 421                 i.transport_usec = pa_timeval_diff(&remote, &local);
 422             else
 423                 i.transport_usec = pa_timeval_diff(&now, &remote);
 424
 425             i.synchronized_clocks = 1;
 426             i.timestamp = remote;
 427         } else {
 428             /* clocks are not synchronized, let's estimate latency then */
 429             i.transport_usec = pa_timeval_diff(&now, &local)/2;
 430             i.synchronized_clocks = 0;
 431             i.timestamp = local;
 432             pa_timeval_add(&i.timestamp, i.transport_usec);
 433         }
 434
 435         if (o->stream->interpolate) {
 436 /*              pa_log("new interpol data\n");  */
 437             o->stream->ipol_timestamp = i.timestamp;
 438             o->stream->ipol_usec = pa_stream_get_time(o->stream, &i);
 439             o->stream->ipol_requested = 0;
 440         }
 441
 442         p = &i;
 443     }
 444
 445     if (o->callback) {
 446         void (*cb)(struct pa_stream *s, const struct pa_latency_info *i, void *userdata) = o->callback;
 447         cb(o->stream, p, o->userdata);
 448     }
 449
 450 finish:
 451     pa_operation_done(o);
 452     pa_operation_unref(o);
 453 }
 454
 455 struct pa_operation* pa_stream_get_latency_info(struct pa_stream *s, void (*cb)(struct pa_stream *p, const struct pa_latency_info*i, void *userdata), void *userdata) {
 456     uint32_t tag;
 457     struct pa_operation *o;
 458     struct pa_tagstruct *t;
 459     struct timeval now;
 460     assert(s && s->direction != PA_STREAM_UPLOAD);
 461
 462     o = pa_operation_new(s->context, s);
 463     assert(o);
 464     o->callback = cb;
 465     o->userdata = userdata;
 466
 467     t = pa_tagstruct_new(NULL, 0);
 468     assert(t);
 469     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY);
 470     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 471     pa_tagstruct_putu32(t, s->channel);
 472
 473     gettimeofday(&now, NULL);
 474     pa_tagstruct_put_timeval(t, &now);
 475     pa_tagstruct_putu64(t, s->counter);
 476
 477     pa_pstream_send_tagstruct(s->context->pstream, t);
 478     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_info_callback, o);
 479
 480     return pa_operation_ref(o);
 481 }
 482
 483 void pa_stream_disconnect_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 484     struct pa_stream *s = userdata;
 485     assert(pd && s && s->ref >= 1);
 486
 487     pa_stream_ref(s);
 488
 489     if (command != PA_COMMAND_REPLY) {
 490         if (pa_context_handle_error(s->context, command, t) < 0)
 491             goto finish;
 492
 493         pa_stream_set_state(s, PA_STREAM_FAILED);
 494         goto finish;
 495     } else if (!pa_tagstruct_eof(t)) {
 496         pa_context_fail(s->context, PA_ERROR_PROTOCOL);
 497         goto finish;
 498     }
 499
 500     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 501
 502 finish:
 503     pa_stream_unref(s);
 504 }
 505
 506 void pa_stream_disconnect(struct pa_stream *s) {
 507     struct pa_tagstruct *t;
 508     uint32_t tag;
 509     assert(s && s->ref >= 1);
 510
 511     if (!s->channel_valid || !s->context->state == PA_CONTEXT_READY)
 512         return;
 513
 514     pa_stream_ref(s);
 515
 516     t = pa_tagstruct_new(NULL, 0);
 517     assert(t);
 518
 519     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 520                         (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM));
 521     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 522     pa_tagstruct_putu32(t, s->channel);
 523     pa_pstream_send_tagstruct(s->context->pstream, t);
 524     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
 525
 526     pa_stream_unref(s);
 527 }
 528
 529 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) {
 530     assert(s && s->ref >= 1);
 531     s->read_callback = cb;
 532     s->read_userdata = userdata;
 533 }
 534
 535 void pa_stream_set_write_callback(struct pa_stream *s, void (*cb)(struct pa_stream *p, size_t length, void *userdata), void *userdata) {
 536     assert(s && s->ref >= 1);
 537     s->write_callback = cb;
 538     s->write_userdata = userdata;
 539 }
 540
 541 void pa_stream_set_state_callback(struct pa_stream *s, void (*cb)(struct pa_stream *s, void *userdata), void *userdata) {
 542     assert(s && s->ref >= 1);
 543     s->state_callback = cb;
 544     s->state_userdata = userdata;
 545 }
 546
 547 void pa_stream_simple_ack_callback(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata) {
 548     struct pa_operation *o = userdata;
 549     int success = 1;
 550     assert(pd && o && o->context && o->ref >= 1);
 551
 552     if (command != PA_COMMAND_REPLY) {
 553         if (pa_context_handle_error(o->context, command, t) < 0)
 554             goto finish;
 555
 556         success = 0;
 557     } else if (!pa_tagstruct_eof(t)) {
 558         pa_context_fail(o->context, PA_ERROR_PROTOCOL);
 559         goto finish;
 560     }
 561
 562     if (o->callback) {
 563         void (*cb)(struct pa_stream *s, int success, void *userdata) = o->callback;
 564         cb(o->stream, success, o->userdata);
 565     }
 566
 567 finish:
 568     pa_operation_done(o);
 569     pa_operation_unref(o);
 570 }
 571
 572 struct pa_operation* pa_stream_cork(struct pa_stream *s, int b, void (*cb) (struct pa_stream*s, int success, void *userdata), void *userdata) {
 573     struct pa_operation *o;
 574     struct pa_tagstruct *t;
 575     uint32_t tag;
 576     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 577
 578     if (s->interpolate) {
 579         if (!s->corked && b)
 580             /* Pausing */
 581             s->ipol_usec = pa_stream_get_interpolated_time(s);
 582         else if (s->corked && !b)
 583             /* Unpausing */
 584             gettimeofday(&s->ipol_timestamp, NULL);
 585     }
 586
 587     s->corked = b;
 588
 589     o = pa_operation_new(s->context, s);
 590     assert(o);
 591     o->callback = cb;
 592     o->userdata = userdata;
 593
 594     t = pa_tagstruct_new(NULL, 0);
 595     assert(t);
 596     pa_tagstruct_putu32(t, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM);
 597     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 598     pa_tagstruct_putu32(t, s->channel);
 599     pa_tagstruct_put_boolean(t, !!b);
 600     pa_pstream_send_tagstruct(s->context->pstream, t);
 601     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 602
 603     pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 604
 605     return pa_operation_ref(o);
 606 }
 607
 608 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) {
 609     struct pa_tagstruct *t;
 610     struct pa_operation *o;
 611     uint32_t tag;
 612     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY);
 613
 614     o = pa_operation_new(s->context, s);
 615     o->callback = cb;
 616     o->userdata = userdata;
 617
 618     t = pa_tagstruct_new(NULL, 0);
 619     pa_tagstruct_putu32(t, command);
 620     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 621     pa_tagstruct_putu32(t, s->channel);
 622     pa_pstream_send_tagstruct(s->context->pstream, t);
 623     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 624
 625     return pa_operation_ref(o);
 626 }
 627
 628 struct pa_operation* pa_stream_flush(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 629     struct pa_operation *o;
 630     o = pa_stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata);
 631     pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 632     return o;
 633 }
 634
 635 struct pa_operation* pa_stream_prebuf(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 636     struct pa_operation *o;
 637     o = pa_stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata);
 638     pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 639     return o;
 640 }
 641
 642 struct pa_operation* pa_stream_trigger(struct pa_stream *s, void (*cb)(struct pa_stream *s, int success, void *userdata), void *userdata) {
 643     struct pa_operation *o;
 644     o = pa_stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata);
 645     pa_operation_unref(pa_stream_get_latency_info(s, NULL, NULL));
 646     return o;
 647 }
 648
 649 struct pa_operation* pa_stream_set_name(struct pa_stream *s, const char *name, void(*cb)(struct pa_stream*c, int success,  void *userdata), void *userdata) {
 650     struct pa_operation *o;
 651     struct pa_tagstruct *t;
 652     uint32_t tag;
 653     assert(s && s->ref >= 1 && s->state == PA_STREAM_READY && name && s->direction != PA_STREAM_UPLOAD);
 654
 655     o = pa_operation_new(s->context, s);
 656     assert(o);
 657     o->callback = cb;
 658     o->userdata = userdata;
 659
 660     t = pa_tagstruct_new(NULL, 0);
 661     assert(t);
 662     pa_tagstruct_putu32(t, s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME);
 663     pa_tagstruct_putu32(t, tag = s->context->ctag++);
 664     pa_tagstruct_putu32(t, s->channel);
 665     pa_tagstruct_puts(t, name);
 666     pa_pstream_send_tagstruct(s->context->pstream, t);
 667     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 668
 669     return pa_operation_ref(o);
 670 }
 671
 672 uint64_t pa_stream_get_counter(struct pa_stream *s) {
 673     assert(s);
 674     return s->counter;
 675 }
 676
 677 pa_usec_t pa_stream_get_time(struct pa_stream *s, const struct pa_latency_info *i) {
 678     pa_usec_t usec;
 679     assert(s);
 680
 681     usec = pa_bytes_to_usec(i->counter, &s->sample_spec);
 682
 683     if (i) {
 684         if (s->direction == PA_STREAM_PLAYBACK) {
 685             pa_usec_t latency = i->transport_usec + i->buffer_usec + i->sink_usec;
 686             if (usec < latency)
 687                 usec = 0;
 688             else
 689                 usec -= latency;
 690
 691         } else if (s->direction == PA_STREAM_RECORD) {
 692             usec += i->source_usec + i->buffer_usec + i->transport_usec;
 693
 694             if (usec > i->sink_usec)
 695                 usec -= i->sink_usec;
 696             else
 697                 usec = 0;
 698         }
 699     }
 700
 701     if (usec < s->previous_time)
 702         usec = s->previous_time;
 703
 704     s->previous_time = usec;
 705
 706     return usec;
 707 }
 708
 709 static pa_usec_t time_counter_diff(struct pa_stream *s, pa_usec_t t, pa_usec_t c, int *negative) {
 710     assert(s);
 711
 712     if (negative)
 713         *negative = 0;
 714
 715     if (c < t) {
 716         if (s->direction == PA_STREAM_RECORD) {
 717             if (negative)
 718                 *negative = 1;
 719
 720             return t-c;
 721         } else
 722             return 0;
 723     } else
 724         return c-t;
 725 }
 726
 727 pa_usec_t pa_stream_get_latency(struct pa_stream *s, const struct pa_latency_info *i, int *negative) {
 728     pa_usec_t t, c;
 729     assert(s && i);
 730
 731     t = pa_stream_get_time(s, i);
 732     c = pa_bytes_to_usec(s->counter, &s->sample_spec);
 733
 734     return time_counter_diff(s, t, c, negative);
 735 }
 736
 737 const struct pa_sample_spec* pa_stream_get_sample_spec(struct pa_stream *s) {
 738     assert(s);
 739     return &s->sample_spec;
 740 }
 741
 742 void pa_stream_trash_ipol(struct pa_stream *s) {
 743     assert(s);
 744
 745     if (!s->interpolate)
 746         return;
 747
 748     memset(&s->ipol_timestamp, 0, sizeof(s->ipol_timestamp));
 749     s->ipol_usec = 0;
 750 }
 751
 752 pa_usec_t pa_stream_get_interpolated_time(struct pa_stream *s) {
 753     pa_usec_t usec;
 754     assert(s && s->interpolate);
 755
 756     if (s->corked)
 757         usec = s->ipol_usec;
 758     else {
 759         if (s->ipol_timestamp.tv_sec == 0)
 760             usec = 0;
 761         else
 762             usec = s->ipol_usec + pa_timeval_age(&s->ipol_timestamp);
 763     }
 764
 765     if (usec < s->previous_time)
 766         usec = s->previous_time;
 767
 768     s->previous_time = usec;
 769     return usec;
 770 }
 771
 772 pa_usec_t pa_stream_get_interpolated_latency(struct pa_stream *s, int *negative) {
 773     pa_usec_t t, c;
 774     assert(s && s->interpolate);
 775
 776     t = pa_stream_get_interpolated_time(s);
 777     c = pa_bytes_to_usec(s->counter, &s->sample_spec);
 778     return time_counter_diff(s, t, c, negative);
 779 }