code.delx.au - pulseaudio/blob - src/polyp/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 <polyp/def.h>
  32 #include <polypcore/xmalloc.h>
  33 #include <polypcore/pstream-util.h>
  34 #include <polypcore/util.h>
  35 #include <polypcore/log.h>
  36 #include <polypcore/hashmap.h>
  37
  38 #include "internal.h"
  39
  40 #define LATENCY_IPOL_INTERVAL_USEC (10000L)
  41
  42 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
  43     pa_stream *s;
  44     int i;
  45
  46     assert(c);
  47
  48     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
  49     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
  50
  51     s = pa_xnew(pa_stream, 1);
  52     s->ref = 1;
  53     s->context = c;
  54     s->mainloop = c->mainloop;
  55
  56     s->read_callback = NULL;
  57     s->read_userdata = NULL;
  58     s->write_callback = NULL;
  59     s->write_userdata = NULL;
  60     s->state_callback = NULL;
  61     s->state_userdata = NULL;
  62     s->overflow_callback = NULL;
  63     s->overflow_userdata = NULL;
  64     s->underflow_callback = NULL;
  65     s->underflow_userdata = NULL;
  66
  67     s->direction = PA_STREAM_NODIRECTION;
  68     s->name = pa_xstrdup(name);
  69     s->sample_spec = *ss;
  70     s->flags = 0;
  71
  72     if (map)
  73         s->channel_map = *map;
  74     else
  75         pa_channel_map_init_auto(&s->channel_map, ss->channels);
  76
  77     s->channel = 0;
  78     s->channel_valid = 0;
  79     s->syncid = c->csyncid++;
  80     s->device_index = PA_INVALID_INDEX;
  81     s->requested_bytes = 0;
  82     s->state = PA_STREAM_UNCONNECTED;
  83     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
  84
  85     s->peek_memchunk.index = 0;
  86     s->peek_memchunk.length = 0;
  87     s->peek_memchunk.memblock = NULL;
  88
  89     s->record_memblockq = NULL;
  90
  91     s->previous_time = 0;
  92     s->timing_info_valid = 0;
  93     s->read_index_not_before = 0;
  94     s->write_index_not_before = 0;
  95
  96     for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
  97         s->write_index_corrections[i].valid = 0;
  98     s->current_write_index_correction = 0;
  99
 100     s->corked = 0;
 101
 102     s->ipol_usec_valid = 0;
 103     s->ipol_timestamp.tv_sec = 0;
 104     s->ipol_timestamp.tv_usec = 0;
 105
 106     s->auto_timing_update_event = NULL;
 107     s->auto_timing_update_requested = 0;
 108
 109     PA_LLIST_PREPEND(pa_stream, c->streams, s);
 110
 111     /* The context and stream will point at each other. We cannot ref count
 112        both though since that will create a loop. */
 113     pa_context_ref(s->context);
 114
 115     return s;
 116 }
 117
 118 static void stream_free(pa_stream *s) {
 119     assert(s && s->context && !s->channel_valid);
 120
 121     PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
 122
 123     pa_context_unref(s->context);
 124
 125     if (s->auto_timing_update_event) {
 126         assert(s->mainloop);
 127         s->mainloop->time_free(s->auto_timing_update_event);
 128     }
 129
 130     if (s->peek_memchunk.memblock)
 131         pa_memblock_unref(s->peek_memchunk.memblock);
 132
 133     if (s->record_memblockq)
 134         pa_memblockq_free(s->record_memblockq);
 135
 136     pa_xfree(s->name);
 137     pa_xfree(s);
 138 }
 139
 140 void pa_stream_unref(pa_stream *s) {
 141     assert(s);
 142     assert(s->ref >= 1);
 143
 144     if (--(s->ref) == 0)
 145         stream_free(s);
 146 }
 147
 148 pa_stream* pa_stream_ref(pa_stream *s) {
 149     assert(s);
 150     assert(s->ref >= 1);
 151
 152     s->ref++;
 153     return s;
 154 }
 155
 156 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
 157     assert(s);
 158     assert(s->ref >= 1);
 159
 160     return s->state;
 161 }
 162
 163 pa_context* pa_stream_get_context(pa_stream *s) {
 164     assert(s);
 165     assert(s->ref >= 1);
 166
 167     return s->context;
 168 }
 169
 170 uint32_t pa_stream_get_index(pa_stream *s) {
 171     assert(s);
 172     assert(s->ref >= 1);
 173
 174     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
 175
 176     return s->device_index;
 177 }
 178
 179 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
 180     assert(s);
 181     assert(s->ref >= 1);
 182
 183     if (s->state == st)
 184         return;
 185
 186     pa_stream_ref(s);
 187
 188     s->state = st;
 189
 190     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
 191         /* Detach from context */
 192
 193         if (s->channel_valid)
 194             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
 195
 196         s->channel = 0;
 197         s->channel_valid = 0;
 198
 199         /* We keep a ref as long as we're connected */
 200         pa_stream_unref(s);
 201     }
 202
 203     if (s->state_callback)
 204         s->state_callback(s, s->state_userdata);
 205
 206     pa_stream_unref(s);
 207 }
 208
 209 void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 210     pa_context *c = userdata;
 211     pa_stream *s;
 212     uint32_t channel;
 213
 214     assert(pd);
 215     assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
 216     assert(t);
 217     assert(c);
 218
 219     pa_context_ref(c);
 220
 221     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 222         !pa_tagstruct_eof(t)) {
 223         pa_context_fail(c, PA_ERR_PROTOCOL);
 224         goto finish;
 225     }
 226
 227     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
 228         goto finish;
 229
 230     pa_context_set_error(c, PA_ERR_KILLED);
 231     pa_stream_set_state(s, PA_STREAM_FAILED);
 232
 233 finish:
 234     pa_context_unref(c);
 235 }
 236
 237 void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 238     pa_stream *s;
 239     pa_context *c = userdata;
 240     uint32_t bytes, channel;
 241
 242     assert(pd);
 243     assert(command == PA_COMMAND_REQUEST);
 244     assert(t);
 245     assert(c);
 246
 247     pa_context_ref(c);
 248
 249     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 250         pa_tagstruct_getu32(t, &bytes) < 0 ||
 251         !pa_tagstruct_eof(t)) {
 252         pa_context_fail(c, PA_ERR_PROTOCOL);
 253         goto finish;
 254     }
 255
 256     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 257         goto finish;
 258
 259     if (s->state == PA_STREAM_READY) {
 260         s->requested_bytes += bytes;
 261
 262         if (s->requested_bytes > 0 && s->write_callback)
 263             s->write_callback(s, s->requested_bytes, s->write_userdata);
 264     }
 265
 266 finish:
 267     pa_context_unref(c);
 268 }
 269
 270 void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 271     pa_stream *s;
 272     pa_context *c = userdata;
 273     uint32_t channel;
 274
 275     assert(pd);
 276     assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
 277     assert(t);
 278     assert(c);
 279
 280     pa_context_ref(c);
 281
 282     if (pa_tagstruct_getu32(t, &channel) < 0 ||
 283         !pa_tagstruct_eof(t)) {
 284         pa_context_fail(c, PA_ERR_PROTOCOL);
 285         goto finish;
 286     }
 287
 288     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
 289         goto finish;
 290
 291     if (s->state == PA_STREAM_READY) {
 292
 293         if (command == PA_COMMAND_OVERFLOW) {
 294             if (s->overflow_callback)
 295                 s->overflow_callback(s, s->overflow_userdata);
 296         } else if (command == PA_COMMAND_UNDERFLOW) {
 297             if (s->underflow_callback)
 298                 s->underflow_callback(s, s->underflow_userdata);
 299         }
 300     }
 301
 302  finish:
 303     pa_context_unref(c);
 304 }
 305
 306 static void request_auto_timing_update(pa_stream *s, int force) {
 307     struct timeval next;
 308     assert(s);
 309
 310     if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
 311         return;
 312
 313     if (s->state == PA_STREAM_READY &&
 314         (force || !s->auto_timing_update_requested)) {
 315         pa_operation *o;
 316
 317 /*         pa_log("automatically requesting new timing data");   */
 318
 319         if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
 320             pa_operation_unref(o);
 321             s->auto_timing_update_requested = 1;
 322         }
 323     }
 324
 325     pa_gettimeofday(&next);
 326     pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
 327     s->mainloop->time_restart(s->auto_timing_update_event, &next);
 328 }
 329
 330 static void invalidate_indexes(pa_stream *s, int r, int w) {
 331     assert(s);
 332
 333 /*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
 334
 335     if (s->state != PA_STREAM_READY)
 336         return;
 337
 338     if (w) {
 339         s->write_index_not_before = s->context->ctag;
 340
 341         if (s->timing_info_valid)
 342             s->timing_info.write_index_corrupt = 1;
 343
 344 /*         pa_log("write_index invalidated"); */
 345     }
 346
 347     if (r) {
 348         s->read_index_not_before = s->context->ctag;
 349
 350         if (s->timing_info_valid)
 351             s->timing_info.read_index_corrupt = 1;
 352
 353 /*         pa_log("read_index invalidated"); */
 354     }
 355
 356     if ((s->direction == PA_STREAM_PLAYBACK && r) ||
 357         (s->direction == PA_STREAM_RECORD && w))
 358         s->ipol_usec_valid = 0;
 359
 360     request_auto_timing_update(s, 1);
 361 }
 362
 363 static void auto_timing_update_callback(pa_mainloop_api *m, pa_time_event *e, PA_GCC_UNUSED const struct timeval *tv, void *userdata) {
 364     pa_stream *s = userdata;
 365
 366 /*     pa_log("time event");    */
 367
 368     pa_stream_ref(s);
 369     request_auto_timing_update(s, 0);
 370     pa_stream_unref(s);
 371 }
 372
 373 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 374     pa_stream *s = userdata;
 375
 376     assert(pd);
 377     assert(t);
 378     assert(s);
 379     assert(s->state == PA_STREAM_CREATING);
 380
 381     pa_stream_ref(s);
 382
 383     if (command != PA_COMMAND_REPLY) {
 384         if (pa_context_handle_error(s->context, command, t) < 0)
 385             goto finish;
 386
 387         pa_stream_set_state(s, PA_STREAM_FAILED);
 388         goto finish;
 389     }
 390
 391     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
 392         ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
 393         ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0) ||
 394         !pa_tagstruct_eof(t)) {
 395         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 396         goto finish;
 397     }
 398
 399     if (s->direction == PA_STREAM_RECORD) {
 400         assert(!s->record_memblockq);
 401
 402         s->record_memblockq = pa_memblockq_new(
 403                 0,
 404                 s->buffer_attr.maxlength,
 405                 0,
 406                 pa_frame_size(&s->sample_spec),
 407                 1,
 408                 0,
 409                 NULL,
 410                 s->context->memblock_stat);
 411     }
 412
 413     s->channel_valid = 1;
 414     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 415
 416     /* We add an extra ref as long as we're connected (i.e. in the dynarray) */
 417     pa_stream_ref(s);
 418
 419     pa_stream_set_state(s, PA_STREAM_READY);
 420
 421     if (s->direction != PA_STREAM_UPLOAD &&
 422         s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
 423         struct timeval tv;
 424
 425         pa_gettimeofday(&tv);
 426         tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
 427
 428         assert(!s->auto_timing_update_event);
 429         s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
 430
 431         request_auto_timing_update(s, 1);
 432     }
 433
 434     if (s->requested_bytes > 0 && s->ref > 1 && s->write_callback)
 435         s->write_callback(s, s->requested_bytes, s->write_userdata);
 436
 437 finish:
 438     pa_stream_unref(s);
 439 }
 440
 441 static int create_stream(
 442         pa_stream_direction_t direction,
 443         pa_stream *s,
 444         const char *dev,
 445         const pa_buffer_attr *attr,
 446         pa_stream_flags_t flags,
 447         const pa_cvolume *volume,
 448         pa_stream *sync_stream) {
 449
 450     pa_tagstruct *t;
 451     uint32_t tag;
 452
 453     assert(s);
 454     assert(s->ref >= 1);
 455
 456     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
 457     PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
 458                                                PA_STREAM_START_CORKED|
 459                                                PA_STREAM_INTERPOLATE_TIMING|
 460                                                PA_STREAM_NOT_MONOTONOUS|
 461                                                PA_STREAM_AUTO_TIMING_UPDATE : 0))), PA_ERR_INVALID);
 462     PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_PLAYBACK || flags == 0, PA_ERR_INVALID);
 463     PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
 464     PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
 465
 466     pa_stream_ref(s);
 467
 468     s->direction = direction;
 469     s->flags = flags;
 470
 471     if (sync_stream)
 472         s->syncid = sync_stream->syncid;
 473
 474     if (attr)
 475         s->buffer_attr = *attr;
 476     else {
 477         /* half a second */
 478         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
 479         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
 480         s->buffer_attr.minreq = s->buffer_attr.tlength/100;
 481         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
 482         s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
 483     }
 484
 485     if (!dev)
 486         dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
 487
 488     t = pa_tagstruct_command(
 489             s->context,
 490             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
 491             &tag);
 492
 493     pa_tagstruct_put(
 494             t,
 495             PA_TAG_STRING, s->name,
 496             PA_TAG_SAMPLE_SPEC, &s->sample_spec,
 497             PA_TAG_CHANNEL_MAP, &s->channel_map,
 498             PA_TAG_U32, PA_INVALID_INDEX,
 499             PA_TAG_STRING, dev,
 500             PA_TAG_U32, s->buffer_attr.maxlength,
 501             PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED),
 502             PA_TAG_INVALID);
 503
 504     if (s->direction == PA_STREAM_PLAYBACK) {
 505         pa_cvolume cv;
 506
 507         pa_tagstruct_put(
 508                 t,
 509                 PA_TAG_U32, s->buffer_attr.tlength,
 510                 PA_TAG_U32, s->buffer_attr.prebuf,
 511                 PA_TAG_U32, s->buffer_attr.minreq,
 512                 PA_TAG_U32, s->syncid,
 513                 PA_TAG_INVALID);
 514
 515         if (!volume)
 516             volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
 517
 518         pa_tagstruct_put_cvolume(t, volume);
 519     } else
 520         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 521
 522     pa_pstream_send_tagstruct(s->context->pstream, t);
 523     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
 524
 525     pa_stream_set_state(s, PA_STREAM_CREATING);
 526
 527     pa_stream_unref(s);
 528     return 0;
 529 }
 530
 531 int pa_stream_connect_playback(
 532         pa_stream *s,
 533         const char *dev,
 534         const pa_buffer_attr *attr,
 535         pa_stream_flags_t flags,
 536         pa_cvolume *volume,
 537         pa_stream *sync_stream) {
 538
 539     assert(s);
 540     assert(s->ref >= 1);
 541
 542     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
 543 }
 544
 545 int pa_stream_connect_record(
 546         pa_stream *s,
 547         const char *dev,
 548         const pa_buffer_attr *attr,
 549         pa_stream_flags_t flags) {
 550
 551     assert(s);
 552     assert(s->ref >= 1);
 553
 554     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
 555 }
 556
 557 int pa_stream_write(
 558         pa_stream *s,
 559         const void *data,
 560         size_t length,
 561         void (*free_cb)(void *p),
 562         int64_t offset,
 563         pa_seek_mode_t seek) {
 564
 565     pa_memchunk chunk;
 566
 567     assert(s);
 568     assert(s->ref >= 1);
 569     assert(data);
 570
 571     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 572     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 573     PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
 574     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
 575
 576     if (length <= 0)
 577         return 0;
 578
 579     if (free_cb)
 580         chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, 1, s->context->memblock_stat);
 581     else {
 582         chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
 583         memcpy(chunk.memblock->data, data, length);
 584     }
 585
 586     chunk.index = 0;
 587     chunk.length = length;
 588
 589     pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
 590     pa_memblock_unref(chunk.memblock);
 591
 592     if (length < s->requested_bytes)
 593         s->requested_bytes -= length;
 594     else
 595         s->requested_bytes = 0;
 596
 597     if (s->direction == PA_STREAM_PLAYBACK) {
 598
 599         /* Update latency request correction */
 600         if (s->write_index_corrections[s->current_write_index_correction].valid) {
 601
 602             if (seek == PA_SEEK_ABSOLUTE) {
 603                 s->write_index_corrections[s->current_write_index_correction].corrupt = 0;
 604                 s->write_index_corrections[s->current_write_index_correction].absolute = 1;
 605                 s->write_index_corrections[s->current_write_index_correction].value = offset + length;
 606             } else if (seek == PA_SEEK_RELATIVE) {
 607                 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
 608                     s->write_index_corrections[s->current_write_index_correction].value += offset + length;
 609             } else
 610                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
 611         }
 612
 613         /* Update the write index in the already available latency data */
 614         if (s->timing_info_valid) {
 615
 616             if (seek == PA_SEEK_ABSOLUTE) {
 617                 s->timing_info.write_index_corrupt = 0;
 618                 s->timing_info.write_index = offset + length;
 619             } else if (seek == PA_SEEK_RELATIVE) {
 620                 if (!s->timing_info.write_index_corrupt)
 621                     s->timing_info.write_index += offset + length;
 622             } else
 623                 s->timing_info.write_index_corrupt = 1;
 624         }
 625
 626         if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
 627             request_auto_timing_update(s, 1);
 628     }
 629
 630     return 0;
 631 }
 632
 633 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
 634     assert(s);
 635     assert(s->ref >= 1);
 636     assert(data);
 637     assert(length);
 638
 639     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 640     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 641
 642     if (!s->peek_memchunk.memblock) {
 643
 644         if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
 645             *data = NULL;
 646             *length = 0;
 647             return 0;
 648         }
 649     }
 650
 651     *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
 652     *length = s->peek_memchunk.length;
 653     return 0;
 654 }
 655
 656 int pa_stream_drop(pa_stream *s) {
 657     assert(s);
 658     assert(s->ref >= 1);
 659
 660     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 661     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
 662     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
 663
 664     pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
 665
 666     /* Fix the simulated local read index */
 667     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
 668         s->timing_info.read_index += s->peek_memchunk.length;
 669
 670     pa_memblock_unref(s->peek_memchunk.memblock);
 671     s->peek_memchunk.length = 0;
 672     s->peek_memchunk.index = 0;
 673     s->peek_memchunk.memblock = NULL;
 674
 675     return 0;
 676 }
 677
 678 size_t pa_stream_writable_size(pa_stream *s) {
 679     assert(s);
 680     assert(s->ref >= 1);
 681
 682     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 683     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 684
 685     return s->requested_bytes;
 686 }
 687
 688 size_t pa_stream_readable_size(pa_stream *s) {
 689     assert(s);
 690     assert(s->ref >= 1);
 691
 692     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
 693     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
 694
 695     return pa_memblockq_get_length(s->record_memblockq);
 696 }
 697
 698 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 699     pa_operation *o;
 700     pa_tagstruct *t;
 701     uint32_t tag;
 702
 703     assert(s);
 704     assert(s->ref >= 1);
 705
 706     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 707     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
 708
 709     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 710
 711     t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
 712     pa_tagstruct_putu32(t, s->channel);
 713     pa_pstream_send_tagstruct(s->context->pstream, t);
 714     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
 715
 716     return pa_operation_ref(o);
 717 }
 718
 719 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
 720     pa_operation *o = userdata;
 721     struct timeval local, remote, now;
 722     pa_timing_info *i;
 723
 724     assert(pd);
 725     assert(o);
 726     assert(o->stream);
 727     assert(o->context);
 728
 729     i = &o->stream->timing_info;
 730
 731 /*     pa_log("pre corrupt w:%u r:%u\n", !o->stream->timing_info_valid || i->write_index_corrupt,!o->stream->timing_info_valid || i->read_index_corrupt); */
 732
 733     o->stream->timing_info_valid = 0;
 734     i->write_index_corrupt = 0;
 735     i->read_index_corrupt = 0;
 736
 737 /*     pa_log("timing update %u\n", tag); */
 738
 739     if (command != PA_COMMAND_REPLY) {
 740         if (pa_context_handle_error(o->context, command, t) < 0)
 741             goto finish;
 742
 743     } else if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
 744                pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
 745                pa_tagstruct_get_boolean(t, &i->playing) < 0 ||
 746                pa_tagstruct_get_timeval(t, &local) < 0 ||
 747                pa_tagstruct_get_timeval(t, &remote) < 0 ||
 748                pa_tagstruct_gets64(t, &i->write_index) < 0 ||
 749                pa_tagstruct_gets64(t, &i->read_index) < 0 ||
 750                !pa_tagstruct_eof(t)) {
 751         pa_context_fail(o->context, PA_ERR_PROTOCOL);
 752         goto finish;
 753
 754     } else {
 755         o->stream->timing_info_valid = 1;
 756
 757         pa_gettimeofday(&now);
 758
 759         /* Calculcate timestamps */
 760         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
 761             /* local and remote seem to have synchronized clocks */
 762
 763             if (o->stream->direction == PA_STREAM_PLAYBACK)
 764                 i->transport_usec = pa_timeval_diff(&remote, &local);
 765             else
 766                 i->transport_usec = pa_timeval_diff(&now, &remote);
 767
 768             i->synchronized_clocks = 1;
 769             i->timestamp = remote;
 770         } else {
 771             /* clocks are not synchronized, let's estimate latency then */
 772             i->transport_usec = pa_timeval_diff(&now, &local)/2;
 773             i->synchronized_clocks = 0;
 774             i->timestamp = local;
 775             pa_timeval_add(&i->timestamp, i->transport_usec);
 776         }
 777
 778         /* Invalidate read and write indexes if necessary */
 779         if (tag < o->stream->read_index_not_before)
 780             i->read_index_corrupt = 1;
 781
 782         if (tag < o->stream->write_index_not_before)
 783             i->write_index_corrupt = 1;
 784
 785         if (o->stream->direction == PA_STREAM_PLAYBACK) {
 786             /* Write index correction */
 787
 788             int n, j;
 789             uint32_t ctag = tag;
 790
 791             /* Go through the saved correction values and add up the total correction.*/
 792
 793             for (n = 0, j = o->stream->current_write_index_correction+1;
 794                  n < PA_MAX_WRITE_INDEX_CORRECTIONS;
 795                  n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
 796
 797                 /* Step over invalid data or out-of-date data */
 798                 if (!o->stream->write_index_corrections[j].valid ||
 799                     o->stream->write_index_corrections[j].tag < ctag)
 800                     continue;
 801
 802                 /* Make sure that everything is in order */
 803                 ctag = o->stream->write_index_corrections[j].tag+1;
 804
 805                 /* Now fix the write index */
 806                 if (o->stream->write_index_corrections[j].corrupt) {
 807                     /* A corrupting seek was made */
 808                     i->write_index = 0;
 809                     i->write_index_corrupt = 1;
 810                 } else if (o->stream->write_index_corrections[j].absolute) {
 811                     /* An absolute seek was made */
 812                     i->write_index = o->stream->write_index_corrections[j].value;
 813                     i->write_index_corrupt = 0;
 814                 } else if (!i->write_index_corrupt) {
 815                     /* A relative seek was made */
 816                     i->write_index += o->stream->write_index_corrections[j].value;
 817                 }
 818             }
 819         }
 820
 821         if (o->stream->direction == PA_STREAM_RECORD) {
 822             /* Read index correction */
 823
 824             if (!i->read_index_corrupt)
 825                 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
 826         }
 827
 828         o->stream->ipol_timestamp = now;
 829         o->stream->ipol_usec_valid = 0;
 830     }
 831
 832     o->stream->auto_timing_update_requested = 0;
 833 /*     pa_log("post corrupt w:%u r:%u\n", i->write_index_corrupt || !o->stream->timing_info_valid, i->read_index_corrupt || !o->stream->timing_info_valid); */
 834
 835     /* Clear old correction entries */
 836     if (o->stream->direction == PA_STREAM_PLAYBACK) {
 837         int n;
 838
 839         for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
 840             if (!o->stream->write_index_corrections[n].valid)
 841                 continue;
 842
 843             if (o->stream->write_index_corrections[n].tag <= tag)
 844                 o->stream->write_index_corrections[n].valid = 0;
 845         }
 846     }
 847
 848     if (o->callback) {
 849         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
 850         cb(o->stream, o->stream->timing_info_valid, o->userdata);
 851     }
 852
 853 finish:
 854
 855     pa_operation_done(o);
 856     pa_operation_unref(o);
 857 }
 858
 859 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
 860     uint32_t tag;
 861     pa_operation *o;
 862     pa_tagstruct *t;
 863     struct timeval now;
 864     int cidx;
 865
 866     assert(s);
 867     assert(s->ref >= 1);
 868
 869     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 870     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 871
 872     if (s->direction == PA_STREAM_PLAYBACK) {
 873         /* Find a place to store the write_index correction data for this entry */
 874         cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
 875
 876         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
 877         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
 878     }
 879       o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 880
 881     t = pa_tagstruct_command(
 882             s->context,
 883             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
 884             &tag);
 885     pa_tagstruct_putu32(t, s->channel);
 886     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
 887
 888     pa_pstream_send_tagstruct(s->context->pstream, t);
 889     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, o);
 890
 891     if (s->direction == PA_STREAM_PLAYBACK) {
 892         /* Fill in initial correction data */
 893         o->stream->current_write_index_correction = cidx;
 894         o->stream->write_index_corrections[cidx].valid = 1;
 895         o->stream->write_index_corrections[cidx].tag = tag;
 896         o->stream->write_index_corrections[cidx].absolute = 0;
 897         o->stream->write_index_corrections[cidx].value = 0;
 898         o->stream->write_index_corrections[cidx].corrupt = 0;
 899     }
 900
 901 /*     pa_log("requesting update %u\n", tag); */
 902
 903     return pa_operation_ref(o);
 904 }
 905
 906 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 907     pa_stream *s = userdata;
 908
 909     assert(pd);
 910     assert(s);
 911     assert(s->ref >= 1);
 912
 913     pa_stream_ref(s);
 914
 915     if (command != PA_COMMAND_REPLY) {
 916         if (pa_context_handle_error(s->context, command, t) < 0)
 917             goto finish;
 918
 919         pa_stream_set_state(s, PA_STREAM_FAILED);
 920         goto finish;
 921     } else if (!pa_tagstruct_eof(t)) {
 922         pa_context_fail(s->context, PA_ERR_PROTOCOL);
 923         goto finish;
 924     }
 925
 926     pa_stream_set_state(s, PA_STREAM_TERMINATED);
 927
 928 finish:
 929     pa_stream_unref(s);
 930 }
 931
 932 int pa_stream_disconnect(pa_stream *s) {
 933     pa_tagstruct *t;
 934     uint32_t tag;
 935
 936     assert(s);
 937     assert(s->ref >= 1);
 938
 939     PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
 940     PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
 941
 942     pa_stream_ref(s);
 943
 944     t = pa_tagstruct_command(
 945             s->context,
 946             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
 947             (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
 948             &tag);
 949     pa_tagstruct_putu32(t, s->channel);
 950     pa_pstream_send_tagstruct(s->context->pstream, t);
 951     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
 952
 953     pa_stream_unref(s);
 954     return 0;
 955 }
 956
 957 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 958     assert(s);
 959     assert(s->ref >= 1);
 960
 961     s->read_callback = cb;
 962     s->read_userdata = userdata;
 963 }
 964
 965 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
 966     assert(s);
 967     assert(s->ref >= 1);
 968
 969     s->write_callback = cb;
 970     s->write_userdata = userdata;
 971 }
 972
 973 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
 974     assert(s);
 975     assert(s->ref >= 1);
 976
 977     s->state_callback = cb;
 978     s->state_userdata = userdata;
 979 }
 980
 981 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
 982     assert(s);
 983     assert(s->ref >= 1);
 984
 985     s->overflow_callback = cb;
 986     s->overflow_userdata = userdata;
 987 }
 988
 989 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
 990     assert(s);
 991     assert(s->ref >= 1);
 992
 993     s->underflow_callback = cb;
 994     s->underflow_userdata = userdata;
 995 }
 996
 997 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
 998     pa_operation *o = userdata;
 999     int success = 1;
1000
1001     assert(pd);
1002     assert(o);
1003     assert(o->context);
1004     assert(o->ref >= 1);
1005
1006     if (command != PA_COMMAND_REPLY) {
1007         if (pa_context_handle_error(o->context, command, t) < 0)
1008             goto finish;
1009
1010         success = 0;
1011     } else if (!pa_tagstruct_eof(t)) {
1012         pa_context_fail(o->context, PA_ERR_PROTOCOL);
1013         goto finish;
1014     }
1015
1016     if (o->callback) {
1017         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1018         cb(o->stream, success, o->userdata);
1019     }
1020
1021 finish:
1022     pa_operation_done(o);
1023     pa_operation_unref(o);
1024 }
1025
1026 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
1027     pa_operation *o;
1028     pa_tagstruct *t;
1029     uint32_t tag;
1030
1031     assert(s);
1032     assert(s->ref >= 1);
1033
1034     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1035     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1036
1037     s->corked = b;
1038
1039     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1040
1041     t = pa_tagstruct_command(
1042             s->context,
1043             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
1044             &tag);
1045     pa_tagstruct_putu32(t, s->channel);
1046     pa_tagstruct_put_boolean(t, !!b);
1047     pa_pstream_send_tagstruct(s->context->pstream, t);
1048     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
1049
1050     if (s->direction == PA_STREAM_PLAYBACK)
1051         invalidate_indexes(s, 1, 0);
1052
1053     return pa_operation_ref(o);
1054 }
1055
1056 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
1057     pa_tagstruct *t;
1058     pa_operation *o;
1059     uint32_t tag;
1060
1061     assert(s);
1062     assert(s->ref >= 1);
1063
1064     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1065
1066     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1067
1068     t = pa_tagstruct_command(s->context, command, &tag);
1069     pa_tagstruct_putu32(t, s->channel);
1070     pa_pstream_send_tagstruct(s->context->pstream, t);
1071     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
1072
1073     return pa_operation_ref(o);
1074 }
1075
1076 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1077     pa_operation *o;
1078
1079     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1080
1081     if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
1082
1083         if (s->direction == PA_STREAM_PLAYBACK) {
1084             if (s->write_index_corrections[s->current_write_index_correction].valid)
1085                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
1086
1087             if (s->timing_info_valid)
1088                 s->timing_info.write_index_corrupt = 1;
1089
1090             if (s->buffer_attr.prebuf > 0)
1091                 invalidate_indexes(s, 1, 0);
1092             else
1093                 request_auto_timing_update(s, 1);
1094         } else
1095             invalidate_indexes(s, 0, 1);
1096     }
1097
1098     return o;
1099 }
1100
1101 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1102     pa_operation *o;
1103
1104     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1105     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1106
1107     if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
1108         invalidate_indexes(s, 1, 0);
1109
1110     return o;
1111 }
1112
1113 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1114     pa_operation *o;
1115
1116     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1117     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1118
1119     if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
1120         invalidate_indexes(s, 1, 0);
1121
1122     return o;
1123 }
1124
1125 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
1126     pa_operation *o;
1127     pa_tagstruct *t;
1128     uint32_t tag;
1129
1130     assert(s);
1131     assert(s->ref >= 1);
1132     assert(name);
1133
1134     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1135     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1136
1137     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1138
1139     t = pa_tagstruct_command(
1140             s->context,
1141             s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
1142             &tag);
1143     pa_tagstruct_putu32(t, s->channel);
1144     pa_tagstruct_puts(t, name);
1145     pa_pstream_send_tagstruct(s->context->pstream, t);
1146     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
1147
1148     return pa_operation_ref(o);
1149 }
1150
1151 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
1152     pa_usec_t usec = 0;
1153
1154     assert(s);
1155     assert(s->ref >= 1);
1156
1157     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1158     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1159     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1160     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1161     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1162
1163     if (s->flags & PA_STREAM_INTERPOLATE_TIMING && s->ipol_usec_valid)
1164         usec = s->ipol_usec;
1165     else {
1166         if (s->direction == PA_STREAM_PLAYBACK) {
1167             /* The last byte that was written into the output device
1168              * had this time value associated */
1169             usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1170
1171             if (!s->corked) {
1172                 /* Because the latency info took a little time to come
1173                  * to us, we assume that the real output time is actually
1174                  * a little ahead */
1175                 usec += s->timing_info.transport_usec;
1176
1177                 /* However, the output device usually maintains a buffer
1178                    too, hence the real sample currently played is a little
1179                    back  */
1180                 if (s->timing_info.sink_usec >= usec)
1181                     usec = 0;
1182                 else
1183                     usec -= s->timing_info.sink_usec;
1184             }
1185
1186         } else if (s->direction == PA_STREAM_RECORD) {
1187             /* The last byte written into the server side queue had
1188              * this time value associated */
1189             usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1190
1191             if (!s->corked) {
1192                 /* Add transport latency */
1193                 usec += s->timing_info.transport_usec;
1194
1195                 /* Add latency of data in device buffer */
1196                 usec += s->timing_info.source_usec;
1197
1198                 /* If this is a monitor source, we need to correct the
1199                  * time by the playback device buffer */
1200                 if (s->timing_info.sink_usec >= usec)
1201                     usec = 0;
1202                 else
1203                     usec -= s->timing_info.sink_usec;
1204             }
1205         }
1206
1207         if (s->flags & PA_STREAM_INTERPOLATE_TIMING) {
1208             s->ipol_usec = usec;
1209             s->ipol_usec_valid = 1;
1210         }
1211     }
1212
1213     /* Interpolate if requested */
1214     if (s->flags & PA_STREAM_INTERPOLATE_TIMING) {
1215
1216         /* We just add the time that passed since the latency info was
1217          * current */
1218         if (!s->corked) {
1219             struct timeval now;
1220
1221             usec += pa_timeval_diff(pa_gettimeofday(&now), &s->ipol_timestamp);
1222             s->ipol_timestamp = now;
1223         }
1224     }
1225
1226     /* Make sure the time runs monotonically */
1227     if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
1228         if (usec < s->previous_time)
1229             usec = s->previous_time;
1230         else
1231             s->previous_time = usec;
1232     }
1233
1234     if (r_usec)
1235         *r_usec = usec;
1236
1237     return 0;
1238 }
1239
1240 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
1241     assert(s);
1242     assert(s->ref >= 1);
1243
1244     if (negative)
1245         *negative = 0;
1246
1247     if (a >= b)
1248         return a-b;
1249     else {
1250         if (negative && s->direction == PA_STREAM_RECORD) {
1251             *negative = 1;
1252             return b-a;
1253         } else
1254             return 0;
1255     }
1256 }
1257
1258 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
1259     pa_usec_t t, c;
1260     int r;
1261     int64_t cindex;
1262
1263     assert(s);
1264     assert(s->ref >= 1);
1265     assert(r_usec);
1266
1267     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1268     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1269     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1270     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1271     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1272
1273     if ((r = pa_stream_get_time(s, &t)) < 0)
1274         return r;
1275
1276     if (s->direction == PA_STREAM_PLAYBACK)
1277         cindex = s->timing_info.write_index;
1278     else
1279         cindex = s->timing_info.read_index;
1280
1281     if (cindex < 0)
1282         cindex = 0;
1283
1284     c = pa_bytes_to_usec(cindex, &s->sample_spec);
1285
1286     if (s->direction == PA_STREAM_PLAYBACK)
1287         *r_usec = time_counter_diff(s, c, t, negative);
1288     else
1289         *r_usec = time_counter_diff(s, t, c, negative);
1290
1291     return 0;
1292 }
1293
1294 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
1295     assert(s);
1296     assert(s->ref >= 1);
1297
1298     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1299     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1300     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
1301
1302     return &s->timing_info;
1303 }
1304
1305 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
1306     assert(s);
1307     assert(s->ref >= 1);
1308
1309     return &s->sample_spec;
1310 }
1311
1312 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
1313     assert(s);
1314     assert(s->ref >= 1);
1315
1316     return &s->channel_map;
1317 }