From f44ba092651aa75055e109e04b4164ea92ae7fdc Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Mon, 19 Jun 2006 21:53:48 +0000
Subject: big s/polyp/pulse/g

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1033 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 1366 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 1366 insertions(+)
 create mode 100644 src/pulse/stream.c

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
new file mode 100644
index 00000000..c8fc09d2
--- /dev/null
+++ b/src/pulse/stream.c
@@ -0,0 +1,1366 @@
+/* $Id$ */
+
+/***
+  This file is part of PulseAudio.
+ 
+  PulseAudio is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Lesser General Public License as published
+  by the Free Software Foundation; either version 2 of the License,
+  or (at your option) any later version.
+ 
+  PulseAudio is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+  General Public License for more details.
+ 
+  You should have received a copy of the GNU Lesser General Public License
+  along with PulseAudio; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+  USA.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <assert.h>
+#include <string.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <pulse/def.h>
+#include <pulse/timeval.h>
+#include <pulse/xmalloc.h>
+
+#include <pulsecore/pstream-util.h>
+#include <pulsecore/log.h>
+#include <pulsecore/hashmap.h>
+
+#include "internal.h"
+
+#define LATENCY_IPOL_INTERVAL_USEC (100000L)
+
+pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
+    pa_stream *s;
+    int i;
+    
+    assert(c);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
+    PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
+
+    s = pa_xnew(pa_stream, 1);
+    s->ref = 1;
+    s->context = c;
+    s->mainloop = c->mainloop;
+
+    s->read_callback = NULL;
+    s->read_userdata = NULL;
+    s->write_callback = NULL;
+    s->write_userdata = NULL;
+    s->state_callback = NULL;
+    s->state_userdata = NULL;
+    s->overflow_callback = NULL;
+    s->overflow_userdata = NULL;
+    s->underflow_callback = NULL;
+    s->underflow_userdata = NULL;
+    s->latency_update_callback = NULL;
+    s->latency_update_userdata = NULL;
+
+    s->direction = PA_STREAM_NODIRECTION;
+    s->name = pa_xstrdup(name);
+    s->sample_spec = *ss;
+    s->flags = 0;
+
+    if (map)
+        s->channel_map = *map;
+    else
+        pa_channel_map_init_auto(&s->channel_map, ss->channels, PA_CHANNEL_MAP_DEFAULT);
+    
+    s->channel = 0;
+    s->channel_valid = 0;
+    s->syncid = c->csyncid++;
+    s->device_index = PA_INVALID_INDEX;
+    s->requested_bytes = 0;
+    s->state = PA_STREAM_UNCONNECTED;
+    memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
+
+    s->peek_memchunk.index = 0;
+    s->peek_memchunk.length = 0;
+    s->peek_memchunk.memblock = NULL;
+
+    s->record_memblockq = NULL;
+
+    s->previous_time = 0;
+    s->timing_info_valid = 0;
+    s->read_index_not_before = 0;
+    s->write_index_not_before = 0;
+
+    for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
+        s->write_index_corrections[i].valid = 0;
+    s->current_write_index_correction = 0;
+    
+    s->corked = 0;
+
+    s->cached_time_valid = 0;
+    
+    s->auto_timing_update_event = NULL;
+    s->auto_timing_update_requested = 0;
+
+    /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
+    PA_LLIST_PREPEND(pa_stream, c->streams, s);
+    pa_stream_ref(s);
+
+    return s;
+}
+
+static void stream_free(pa_stream *s) {
+    assert(s && !s->context && !s->channel_valid);
+
+    if (s->auto_timing_update_event) {
+        assert(s->mainloop);
+        s->mainloop->time_free(s->auto_timing_update_event);
+    }
+
+    if (s->peek_memchunk.memblock)
+        pa_memblock_unref(s->peek_memchunk.memblock);
+
+    if (s->record_memblockq)
+        pa_memblockq_free(s->record_memblockq);
+
+    pa_xfree(s->name);
+    pa_xfree(s);
+}
+
+void pa_stream_unref(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    if (--(s->ref) == 0)
+        stream_free(s);
+}
+
+pa_stream* pa_stream_ref(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    s->ref++;
+    return s;
+}
+
+pa_stream_state_t pa_stream_get_state(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    return s->state;
+}
+
+pa_context* pa_stream_get_context(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    return s->context;
+}
+
+uint32_t pa_stream_get_index(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
+    
+    return s->device_index;
+}
+    
+void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    if (s->state == st)
+        return;
+    
+    pa_stream_ref(s);
+
+    s->state = st;
+    if (s->state_callback)
+        s->state_callback(s, s->state_userdata);
+
+    if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
+
+        /* Detach from context */
+        pa_operation *o, *n;
+
+        /* Unref all operatio object that point to us */
+        for (o = s->context->operations; o; o = n) {
+            n = o->next;
+            
+            if (o->stream == s)
+                pa_operation_cancel(o);
+        }
+        
+        /* Drop all outstanding replies for this stream */
+        if (s->context->pdispatch)
+            pa_pdispatch_unregister_reply(s->context->pdispatch, s);
+
+        if (s->channel_valid)
+            pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
+        
+        PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
+        pa_stream_unref(s);
+
+        s->channel = 0;
+        s->channel_valid = 0;
+ 
+        s->context = NULL;
+    }
+
+    pa_stream_unref(s);
+}
+
+void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_context *c = userdata;
+    pa_stream *s;
+    uint32_t channel;
+    
+    assert(pd);
+    assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
+    assert(t);
+    assert(c);
+
+    pa_context_ref(c);
+    
+    if (pa_tagstruct_getu32(t, &channel) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+    
+    if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
+        goto finish;
+
+    pa_context_set_error(c, PA_ERR_KILLED);
+    pa_stream_set_state(s, PA_STREAM_FAILED);
+
+finish:
+    pa_context_unref(c);
+}
+
+void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_stream *s;
+    pa_context *c = userdata;
+    uint32_t bytes, channel;
+    
+    assert(pd);
+    assert(command == PA_COMMAND_REQUEST);
+    assert(t);
+    assert(c);
+
+    pa_context_ref(c);
+    
+    if (pa_tagstruct_getu32(t, &channel) < 0 ||
+        pa_tagstruct_getu32(t, &bytes) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+    
+    if (!(s = pa_dynarray_get(c->playback_streams, channel)))
+        goto finish;
+
+    if (s->state == PA_STREAM_READY) {
+        s->requested_bytes += bytes;
+        
+        if (s->requested_bytes > 0 && s->write_callback)
+            s->write_callback(s, s->requested_bytes, s->write_userdata);
+    }
+
+finish:
+    pa_context_unref(c);
+}
+
+void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_stream *s;
+    pa_context *c = userdata;
+    uint32_t channel;
+
+    assert(pd);
+    assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
+    assert(t);
+    assert(c);
+
+    pa_context_ref(c);
+    
+    if (pa_tagstruct_getu32(t, &channel) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+    
+    if (!(s = pa_dynarray_get(c->playback_streams, channel)))
+        goto finish;
+
+    if (s->state == PA_STREAM_READY) {
+
+        if (command == PA_COMMAND_OVERFLOW) {
+            if (s->overflow_callback)
+                s->overflow_callback(s, s->overflow_userdata);
+        } else if (command == PA_COMMAND_UNDERFLOW) {
+            if (s->underflow_callback)
+                s->underflow_callback(s, s->underflow_userdata);
+        }
+    }
+
+ finish:
+    pa_context_unref(c);
+}
+
+static void request_auto_timing_update(pa_stream *s, int force) {
+    struct timeval next;
+    assert(s);
+
+    if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
+        return;
+    
+    if (s->state == PA_STREAM_READY &&
+        (force || !s->auto_timing_update_requested)) {
+        pa_operation *o;
+        
+/*         pa_log("automatically requesting new timing data");   */
+
+        if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
+            pa_operation_unref(o);
+            s->auto_timing_update_requested = 1;
+        }
+    }
+
+    pa_gettimeofday(&next);
+    pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
+    s->mainloop->time_restart(s->auto_timing_update_event, &next);
+}
+
+static void invalidate_indexes(pa_stream *s, int r, int w) {
+    assert(s);
+
+/*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
+    
+    if (s->state != PA_STREAM_READY)
+        return;
+
+    if (w) {
+        s->write_index_not_before = s->context->ctag;
+        
+        if (s->timing_info_valid)
+            s->timing_info.write_index_corrupt = 1;
+        
+/*         pa_log("write_index invalidated"); */
+    }
+    
+    if (r) {
+        s->read_index_not_before = s->context->ctag;
+        
+        if (s->timing_info_valid)
+            s->timing_info.read_index_corrupt = 1;
+        
+/*         pa_log("read_index invalidated"); */
+    }
+    
+    if ((s->direction == PA_STREAM_PLAYBACK && r) ||
+        (s->direction == PA_STREAM_RECORD && w))
+        s->cached_time_valid = 0;
+    
+    request_auto_timing_update(s, 1);
+}
+
+static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC_UNUSED pa_time_event *e, PA_GCC_UNUSED const struct timeval *tv, void *userdata) {
+    pa_stream *s = userdata;
+
+/*     pa_log("time event");    */
+
+    pa_stream_ref(s);
+    request_auto_timing_update(s, 0);
+    pa_stream_unref(s);
+}
+
+void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_stream *s = userdata;
+    
+    assert(pd);
+    assert(s);
+    assert(s->state == PA_STREAM_CREATING);
+        
+    pa_stream_ref(s);
+    
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(s->context, command, t) < 0)
+            goto finish;
+        
+        pa_stream_set_state(s, PA_STREAM_FAILED);
+        goto finish;
+    }
+
+    if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
+        ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
+        ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0)) {
+        pa_context_fail(s->context, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (pa_context_get_server_protocol_version(s->context) >= 9) {
+        if (s->direction == PA_STREAM_PLAYBACK) {
+            if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
+                pa_tagstruct_getu32(t, &s->buffer_attr.prebuf) < 0 ||
+                pa_tagstruct_getu32(t, &s->buffer_attr.minreq) < 0) {
+                pa_context_fail(s->context, PA_ERR_PROTOCOL);
+                goto finish;
+            }
+        } else if (s->direction == PA_STREAM_RECORD) {
+            if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &s->buffer_attr.fragsize) < 0) {
+                pa_context_fail(s->context, PA_ERR_PROTOCOL);
+                goto finish;
+            }
+        }
+    }
+
+    if (!pa_tagstruct_eof(t)) {
+        pa_context_fail(s->context, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (s->direction == PA_STREAM_RECORD) {
+        assert(!s->record_memblockq);
+        
+        s->record_memblockq = pa_memblockq_new(
+                0, 
+                s->buffer_attr.maxlength,
+                0,
+                pa_frame_size(&s->sample_spec),
+                1,
+                0,
+                NULL,
+                s->context->memblock_stat);
+    }
+
+    s->channel_valid = 1;
+    pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
+
+    pa_stream_set_state(s, PA_STREAM_READY);
+    
+    if (s->direction != PA_STREAM_UPLOAD &&
+        s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
+        struct timeval tv;
+
+        pa_gettimeofday(&tv);
+        tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
+
+        assert(!s->auto_timing_update_event);
+        s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
+
+        request_auto_timing_update(s, 1);
+    }
+
+    if (s->requested_bytes > 0 && s->ref > 1 && s->write_callback)
+        s->write_callback(s, s->requested_bytes, s->write_userdata);
+
+finish:
+    pa_stream_unref(s);
+}
+
+static int create_stream(
+        pa_stream_direction_t direction,
+        pa_stream *s,
+        const char *dev,
+        const pa_buffer_attr *attr,
+        pa_stream_flags_t flags,
+        const pa_cvolume *volume,
+        pa_stream *sync_stream) {
+    
+    pa_tagstruct *t;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+    
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
+                                               PA_STREAM_START_CORKED|
+                                               PA_STREAM_INTERPOLATE_TIMING|
+                                               PA_STREAM_NOT_MONOTONOUS|
+                                               PA_STREAM_AUTO_TIMING_UPDATE : 0))), PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
+
+    pa_stream_ref(s);
+
+    s->direction = direction;
+    s->flags = flags;
+    
+    if (sync_stream)
+        s->syncid = sync_stream->syncid;
+    
+    if (attr)
+        s->buffer_attr = *attr;
+    else {
+        /* half a second */
+        s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
+        s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
+        s->buffer_attr.minreq = s->buffer_attr.tlength/100;
+        s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
+        s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
+    }
+
+    if (!dev)
+        dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
+            &tag);
+    
+    pa_tagstruct_put(
+            t,
+            PA_TAG_STRING, s->name,
+            PA_TAG_SAMPLE_SPEC, &s->sample_spec,
+            PA_TAG_CHANNEL_MAP, &s->channel_map,
+            PA_TAG_U32, PA_INVALID_INDEX,
+            PA_TAG_STRING, dev,
+            PA_TAG_U32, s->buffer_attr.maxlength,
+            PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED),
+            PA_TAG_INVALID);
+    
+    if (s->direction == PA_STREAM_PLAYBACK) {
+        pa_cvolume cv;
+
+        pa_tagstruct_put(
+                t,
+                PA_TAG_U32, s->buffer_attr.tlength,
+                PA_TAG_U32, s->buffer_attr.prebuf,
+                PA_TAG_U32, s->buffer_attr.minreq,
+                PA_TAG_U32, s->syncid,
+                PA_TAG_INVALID);
+
+        if (!volume)
+            volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
+        
+        pa_tagstruct_put_cvolume(t, volume);
+    } else
+        pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
+
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
+
+    pa_stream_set_state(s, PA_STREAM_CREATING);
+    
+    pa_stream_unref(s);
+    return 0;
+}
+
+int pa_stream_connect_playback(
+        pa_stream *s,
+        const char *dev,
+        const pa_buffer_attr *attr,
+        pa_stream_flags_t flags,
+        pa_cvolume *volume,
+        pa_stream *sync_stream) {
+    
+    assert(s);
+    assert(s->ref >= 1);
+    
+    return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
+}
+
+int pa_stream_connect_record(
+        pa_stream *s,
+        const char *dev,
+        const pa_buffer_attr *attr,
+        pa_stream_flags_t flags) {
+    
+    assert(s);
+    assert(s->ref >= 1);
+
+    return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
+}
+
+int pa_stream_write(
+        pa_stream *s,
+        const void *data,
+        size_t length,
+        void (*free_cb)(void *p),
+        int64_t offset,
+        pa_seek_mode_t seek) {
+    
+    pa_memchunk chunk;
+    
+    assert(s);
+    assert(s->ref >= 1);
+    assert(data);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
+
+    if (length <= 0)
+        return 0;
+
+    if (free_cb) 
+        chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, 1, s->context->memblock_stat);
+    else {
+        chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
+        memcpy(chunk.memblock->data, data, length);
+    }
+        
+    chunk.index = 0;
+    chunk.length = length;
+
+    pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
+    pa_memblock_unref(chunk.memblock);
+    
+    if (length < s->requested_bytes)
+        s->requested_bytes -= length;
+    else
+        s->requested_bytes = 0;
+
+    if (s->direction == PA_STREAM_PLAYBACK) {
+        
+        /* Update latency request correction */
+        if (s->write_index_corrections[s->current_write_index_correction].valid) {
+            
+            if (seek == PA_SEEK_ABSOLUTE) {
+                s->write_index_corrections[s->current_write_index_correction].corrupt = 0;
+                s->write_index_corrections[s->current_write_index_correction].absolute = 1;
+                s->write_index_corrections[s->current_write_index_correction].value = offset + length;
+            } else if (seek == PA_SEEK_RELATIVE) {
+                if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
+                    s->write_index_corrections[s->current_write_index_correction].value += offset + length;
+            } else
+                s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
+        }
+        
+        /* Update the write index in the already available latency data */
+        if (s->timing_info_valid) {
+            
+            if (seek == PA_SEEK_ABSOLUTE) {
+                s->timing_info.write_index_corrupt = 0;
+                s->timing_info.write_index = offset + length;
+            } else if (seek == PA_SEEK_RELATIVE) {
+                if (!s->timing_info.write_index_corrupt)
+                    s->timing_info.write_index += offset + length;
+            } else
+                s->timing_info.write_index_corrupt = 1;
+        }
+
+        if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
+            request_auto_timing_update(s, 1);
+    }
+    
+    return 0;
+}
+
+int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
+    assert(s);
+    assert(s->ref >= 1);
+    assert(data);
+    assert(length);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
+    
+    if (!s->peek_memchunk.memblock) {
+
+        if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
+            *data = NULL;
+            *length = 0;
+            return 0;
+        }
+    }
+
+    *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
+    *length = s->peek_memchunk.length;
+    return 0;
+}
+
+int pa_stream_drop(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
+    
+    pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
+
+    /* Fix the simulated local read index */
+    if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
+        s->timing_info.read_index += s->peek_memchunk.length;
+    
+    pa_memblock_unref(s->peek_memchunk.memblock);
+    s->peek_memchunk.length = 0;
+    s->peek_memchunk.index = 0;
+    s->peek_memchunk.memblock = NULL;
+
+    return 0;
+}
+
+size_t pa_stream_writable_size(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
+    
+    return s->requested_bytes;
+}
+
+size_t pa_stream_readable_size(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
+
+    return pa_memblockq_get_length(s->record_memblockq);
+}
+
+pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    pa_tagstruct *t;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
+
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+
+    t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    return o;
+}
+
+static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_operation *o = userdata;
+    struct timeval local, remote, now;
+    pa_timing_info *i;
+    
+    assert(pd);
+    assert(o);
+
+    if (!o->context || !o->stream)
+        goto finish;
+
+    i = &o->stream->timing_info;
+
+/*     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); */
+    
+    o->stream->timing_info_valid = 0;
+    i->write_index_corrupt = 0;
+    i->read_index_corrupt = 0;
+
+/*     pa_log("timing update %u\n", tag); */
+
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(o->context, command, t) < 0)
+            goto finish;
+
+    } else if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
+               pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
+               pa_tagstruct_get_boolean(t, &i->playing) < 0 ||
+               pa_tagstruct_get_timeval(t, &local) < 0 ||
+               pa_tagstruct_get_timeval(t, &remote) < 0 ||
+               pa_tagstruct_gets64(t, &i->write_index) < 0 ||
+               pa_tagstruct_gets64(t, &i->read_index) < 0 ||
+               !pa_tagstruct_eof(t)) {
+        pa_context_fail(o->context, PA_ERR_PROTOCOL);
+        goto finish;
+        
+    } else {
+        o->stream->timing_info_valid = 1;
+
+        pa_gettimeofday(&now);
+        
+        /* Calculcate timestamps */
+        if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
+            /* local and remote seem to have synchronized clocks */
+            
+            if (o->stream->direction == PA_STREAM_PLAYBACK)
+                i->transport_usec = pa_timeval_diff(&remote, &local);
+            else
+                i->transport_usec = pa_timeval_diff(&now, &remote);
+            
+            i->synchronized_clocks = 1;
+            i->timestamp = remote;
+        } else {
+            /* clocks are not synchronized, let's estimate latency then */
+            i->transport_usec = pa_timeval_diff(&now, &local)/2;
+            i->synchronized_clocks = 0;
+            i->timestamp = local;
+            pa_timeval_add(&i->timestamp, i->transport_usec);
+        }
+
+        /* Invalidate read and write indexes if necessary */
+        if (tag < o->stream->read_index_not_before)
+            i->read_index_corrupt = 1;
+
+        if (tag < o->stream->write_index_not_before)
+            i->write_index_corrupt = 1;
+
+        if (o->stream->direction == PA_STREAM_PLAYBACK) {
+            /* Write index correction */
+
+            int n, j;
+            uint32_t ctag = tag;
+
+            /* Go through the saved correction values and add up the total correction.*/
+            
+            for (n = 0, j = o->stream->current_write_index_correction+1;
+                 n < PA_MAX_WRITE_INDEX_CORRECTIONS;
+                 n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
+
+                /* Step over invalid data or out-of-date data */
+                if (!o->stream->write_index_corrections[j].valid ||
+                    o->stream->write_index_corrections[j].tag < ctag)
+                    continue;
+
+                /* Make sure that everything is in order */
+                ctag = o->stream->write_index_corrections[j].tag+1;
+
+                /* Now fix the write index */
+                if (o->stream->write_index_corrections[j].corrupt) {
+                    /* A corrupting seek was made */
+                    i->write_index = 0;
+                    i->write_index_corrupt = 1;
+                } else if (o->stream->write_index_corrections[j].absolute) {
+                    /* An absolute seek was made */
+                    i->write_index = o->stream->write_index_corrections[j].value;
+                    i->write_index_corrupt = 0;
+                } else if (!i->write_index_corrupt) {
+                    /* A relative seek was made */
+                    i->write_index += o->stream->write_index_corrections[j].value;
+                }
+            }
+        }
+
+        if (o->stream->direction == PA_STREAM_RECORD) {
+            /* Read index correction */
+
+            if (!i->read_index_corrupt)
+                i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
+        }
+        
+        o->stream->cached_time_valid = 0;
+    }
+
+    o->stream->auto_timing_update_requested = 0;
+/*     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); */
+    
+    /* Clear old correction entries */
+    if (o->stream->direction == PA_STREAM_PLAYBACK) {
+        int n;
+        
+        for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
+            if (!o->stream->write_index_corrections[n].valid)
+                continue;
+            
+            if (o->stream->write_index_corrections[n].tag <= tag)
+                o->stream->write_index_corrections[n].valid = 0;
+        }
+    }
+
+    if (o->stream->latency_update_callback)
+        o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
+    
+    if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
+        pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
+        cb(o->stream, o->stream->timing_info_valid, o->userdata);
+    }
+    
+finish:
+
+    pa_operation_done(o);
+    pa_operation_unref(o);
+}
+
+pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
+    uint32_t tag;
+    pa_operation *o;
+    pa_tagstruct *t;
+    struct timeval now;
+    int cidx = 0;
+    
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    
+    if (s->direction == PA_STREAM_PLAYBACK) {
+        /* Find a place to store the write_index correction data for this entry */
+        cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
+        
+        /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
+        PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
+    }
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+    
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
+    
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    if (s->direction == PA_STREAM_PLAYBACK) {
+        /* Fill in initial correction data */
+        o->stream->current_write_index_correction = cidx;
+        o->stream->write_index_corrections[cidx].valid = 1;
+        o->stream->write_index_corrections[cidx].tag = tag;
+        o->stream->write_index_corrections[cidx].absolute = 0;
+        o->stream->write_index_corrections[cidx].value = 0;
+        o->stream->write_index_corrections[cidx].corrupt = 0;
+    }
+    
+/*     pa_log("requesting update %u\n", tag); */
+    
+    return o;
+}
+
+void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_stream *s = userdata;
+    
+    assert(pd);
+    assert(s);
+    assert(s->ref >= 1);
+
+    pa_stream_ref(s);
+
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(s->context, command, t) < 0)
+            goto finish;
+
+        pa_stream_set_state(s, PA_STREAM_FAILED);
+        goto finish;
+    } else if (!pa_tagstruct_eof(t)) {
+        pa_context_fail(s->context, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    pa_stream_set_state(s, PA_STREAM_TERMINATED);
+
+finish:
+    pa_stream_unref(s);
+}
+
+int pa_stream_disconnect(pa_stream *s) {
+    pa_tagstruct *t;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
+
+    pa_stream_ref(s);
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
+            (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
+
+    pa_stream_unref(s);
+    return 0;
+}
+
+void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    s->read_callback = cb;
+    s->read_userdata = userdata;
+}
+
+void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    s->write_callback = cb;
+    s->write_userdata = userdata;
+}
+
+void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    s->state_callback = cb;
+    s->state_userdata = userdata;
+}
+
+void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    s->overflow_callback = cb;
+    s->overflow_userdata = userdata;
+}
+
+void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    s->underflow_callback = cb;
+    s->underflow_userdata = userdata;
+}
+
+void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    s->latency_update_callback = cb;
+    s->latency_update_userdata = userdata;
+}
+
+void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_operation *o = userdata;
+    int success = 1;
+    
+    assert(pd);
+    assert(o);
+    assert(o->ref >= 1);
+
+    if (!o->context)
+        goto finish;
+    
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(o->context, command, t) < 0)
+            goto finish;
+
+        success = 0;
+    } else if (!pa_tagstruct_eof(t)) {
+        pa_context_fail(o->context, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (o->callback) {
+        pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
+        cb(o->stream, success, o->userdata);
+    }
+
+finish:
+    pa_operation_done(o);
+    pa_operation_unref(o);
+}
+
+pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    pa_tagstruct *t;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+    
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+
+    s->corked = b;
+    
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_tagstruct_put_boolean(t, !!b);
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    if (s->direction == PA_STREAM_PLAYBACK)
+        invalidate_indexes(s, 1, 0);
+
+    return o;
+}
+
+static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
+    pa_tagstruct *t;
+    pa_operation *o;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+
+    t = pa_tagstruct_command(s->context, command, &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    return o;
+}
+
+pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    
+    if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
+
+        if (s->direction == PA_STREAM_PLAYBACK) {
+            if (s->write_index_corrections[s->current_write_index_correction].valid)
+                s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
+        
+            if (s->timing_info_valid)
+                s->timing_info.write_index_corrupt = 1;
+
+            if (s->buffer_attr.prebuf > 0) 
+                invalidate_indexes(s, 1, 0);
+            else
+                request_auto_timing_update(s, 1);
+        } else
+            invalidate_indexes(s, 0, 1);
+    }
+    
+    return o;
+}
+
+pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
+
+    if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
+        invalidate_indexes(s, 1, 0);
+
+    return o;
+}
+
+pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
+
+    if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
+        invalidate_indexes(s, 1, 0);
+    
+    return o;
+}
+
+pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    pa_tagstruct *t;
+    uint32_t tag;
+    
+    assert(s);
+    assert(s->ref >= 1);
+    assert(name);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_tagstruct_puts(t, name);
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    return o;
+}
+
+int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
+    pa_usec_t usec = 0;
+
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
+
+    if (s->cached_time_valid)
+        /* We alredy calculated the time value for this timing info, so let's reuse it */
+        usec = s->cached_time;
+    else {
+        if (s->direction == PA_STREAM_PLAYBACK) {
+            /* The last byte that was written into the output device
+             * had this time value associated */
+            usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
+
+            if (!s->corked) {
+                /* Because the latency info took a little time to come
+                 * to us, we assume that the real output time is actually
+                 * a little ahead */
+                usec += s->timing_info.transport_usec;
+                
+                /* However, the output device usually maintains a buffer
+                   too, hence the real sample currently played is a little
+                   back  */
+                if (s->timing_info.sink_usec >= usec)
+                    usec = 0;
+                else
+                    usec -= s->timing_info.sink_usec;
+            }
+            
+        } else if (s->direction == PA_STREAM_RECORD) {
+            /* The last byte written into the server side queue had
+             * this time value associated */
+            usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
+
+            if (!s->corked) {
+                /* Add transport latency */
+                usec += s->timing_info.transport_usec;
+                
+                /* Add latency of data in device buffer */
+                usec += s->timing_info.source_usec;
+                
+                /* If this is a monitor source, we need to correct the
+                 * time by the playback device buffer */
+                if (s->timing_info.sink_usec >= usec)
+                    usec = 0;
+                else
+                    usec -= s->timing_info.sink_usec;
+            }
+        }
+
+        s->cached_time = usec;
+        s->cached_time_valid = 1;
+    }
+
+    /* Interpolate if requested */
+    if (s->flags & PA_STREAM_INTERPOLATE_TIMING) {
+
+        /* We just add the time that passed since the latency info was
+         * current */
+        if (!s->corked) {
+            struct timeval now;
+            usec += pa_timeval_diff(pa_gettimeofday(&now), &s->timing_info.timestamp);
+        }
+    }
+
+    /* Make sure the time runs monotonically */
+    if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
+        if (usec < s->previous_time)
+            usec = s->previous_time;
+        else
+            s->previous_time = usec;
+    }
+
+    if (r_usec)
+        *r_usec = usec;
+    
+    return 0;
+}
+
+static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
+    assert(s);
+    assert(s->ref >= 1);
+    
+    if (negative)
+        *negative = 0;
+
+    if (a >= b) 
+        return a-b;
+    else {
+        if (negative && s->direction == PA_STREAM_RECORD) {
+            *negative = 1;
+            return b-a;
+        } else
+            return 0;
+    }
+}
+
+int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
+    pa_usec_t t, c;
+    int r;
+    int64_t cindex;
+
+    assert(s);
+    assert(s->ref >= 1);
+    assert(r_usec);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
+    
+    if ((r = pa_stream_get_time(s, &t)) < 0)
+        return r;
+
+    if (s->direction == PA_STREAM_PLAYBACK)
+        cindex = s->timing_info.write_index;
+    else
+        cindex = s->timing_info.read_index;
+
+    if (cindex < 0)
+        cindex = 0;
+    
+    c = pa_bytes_to_usec(cindex, &s->sample_spec);
+
+    if (s->direction == PA_STREAM_PLAYBACK)
+        *r_usec = time_counter_diff(s, c, t, negative);
+    else
+        *r_usec = time_counter_diff(s, t, c, negative);
+
+    return 0;
+}
+
+const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
+
+    return &s->timing_info;
+}
+
+const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    return &s->sample_spec;
+}
+
+const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    return &s->channel_map;
+}
+
+const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
+    assert(s);
+    assert(s->ref >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context,
+        pa_context_get_server_protocol_version(s->context) >= 9, PA_ERR_NODATA);
+
+    return &s->buffer_attr;
+}
-- 
cgit 


From 1710041eaffc917d45eaeb6143db24a9773842b3 Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Wed, 21 Jun 2006 00:18:43 +0000
Subject: only interpolate when the last timing info told us the stream is
 indeed playing

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1051 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index c8fc09d2..677df009 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -1254,7 +1254,7 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
 
         /* We just add the time that passed since the latency info was
          * current */
-        if (!s->corked) {
+        if (!s->corked && s->timing_info.playing) {
             struct timeval now;
             usec += pa_timeval_diff(pa_gettimeofday(&now), &s->timing_info.timestamp);
         }
-- 
cgit 


From 0e436a6926af56f37a74a03bb5e143e078ca0d55 Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Fri, 18 Aug 2006 19:55:18 +0000
Subject: Rework memory management to allow shared memory data transfer. The
 central idea is to allocate all audio memory blocks from a per-process memory
 pool which is available as read-only SHM segment to other local processes.
 Then, instead of writing the actual audio data to the socket just write
 references to this shared memory pool.

To work optimally all memory blocks should now be of type PA_MEMBLOCK_POOL or
PA_MEMBLOCK_POOL_EXTERNAL. The function pa_memblock_new() now generates memory
blocks of this type by default.



git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1266 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 677df009..180cd096 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -437,8 +437,7 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
                 pa_frame_size(&s->sample_spec),
                 1,
                 0,
-                NULL,
-                s->context->memblock_stat);
+                NULL);
     }
 
     s->channel_valid = 1;
@@ -604,9 +603,9 @@ int pa_stream_write(
         return 0;
 
     if (free_cb) 
-        chunk.memblock = pa_memblock_new_user((void*) data, length, free_cb, 1, s->context->memblock_stat);
+        chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
     else {
-        chunk.memblock = pa_memblock_new(length, s->context->memblock_stat);
+        chunk.memblock = pa_memblock_new(s->context->mempool, length);
         memcpy(chunk.memblock->data, data, length);
     }
         
-- 
cgit 


From d210ebbb09daddb2c8c8e8e77243e088b0b19c4d Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Tue, 26 Sep 2006 23:50:56 +0000
Subject: rework memory block management to be thread-safe and mostly
 lock-free.

pa_memblock is now an opaque structure. Access to its fields is now done
through various accessor functions in a thread-safe manner.

pa_memblock_acquire() and pa_memblock_release() are now used to access the
attached audio data. Why? To allow safe manipulation of the memory pointer
maintained by the memory block. Internally _acquire() and _release() maintain a
reference counter. Please do not confuse this reference counter whith the one
maintained by pa_memblock_ref()/_unref()!

As a side effect this patch removes all direct usages of AO_t and replaces it
with pa_atomic_xxx based code.

This stuff needs some serious testing love. Especially if threads are actively
used.



git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1404 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 20 ++++++++++++++++----
 1 file changed, 16 insertions(+), 4 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 180cd096..d31127d8 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -88,6 +88,7 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->peek_memchunk.index = 0;
     s->peek_memchunk.length = 0;
     s->peek_memchunk.memblock = NULL;
+    s->peek_data = NULL;
 
     s->record_memblockq = NULL;
 
@@ -122,8 +123,11 @@ static void stream_free(pa_stream *s) {
         s->mainloop->time_free(s->auto_timing_update_event);
     }
 
-    if (s->peek_memchunk.memblock)
+    if (s->peek_memchunk.memblock) {
+        if (s->peek_data)
+            pa_memblock_release(s->peek_memchunk.memblock);
         pa_memblock_unref(s->peek_memchunk.memblock);
+    }
 
     if (s->record_memblockq)
         pa_memblockq_free(s->record_memblockq);
@@ -605,8 +609,11 @@ int pa_stream_write(
     if (free_cb) 
         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
     else {
+        void *tdata;
         chunk.memblock = pa_memblock_new(s->context->mempool, length);
-        memcpy(chunk.memblock->data, data, length);
+        tdata = pa_memblock_acquire(chunk.memblock);
+        memcpy(tdata, data, length);
+        pa_memblock_release(chunk.memblock);
     }
         
     chunk.index = 0;
@@ -672,9 +679,12 @@ int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
             *length = 0;
             return 0;
         }
+
+        s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
     }
 
-    *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
+    assert(s->peek_data);
+    *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
     *length = s->peek_memchunk.length;
     return 0;
 }
@@ -692,7 +702,9 @@ int pa_stream_drop(pa_stream *s) {
     /* Fix the simulated local read index */
     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
         s->timing_info.read_index += s->peek_memchunk.length;
-    
+
+    assert(s->peek_data);
+    pa_memblock_release(s->peek_memchunk.memblock);
     pa_memblock_unref(s->peek_memchunk.memblock);
     s->peek_memchunk.length = 0;
     s->peek_memchunk.index = 0;
-- 
cgit 


From 8dc62142765249addf131b058c27f931ede1776b Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Mon, 6 Nov 2006 13:06:01 +0000
Subject: Revert r1404 and keep it on a development branch until it is fully
 tested.

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1409 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 20 ++++----------------
 1 file changed, 4 insertions(+), 16 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index d31127d8..180cd096 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -88,7 +88,6 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->peek_memchunk.index = 0;
     s->peek_memchunk.length = 0;
     s->peek_memchunk.memblock = NULL;
-    s->peek_data = NULL;
 
     s->record_memblockq = NULL;
 
@@ -123,11 +122,8 @@ static void stream_free(pa_stream *s) {
         s->mainloop->time_free(s->auto_timing_update_event);
     }
 
-    if (s->peek_memchunk.memblock) {
-        if (s->peek_data)
-            pa_memblock_release(s->peek_memchunk.memblock);
+    if (s->peek_memchunk.memblock)
         pa_memblock_unref(s->peek_memchunk.memblock);
-    }
 
     if (s->record_memblockq)
         pa_memblockq_free(s->record_memblockq);
@@ -609,11 +605,8 @@ int pa_stream_write(
     if (free_cb) 
         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
     else {
-        void *tdata;
         chunk.memblock = pa_memblock_new(s->context->mempool, length);
-        tdata = pa_memblock_acquire(chunk.memblock);
-        memcpy(tdata, data, length);
-        pa_memblock_release(chunk.memblock);
+        memcpy(chunk.memblock->data, data, length);
     }
         
     chunk.index = 0;
@@ -679,12 +672,9 @@ int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
             *length = 0;
             return 0;
         }
-
-        s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
     }
 
-    assert(s->peek_data);
-    *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
+    *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
     *length = s->peek_memchunk.length;
     return 0;
 }
@@ -702,9 +692,7 @@ int pa_stream_drop(pa_stream *s) {
     /* Fix the simulated local read index */
     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
         s->timing_info.read_index += s->peek_memchunk.length;
-
-    assert(s->peek_data);
-    pa_memblock_release(s->peek_memchunk.memblock);
+    
     pa_memblock_unref(s->peek_memchunk.memblock);
     s->peek_memchunk.length = 0;
     s->peek_memchunk.index = 0;
-- 
cgit 


From 521daf6f0ac4fa6a2fbfb5d523c0c743342dca2b Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Thu, 4 Jan 2007 13:43:45 +0000
Subject: Huge trailing whitespace cleanup. Let's keep the tree pure from here
 on, mmmkay?

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1418 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 238 ++++++++++++++++++++++++++---------------------------
 1 file changed, 119 insertions(+), 119 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 180cd096..009eb3cf 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -2,17 +2,17 @@
 
 /***
   This file is part of PulseAudio.
- 
+
   PulseAudio is free software; you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published
   by the Free Software Foundation; either version 2 of the License,
   or (at your option) any later version.
- 
+
   PulseAudio is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
   General Public License for more details.
- 
+
   You should have received a copy of the GNU Lesser General Public License
   along with PulseAudio; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
@@ -43,7 +43,7 @@
 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
     pa_stream *s;
     int i;
-    
+
     assert(c);
 
     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
@@ -76,7 +76,7 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
         s->channel_map = *map;
     else
         pa_channel_map_init_auto(&s->channel_map, ss->channels, PA_CHANNEL_MAP_DEFAULT);
-    
+
     s->channel = 0;
     s->channel_valid = 0;
     s->syncid = c->csyncid++;
@@ -99,11 +99,11 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
         s->write_index_corrections[i].valid = 0;
     s->current_write_index_correction = 0;
-    
+
     s->corked = 0;
 
     s->cached_time_valid = 0;
-    
+
     s->auto_timing_update_event = NULL;
     s->auto_timing_update_requested = 0;
 
@@ -143,7 +143,7 @@ void pa_stream_unref(pa_stream *s) {
 pa_stream* pa_stream_ref(pa_stream *s) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     s->ref++;
     return s;
 }
@@ -151,14 +151,14 @@ pa_stream* pa_stream_ref(pa_stream *s) {
 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     return s->state;
 }
 
 pa_context* pa_stream_get_context(pa_stream *s) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     return s->context;
 }
 
@@ -167,17 +167,17 @@ uint32_t pa_stream_get_index(pa_stream *s) {
     assert(s->ref >= 1);
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
-    
+
     return s->device_index;
 }
-    
+
 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
     assert(s);
     assert(s->ref >= 1);
 
     if (s->state == st)
         return;
-    
+
     pa_stream_ref(s);
 
     s->state = st;
@@ -192,24 +192,24 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
         /* Unref all operatio object that point to us */
         for (o = s->context->operations; o; o = n) {
             n = o->next;
-            
+
             if (o->stream == s)
                 pa_operation_cancel(o);
         }
-        
+
         /* Drop all outstanding replies for this stream */
         if (s->context->pdispatch)
             pa_pdispatch_unregister_reply(s->context->pdispatch, s);
 
         if (s->channel_valid)
             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
-        
+
         PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
         pa_stream_unref(s);
 
         s->channel = 0;
         s->channel_valid = 0;
- 
+
         s->context = NULL;
     }
 
@@ -220,20 +220,20 @@ void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     pa_context *c = userdata;
     pa_stream *s;
     uint32_t channel;
-    
+
     assert(pd);
     assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
     assert(t);
     assert(c);
 
     pa_context_ref(c);
-    
+
     if (pa_tagstruct_getu32(t, &channel) < 0 ||
         !pa_tagstruct_eof(t)) {
         pa_context_fail(c, PA_ERR_PROTOCOL);
         goto finish;
     }
-    
+
     if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
         goto finish;
 
@@ -248,27 +248,27 @@ void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32
     pa_stream *s;
     pa_context *c = userdata;
     uint32_t bytes, channel;
-    
+
     assert(pd);
     assert(command == PA_COMMAND_REQUEST);
     assert(t);
     assert(c);
 
     pa_context_ref(c);
-    
+
     if (pa_tagstruct_getu32(t, &channel) < 0 ||
         pa_tagstruct_getu32(t, &bytes) < 0 ||
         !pa_tagstruct_eof(t)) {
         pa_context_fail(c, PA_ERR_PROTOCOL);
         goto finish;
     }
-    
+
     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
         goto finish;
 
     if (s->state == PA_STREAM_READY) {
         s->requested_bytes += bytes;
-        
+
         if (s->requested_bytes > 0 && s->write_callback)
             s->write_callback(s, s->requested_bytes, s->write_userdata);
     }
@@ -288,13 +288,13 @@ void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC
     assert(c);
 
     pa_context_ref(c);
-    
+
     if (pa_tagstruct_getu32(t, &channel) < 0 ||
         !pa_tagstruct_eof(t)) {
         pa_context_fail(c, PA_ERR_PROTOCOL);
         goto finish;
     }
-    
+
     if (!(s = pa_dynarray_get(c->playback_streams, channel)))
         goto finish;
 
@@ -319,11 +319,11 @@ static void request_auto_timing_update(pa_stream *s, int force) {
 
     if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
         return;
-    
+
     if (s->state == PA_STREAM_READY &&
         (force || !s->auto_timing_update_requested)) {
         pa_operation *o;
-        
+
 /*         pa_log("automatically requesting new timing data");   */
 
         if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
@@ -341,32 +341,32 @@ static void invalidate_indexes(pa_stream *s, int r, int w) {
     assert(s);
 
 /*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
-    
+
     if (s->state != PA_STREAM_READY)
         return;
 
     if (w) {
         s->write_index_not_before = s->context->ctag;
-        
+
         if (s->timing_info_valid)
             s->timing_info.write_index_corrupt = 1;
-        
+
 /*         pa_log("write_index invalidated"); */
     }
-    
+
     if (r) {
         s->read_index_not_before = s->context->ctag;
-        
+
         if (s->timing_info_valid)
             s->timing_info.read_index_corrupt = 1;
-        
+
 /*         pa_log("read_index invalidated"); */
     }
-    
+
     if ((s->direction == PA_STREAM_PLAYBACK && r) ||
         (s->direction == PA_STREAM_RECORD && w))
         s->cached_time_valid = 0;
-    
+
     request_auto_timing_update(s, 1);
 }
 
@@ -382,17 +382,17 @@ static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC
 
 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s = userdata;
-    
+
     assert(pd);
     assert(s);
     assert(s->state == PA_STREAM_CREATING);
-        
+
     pa_stream_ref(s);
-    
+
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(s->context, command, t) < 0)
             goto finish;
-        
+
         pa_stream_set_state(s, PA_STREAM_FAILED);
         goto finish;
     }
@@ -429,9 +429,9 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
 
     if (s->direction == PA_STREAM_RECORD) {
         assert(!s->record_memblockq);
-        
+
         s->record_memblockq = pa_memblockq_new(
-                0, 
+                0,
                 s->buffer_attr.maxlength,
                 0,
                 pa_frame_size(&s->sample_spec),
@@ -444,7 +444,7 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 
     pa_stream_set_state(s, PA_STREAM_READY);
-    
+
     if (s->direction != PA_STREAM_UPLOAD &&
         s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
         struct timeval tv;
@@ -473,13 +473,13 @@ static int create_stream(
         pa_stream_flags_t flags,
         const pa_cvolume *volume,
         pa_stream *sync_stream) {
-    
+
     pa_tagstruct *t;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
-    
+
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
                                                PA_STREAM_START_CORKED|
@@ -493,10 +493,10 @@ static int create_stream(
 
     s->direction = direction;
     s->flags = flags;
-    
+
     if (sync_stream)
         s->syncid = sync_stream->syncid;
-    
+
     if (attr)
         s->buffer_attr = *attr;
     else {
@@ -515,7 +515,7 @@ static int create_stream(
             s->context,
             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
             &tag);
-    
+
     pa_tagstruct_put(
             t,
             PA_TAG_STRING, s->name,
@@ -526,7 +526,7 @@ static int create_stream(
             PA_TAG_U32, s->buffer_attr.maxlength,
             PA_TAG_BOOLEAN, !!(flags & PA_STREAM_START_CORKED),
             PA_TAG_INVALID);
-    
+
     if (s->direction == PA_STREAM_PLAYBACK) {
         pa_cvolume cv;
 
@@ -540,7 +540,7 @@ static int create_stream(
 
         if (!volume)
             volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
-        
+
         pa_tagstruct_put_cvolume(t, volume);
     } else
         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
@@ -549,7 +549,7 @@ static int create_stream(
     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 
     pa_stream_set_state(s, PA_STREAM_CREATING);
-    
+
     pa_stream_unref(s);
     return 0;
 }
@@ -561,10 +561,10 @@ int pa_stream_connect_playback(
         pa_stream_flags_t flags,
         pa_cvolume *volume,
         pa_stream *sync_stream) {
-    
+
     assert(s);
     assert(s->ref >= 1);
-    
+
     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
 }
 
@@ -573,7 +573,7 @@ int pa_stream_connect_record(
         const char *dev,
         const pa_buffer_attr *attr,
         pa_stream_flags_t flags) {
-    
+
     assert(s);
     assert(s->ref >= 1);
 
@@ -587,9 +587,9 @@ int pa_stream_write(
         void (*free_cb)(void *p),
         int64_t offset,
         pa_seek_mode_t seek) {
-    
+
     pa_memchunk chunk;
-    
+
     assert(s);
     assert(s->ref >= 1);
     assert(data);
@@ -602,29 +602,29 @@ int pa_stream_write(
     if (length <= 0)
         return 0;
 
-    if (free_cb) 
+    if (free_cb)
         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
     else {
         chunk.memblock = pa_memblock_new(s->context->mempool, length);
         memcpy(chunk.memblock->data, data, length);
     }
-        
+
     chunk.index = 0;
     chunk.length = length;
 
     pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
     pa_memblock_unref(chunk.memblock);
-    
+
     if (length < s->requested_bytes)
         s->requested_bytes -= length;
     else
         s->requested_bytes = 0;
 
     if (s->direction == PA_STREAM_PLAYBACK) {
-        
+
         /* Update latency request correction */
         if (s->write_index_corrections[s->current_write_index_correction].valid) {
-            
+
             if (seek == PA_SEEK_ABSOLUTE) {
                 s->write_index_corrections[s->current_write_index_correction].corrupt = 0;
                 s->write_index_corrections[s->current_write_index_correction].absolute = 1;
@@ -635,10 +635,10 @@ int pa_stream_write(
             } else
                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
         }
-        
+
         /* Update the write index in the already available latency data */
         if (s->timing_info_valid) {
-            
+
             if (seek == PA_SEEK_ABSOLUTE) {
                 s->timing_info.write_index_corrupt = 0;
                 s->timing_info.write_index = offset + length;
@@ -652,7 +652,7 @@ int pa_stream_write(
         if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
             request_auto_timing_update(s, 1);
     }
-    
+
     return 0;
 }
 
@@ -664,7 +664,7 @@ int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
-    
+
     if (!s->peek_memchunk.memblock) {
 
         if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
@@ -686,13 +686,13 @@ int pa_stream_drop(pa_stream *s) {
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
-    
+
     pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
 
     /* Fix the simulated local read index */
     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
         s->timing_info.read_index += s->peek_memchunk.length;
-    
+
     pa_memblock_unref(s->peek_memchunk.memblock);
     s->peek_memchunk.length = 0;
     s->peek_memchunk.index = 0;
@@ -707,7 +707,7 @@ size_t pa_stream_writable_size(pa_stream *s) {
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
-    
+
     return s->requested_bytes;
 }
 
@@ -725,7 +725,7 @@ pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *us
     pa_operation *o;
     pa_tagstruct *t;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
 
@@ -746,7 +746,7 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
     pa_operation *o = userdata;
     struct timeval local, remote, now;
     pa_timing_info *i;
-    
+
     assert(pd);
     assert(o);
 
@@ -756,7 +756,7 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
     i = &o->stream->timing_info;
 
 /*     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); */
-    
+
     o->stream->timing_info_valid = 0;
     i->write_index_corrupt = 0;
     i->read_index_corrupt = 0;
@@ -777,21 +777,21 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
                !pa_tagstruct_eof(t)) {
         pa_context_fail(o->context, PA_ERR_PROTOCOL);
         goto finish;
-        
+
     } else {
         o->stream->timing_info_valid = 1;
 
         pa_gettimeofday(&now);
-        
+
         /* Calculcate timestamps */
         if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
             /* local and remote seem to have synchronized clocks */
-            
+
             if (o->stream->direction == PA_STREAM_PLAYBACK)
                 i->transport_usec = pa_timeval_diff(&remote, &local);
             else
                 i->transport_usec = pa_timeval_diff(&now, &remote);
-            
+
             i->synchronized_clocks = 1;
             i->timestamp = remote;
         } else {
@@ -816,7 +816,7 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
             uint32_t ctag = tag;
 
             /* Go through the saved correction values and add up the total correction.*/
-            
+
             for (n = 0, j = o->stream->current_write_index_correction+1;
                  n < PA_MAX_WRITE_INDEX_CORRECTIONS;
                  n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
@@ -851,21 +851,21 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
             if (!i->read_index_corrupt)
                 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
         }
-        
+
         o->stream->cached_time_valid = 0;
     }
 
     o->stream->auto_timing_update_requested = 0;
 /*     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); */
-    
+
     /* Clear old correction entries */
     if (o->stream->direction == PA_STREAM_PLAYBACK) {
         int n;
-        
+
         for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
             if (!o->stream->write_index_corrections[n].valid)
                 continue;
-            
+
             if (o->stream->write_index_corrections[n].tag <= tag)
                 o->stream->write_index_corrections[n].valid = 0;
         }
@@ -873,12 +873,12 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
 
     if (o->stream->latency_update_callback)
         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
-    
+
     if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
         pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
         cb(o->stream, o->stream->timing_info_valid, o->userdata);
     }
-    
+
 finish:
 
     pa_operation_done(o);
@@ -891,29 +891,29 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
     pa_tagstruct *t;
     struct timeval now;
     int cidx = 0;
-    
+
     assert(s);
     assert(s->ref >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
-    
+
     if (s->direction == PA_STREAM_PLAYBACK) {
         /* Find a place to store the write_index correction data for this entry */
         cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
-        
+
         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
     }
     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
-    
+
     t = pa_tagstruct_command(
             s->context,
             s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
             &tag);
     pa_tagstruct_putu32(t, s->channel);
     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
-    
+
     pa_pstream_send_tagstruct(s->context->pstream, t);
     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
@@ -926,15 +926,15 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
         o->stream->write_index_corrections[cidx].value = 0;
         o->stream->write_index_corrections[cidx].corrupt = 0;
     }
-    
+
 /*     pa_log("requesting update %u\n", tag); */
-    
+
     return o;
 }
 
 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s = userdata;
-    
+
     assert(pd);
     assert(s);
     assert(s->ref >= 1);
@@ -961,7 +961,7 @@ finish:
 int pa_stream_disconnect(pa_stream *s) {
     pa_tagstruct *t;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
 
@@ -1010,7 +1010,7 @@ void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *
 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     s->overflow_callback = cb;
     s->overflow_userdata = userdata;
 }
@@ -1018,7 +1018,7 @@ void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, voi
 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     s->underflow_callback = cb;
     s->underflow_userdata = userdata;
 }
@@ -1026,7 +1026,7 @@ void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, vo
 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     s->latency_update_callback = cb;
     s->latency_update_userdata = userdata;
 }
@@ -1034,14 +1034,14 @@ void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t c
 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_operation *o = userdata;
     int success = 1;
-    
+
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
 
     if (!o->context)
         goto finish;
-    
+
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
@@ -1066,15 +1066,15 @@ pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, voi
     pa_operation *o;
     pa_tagstruct *t;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
-    
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 
     s->corked = b;
-    
+
     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 
     t = pa_tagstruct_command(
@@ -1096,12 +1096,12 @@ static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command,
     pa_tagstruct *t;
     pa_operation *o;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
-    
+
     o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
 
     t = pa_tagstruct_command(s->context, command, &tag);
@@ -1114,26 +1114,26 @@ static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command,
 
 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
     pa_operation *o;
-    
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
-    
+
     if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
 
         if (s->direction == PA_STREAM_PLAYBACK) {
             if (s->write_index_corrections[s->current_write_index_correction].valid)
                 s->write_index_corrections[s->current_write_index_correction].corrupt = 1;
-        
+
             if (s->timing_info_valid)
                 s->timing_info.write_index_corrupt = 1;
 
-            if (s->buffer_attr.prebuf > 0) 
+            if (s->buffer_attr.prebuf > 0)
                 invalidate_indexes(s, 1, 0);
             else
                 request_auto_timing_update(s, 1);
         } else
             invalidate_indexes(s, 0, 1);
     }
-    
+
     return o;
 }
 
@@ -1151,13 +1151,13 @@ pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *us
 
 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
     pa_operation *o;
-    
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
 
     if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
         invalidate_indexes(s, 1, 0);
-    
+
     return o;
 }
 
@@ -1165,7 +1165,7 @@ pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_succe
     pa_operation *o;
     pa_tagstruct *t;
     uint32_t tag;
-    
+
     assert(s);
     assert(s->ref >= 1);
     assert(name);
@@ -1213,7 +1213,7 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
                  * to us, we assume that the real output time is actually
                  * a little ahead */
                 usec += s->timing_info.transport_usec;
-                
+
                 /* However, the output device usually maintains a buffer
                    too, hence the real sample currently played is a little
                    back  */
@@ -1222,7 +1222,7 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
                 else
                     usec -= s->timing_info.sink_usec;
             }
-            
+
         } else if (s->direction == PA_STREAM_RECORD) {
             /* The last byte written into the server side queue had
              * this time value associated */
@@ -1231,10 +1231,10 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
             if (!s->corked) {
                 /* Add transport latency */
                 usec += s->timing_info.transport_usec;
-                
+
                 /* Add latency of data in device buffer */
                 usec += s->timing_info.source_usec;
-                
+
                 /* If this is a monitor source, we need to correct the
                  * time by the playback device buffer */
                 if (s->timing_info.sink_usec >= usec)
@@ -1269,18 +1269,18 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
 
     if (r_usec)
         *r_usec = usec;
-    
+
     return 0;
 }
 
 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
     assert(s);
     assert(s->ref >= 1);
-    
+
     if (negative)
         *negative = 0;
 
-    if (a >= b) 
+    if (a >= b)
         return a-b;
     else {
         if (negative && s->direction == PA_STREAM_RECORD) {
@@ -1305,7 +1305,7 @@ int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
     PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
-    
+
     if ((r = pa_stream_get_time(s, &t)) < 0)
         return r;
 
@@ -1316,7 +1316,7 @@ int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
 
     if (cindex < 0)
         cindex = 0;
-    
+
     c = pa_bytes_to_usec(cindex, &s->sample_spec);
 
     if (s->direction == PA_STREAM_PLAYBACK)
-- 
cgit 


From 06211b7c8fd329137ae9003818543912a87d9898 Mon Sep 17 00:00:00 2001
From: Pierre Ossman <ossman@cendio.se>
Date: Tue, 13 Feb 2007 15:35:19 +0000
Subject: Add copyright notices to all relevant files. (based on svn log)

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1426 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 3 +++
 1 file changed, 3 insertions(+)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 009eb3cf..f20c17ae 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -3,6 +3,9 @@
 /***
   This file is part of PulseAudio.
 
+  Copyright 2004-2006 Lennart Poettering
+  Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
+
   PulseAudio is free software; you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published
   by the Free Software Foundation; either version 2 of the License,
-- 
cgit 


From a67c21f093202f142438689d3f7cfbdf4ea82eea Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Sun, 28 Oct 2007 19:13:50 +0000
Subject: merge 'lennart' branch back into trunk.

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@1971 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 319 +++++++++++++++++++++++++++++++----------------------
 1 file changed, 189 insertions(+), 130 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index f20c17ae..47906a5c 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -26,7 +26,6 @@
 #include <config.h>
 #endif
 
-#include <assert.h>
 #include <string.h>
 #include <stdio.h>
 #include <string.h>
@@ -38,6 +37,7 @@
 #include <pulsecore/pstream-util.h>
 #include <pulsecore/log.h>
 #include <pulsecore/hashmap.h>
+#include <pulsecore/macro.h>
 
 #include "internal.h"
 
@@ -47,13 +47,14 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     pa_stream *s;
     int i;
 
-    assert(c);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
 
     s = pa_xnew(pa_stream, 1);
-    s->ref = 1;
+    PA_REFCNT_INIT(s);
     s->context = c;
     s->mainloop = c->mainloop;
 
@@ -91,6 +92,7 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->peek_memchunk.index = 0;
     s->peek_memchunk.length = 0;
     s->peek_memchunk.memblock = NULL;
+    s->peek_data = NULL;
 
     s->record_memblockq = NULL;
 
@@ -118,15 +120,20 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
 }
 
 static void stream_free(pa_stream *s) {
-    assert(s && !s->context && !s->channel_valid);
+    pa_assert(s);
+    pa_assert(!s->context);
+    pa_assert(!s->channel_valid);
 
     if (s->auto_timing_update_event) {
-        assert(s->mainloop);
+        pa_assert(s->mainloop);
         s->mainloop->time_free(s->auto_timing_update_event);
     }
 
-    if (s->peek_memchunk.memblock)
+    if (s->peek_memchunk.memblock) {
+        if (s->peek_data)
+            pa_memblock_release(s->peek_memchunk.memblock);
         pa_memblock_unref(s->peek_memchunk.memblock);
+    }
 
     if (s->record_memblockq)
         pa_memblockq_free(s->record_memblockq);
@@ -136,38 +143,38 @@ static void stream_free(pa_stream *s) {
 }
 
 void pa_stream_unref(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
-    if (--(s->ref) == 0)
+    if (PA_REFCNT_DEC(s) <= 0)
         stream_free(s);
 }
 
 pa_stream* pa_stream_ref(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
-    s->ref++;
+    PA_REFCNT_INC(s);
     return s;
 }
 
 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return s->state;
 }
 
 pa_context* pa_stream_get_context(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return s->context;
 }
 
 uint32_t pa_stream_get_index(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
 
@@ -175,8 +182,8 @@ uint32_t pa_stream_get_index(pa_stream *s) {
 }
 
 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     if (s->state == st)
         return;
@@ -214,6 +221,13 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
         s->channel_valid = 0;
 
         s->context = NULL;
+
+        s->read_callback = NULL;
+        s->write_callback = NULL;
+        s->state_callback = NULL;
+        s->overflow_callback = NULL;
+        s->underflow_callback = NULL;
+        s->latency_update_callback = NULL;
     }
 
     pa_stream_unref(s);
@@ -224,10 +238,11 @@ void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     pa_stream *s;
     uint32_t channel;
 
-    assert(pd);
-    assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
-    assert(t);
-    assert(c);
+    pa_assert(pd);
+    pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
+    pa_assert(t);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
 
     pa_context_ref(c);
 
@@ -252,10 +267,11 @@ void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32
     pa_context *c = userdata;
     uint32_t bytes, channel;
 
-    assert(pd);
-    assert(command == PA_COMMAND_REQUEST);
-    assert(t);
-    assert(c);
+    pa_assert(pd);
+    pa_assert(command == PA_COMMAND_REQUEST);
+    pa_assert(t);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
 
     pa_context_ref(c);
 
@@ -285,10 +301,11 @@ void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC
     pa_context *c = userdata;
     uint32_t channel;
 
-    assert(pd);
-    assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
-    assert(t);
-    assert(c);
+    pa_assert(pd);
+    pa_assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
+    pa_assert(t);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
 
     pa_context_ref(c);
 
@@ -317,8 +334,8 @@ void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC
 }
 
 static void request_auto_timing_update(pa_stream *s, int force) {
-    struct timeval next;
-    assert(s);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
         return;
@@ -335,13 +352,17 @@ static void request_auto_timing_update(pa_stream *s, int force) {
         }
     }
 
-    pa_gettimeofday(&next);
-    pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
-    s->mainloop->time_restart(s->auto_timing_update_event, &next);
+    if (s->auto_timing_update_event) {
+        struct timeval next;
+        pa_gettimeofday(&next);
+        pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
+        s->mainloop->time_restart(s->auto_timing_update_event, &next);
+    }
 }
 
 static void invalidate_indexes(pa_stream *s, int r, int w) {
-    assert(s);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
 /*     pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
 
@@ -376,6 +397,9 @@ static void invalidate_indexes(pa_stream *s, int r, int w) {
 static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC_UNUSED pa_time_event *e, PA_GCC_UNUSED const struct timeval *tv, void *userdata) {
     pa_stream *s = userdata;
 
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
 /*     pa_log("time event");    */
 
     pa_stream_ref(s);
@@ -383,12 +407,32 @@ static void auto_timing_update_callback(PA_GCC_UNUSED pa_mainloop_api *m, PA_GCC
     pa_stream_unref(s);
 }
 
+static void create_stream_complete(pa_stream *s) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(s->state == PA_STREAM_CREATING);
+
+    pa_stream_set_state(s, PA_STREAM_READY);
+
+    if (s->requested_bytes > 0 && s->write_callback)
+        s->write_callback(s, s->requested_bytes, s->write_userdata);
+
+    if (s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
+        struct timeval tv;
+        pa_gettimeofday(&tv);
+        tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
+        pa_assert(!s->auto_timing_update_event);
+        s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
+    }
+}
+
 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s = userdata;
 
-    assert(pd);
-    assert(s);
-    assert(s->state == PA_STREAM_CREATING);
+    pa_assert(pd);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(s->state == PA_STREAM_CREATING);
 
     pa_stream_ref(s);
 
@@ -431,7 +475,7 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     }
 
     if (s->direction == PA_STREAM_RECORD) {
-        assert(!s->record_memblockq);
+        pa_assert(!s->record_memblockq);
 
         s->record_memblockq = pa_memblockq_new(
                 0,
@@ -446,23 +490,16 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     s->channel_valid = 1;
     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 
-    pa_stream_set_state(s, PA_STREAM_READY);
-
-    if (s->direction != PA_STREAM_UPLOAD &&
-        s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
-        struct timeval tv;
-
-        pa_gettimeofday(&tv);
-        tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
-
-        assert(!s->auto_timing_update_event);
-        s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
-
+    if (s->direction != PA_STREAM_UPLOAD && s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
+        /* If automatic timing updates are active, we wait for the
+         * first timing update before going to PA_STREAM_READY
+         * state */
+        s->state = PA_STREAM_READY;
         request_auto_timing_update(s, 1);
-    }
+        s->state = PA_STREAM_CREATING;
 
-    if (s->requested_bytes > 0 && s->ref > 1 && s->write_callback)
-        s->write_callback(s, s->requested_bytes, s->write_userdata);
+    } else
+        create_stream_complete(s);
 
 finish:
     pa_stream_unref(s);
@@ -480,8 +517,8 @@ static int create_stream(
     pa_tagstruct *t;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, !(flags & ~((direction != PA_STREAM_UPLOAD ?
@@ -503,12 +540,12 @@ static int create_stream(
     if (attr)
         s->buffer_attr = *attr;
     else {
-        /* half a second */
+        /* half a second, with minimum request of 10 ms */
         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
-        s->buffer_attr.minreq = s->buffer_attr.tlength/100;
+        s->buffer_attr.minreq = s->buffer_attr.tlength/50;
         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
-        s->buffer_attr.fragsize = s->buffer_attr.tlength/100;
+        s->buffer_attr.fragsize = s->buffer_attr.tlength/50;
     }
 
     if (!dev)
@@ -565,8 +602,8 @@ int pa_stream_connect_playback(
         pa_cvolume *volume,
         pa_stream *sync_stream) {
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
 }
@@ -577,8 +614,8 @@ int pa_stream_connect_record(
         const pa_buffer_attr *attr,
         pa_stream_flags_t flags) {
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
 }
@@ -593,9 +630,9 @@ int pa_stream_write(
 
     pa_memchunk chunk;
 
-    assert(s);
-    assert(s->ref >= 1);
-    assert(data);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(data);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -608,8 +645,11 @@ int pa_stream_write(
     if (free_cb)
         chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) data, length, free_cb, 1);
     else {
+        void *tdata;
         chunk.memblock = pa_memblock_new(s->context->mempool, length);
-        memcpy(chunk.memblock->data, data, length);
+        tdata = pa_memblock_acquire(chunk.memblock);
+        memcpy(tdata, data, length);
+        pa_memblock_release(chunk.memblock);
     }
 
     chunk.index = 0;
@@ -660,10 +700,10 @@ int pa_stream_write(
 }
 
 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
-    assert(s);
-    assert(s->ref >= 1);
-    assert(data);
-    assert(length);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(data);
+    pa_assert(length);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
@@ -675,27 +715,32 @@ int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
             *length = 0;
             return 0;
         }
+
+        s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
     }
 
-    *data = (const char*) s->peek_memchunk.memblock->data + s->peek_memchunk.index;
+    pa_assert(s->peek_data);
+    *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
     *length = s->peek_memchunk.length;
     return 0;
 }
 
 int pa_stream_drop(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
 
-    pa_memblockq_drop(s->record_memblockq, &s->peek_memchunk, s->peek_memchunk.length);
+    pa_memblockq_drop(s->record_memblockq, s->peek_memchunk.length);
 
     /* Fix the simulated local read index */
     if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
         s->timing_info.read_index += s->peek_memchunk.length;
 
+    pa_assert(s->peek_data);
+    pa_memblock_release(s->peek_memchunk.memblock);
     pa_memblock_unref(s->peek_memchunk.memblock);
     s->peek_memchunk.length = 0;
     s->peek_memchunk.index = 0;
@@ -705,8 +750,8 @@ int pa_stream_drop(pa_stream *s) {
 }
 
 size_t pa_stream_writable_size(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
@@ -715,8 +760,8 @@ size_t pa_stream_writable_size(pa_stream *s) {
 }
 
 size_t pa_stream_readable_size(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
@@ -729,8 +774,8 @@ pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *us
     pa_tagstruct *t;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
@@ -750,8 +795,9 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
     struct timeval local, remote, now;
     pa_timing_info *i;
 
-    assert(pd);
-    assert(o);
+    pa_assert(pd);
+    pa_assert(o);
+    pa_assert(PA_REFCNT_VALUE(o) >= 1);
 
     if (!o->context || !o->stream)
         goto finish;
@@ -874,6 +920,10 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
         }
     }
 
+    /* First, let's complete the initialization, if necessary. */
+    if (o->stream->state == PA_STREAM_CREATING)
+        create_stream_complete(o->stream);
+
     if (o->stream->latency_update_callback)
         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
 
@@ -895,8 +945,8 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
     struct timeval now;
     int cidx = 0;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -938,9 +988,9 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s = userdata;
 
-    assert(pd);
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(pd);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     pa_stream_ref(s);
 
@@ -965,8 +1015,8 @@ int pa_stream_disconnect(pa_stream *s) {
     pa_tagstruct *t;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
@@ -987,48 +1037,48 @@ int pa_stream_disconnect(pa_stream *s) {
 }
 
 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->read_callback = cb;
     s->read_userdata = userdata;
 }
 
 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->write_callback = cb;
     s->write_userdata = userdata;
 }
 
 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->state_callback = cb;
     s->state_userdata = userdata;
 }
 
 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->overflow_callback = cb;
     s->overflow_userdata = userdata;
 }
 
 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->underflow_callback = cb;
     s->underflow_userdata = userdata;
 }
 
 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     s->latency_update_callback = cb;
     s->latency_update_userdata = userdata;
@@ -1038,9 +1088,9 @@ void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UN
     pa_operation *o = userdata;
     int success = 1;
 
-    assert(pd);
-    assert(o);
-    assert(o->ref >= 1);
+    pa_assert(pd);
+    pa_assert(o);
+    pa_assert(PA_REFCNT_VALUE(o) >= 1);
 
     if (!o->context)
         goto finish;
@@ -1070,8 +1120,8 @@ pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, voi
     pa_tagstruct *t;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -1100,8 +1150,8 @@ static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command,
     pa_operation *o;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
 
@@ -1118,6 +1168,9 @@ static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command,
 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
     pa_operation *o;
 
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
 
     if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
@@ -1143,6 +1196,9 @@ pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *use
 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
     pa_operation *o;
 
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
 
@@ -1155,6 +1211,9 @@ pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *us
 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
     pa_operation *o;
 
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
 
@@ -1169,9 +1228,9 @@ pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_succe
     pa_tagstruct *t;
     uint32_t tag;
 
-    assert(s);
-    assert(s->ref >= 1);
-    assert(name);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(name);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -1193,8 +1252,8 @@ pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_succe
 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
     pa_usec_t usec = 0;
 
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -1277,8 +1336,8 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
 }
 
 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     if (negative)
         *negative = 0;
@@ -1299,9 +1358,9 @@ int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
     int r;
     int64_t cindex;
 
-    assert(s);
-    assert(s->ref >= 1);
-    assert(r_usec);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(r_usec);
 
     PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -1331,8 +1390,8 @@ int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
 }
 
 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
@@ -1342,22 +1401,22 @@ const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
 }
 
 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return &s->sample_spec;
 }
 
 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     return &s->channel_map;
 }
 
 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
-    assert(s);
-    assert(s->ref >= 1);
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
-- 
cgit 


From 14a9b80afbb0bddc216462b72156f14e032e1b5e Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Wed, 21 Nov 2007 01:30:40 +0000
Subject: - Check process name when dealing with PID files - Add new
 PA_STREAM_FIX_CHANNELS, FIX_RATE, FIX_FORMAT, DONT_MOVE, VARIABLE_RATES to
 pa_sream_flags_t adn implement it - Expose those flags in pacat - Add
 notifications about device suspend/resume to the protocol and expose them in
 libpulse - Allow changing of buffer_attr during playback - allow disabling
 for remixing globally - hookup polkit support

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@2067 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 436 +++++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 424 insertions(+), 12 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 47906a5c..92420825 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -51,6 +51,7 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     pa_assert(PA_REFCNT_VALUE(c) >= 1);
 
     PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
+    PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 12 || (ss->format != PA_SAMPLE_S32LE || ss->format != PA_SAMPLE_S32NE), PA_ERR_NOTSUPPORTED);
     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
 
     s = pa_xnew(pa_stream, 1);
@@ -58,6 +59,8 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->context = c;
     s->mainloop = c->mainloop;
 
+    s->buffer_attr_not_ready = s->timing_info_not_ready = FALSE;
+
     s->read_callback = NULL;
     s->read_userdata = NULL;
     s->write_callback = NULL;
@@ -70,6 +73,10 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->underflow_userdata = NULL;
     s->latency_update_callback = NULL;
     s->latency_update_userdata = NULL;
+    s->moved_callback = NULL;
+    s->moved_userdata = NULL;
+    s->suspended_callback = NULL;
+    s->suspended_userdata = NULL;
 
     s->direction = PA_STREAM_NODIRECTION;
     s->name = pa_xstrdup(name);
@@ -84,11 +91,17 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->channel = 0;
     s->channel_valid = 0;
     s->syncid = c->csyncid++;
-    s->device_index = PA_INVALID_INDEX;
+    s->stream_index = PA_INVALID_INDEX;
     s->requested_bytes = 0;
     s->state = PA_STREAM_UNCONNECTED;
+
+    s->manual_buffer_attr = FALSE;
     memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
 
+    s->device_index = PA_INVALID_INDEX;
+    s->device_name = NULL;
+    s->suspended = FALSE;
+
     s->peek_memchunk.index = 0;
     s->peek_memchunk.length = 0;
     s->peek_memchunk.memblock = NULL;
@@ -139,6 +152,7 @@ static void stream_free(pa_stream *s) {
         pa_memblockq_free(s->record_memblockq);
 
     pa_xfree(s->name);
+    pa_xfree(s->device_name);
     pa_xfree(s);
 }
 
@@ -178,7 +192,7 @@ uint32_t pa_stream_get_index(pa_stream *s) {
 
     PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
 
-    return s->device_index;
+    return s->stream_index;
 }
 
 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
@@ -262,6 +276,95 @@ finish:
     pa_context_unref(c);
 }
 
+void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_context *c = userdata;
+    pa_stream *s;
+    uint32_t channel;
+    const char *dn;
+    int suspended;
+    uint32_t di;
+
+    pa_assert(pd);
+    pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED);
+    pa_assert(t);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
+
+    pa_context_ref(c);
+
+    if (c->version < 12) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (pa_tagstruct_getu32(t, &channel) < 0 ||
+        pa_tagstruct_getu32(t, &di) < 0 ||
+        pa_tagstruct_gets(t, &dn) < 0 ||
+        pa_tagstruct_get_boolean(t, &suspended) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (!dn || di == PA_INVALID_INDEX) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_MOVED ? c->playback_streams : c->record_streams, channel)))
+        goto finish;
+
+    pa_xfree(s->device_name);
+    s->device_name = pa_xstrdup(dn);
+    s->device_index = di;
+
+    s->suspended = suspended;
+
+    if (s->moved_callback)
+        s->moved_callback(s, s->moved_userdata);
+
+finish:
+    pa_context_unref(c);
+}
+
+void pa_command_stream_suspended(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_context *c = userdata;
+    pa_stream *s;
+    uint32_t channel;
+    int suspended;
+
+    pa_assert(pd);
+    pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED || command == PA_COMMAND_RECORD_STREAM_SUSPENDED);
+    pa_assert(t);
+    pa_assert(c);
+    pa_assert(PA_REFCNT_VALUE(c) >= 1);
+
+    pa_context_ref(c);
+
+    if (c->version < 12) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (pa_tagstruct_getu32(t, &channel) < 0 ||
+        pa_tagstruct_get_boolean(t, &suspended) < 0 ||
+        !pa_tagstruct_eof(t)) {
+        pa_context_fail(c, PA_ERR_PROTOCOL);
+        goto finish;
+    }
+
+    if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED ? c->playback_streams : c->record_streams, channel)))
+        goto finish;
+
+    s->suspended = suspended;
+
+    if (s->suspended_callback)
+        s->suspended_callback(s, s->suspended_userdata);
+
+finish:
+    pa_context_unref(c);
+}
+
 void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s;
     pa_context *c = userdata;
@@ -412,6 +515,9 @@ static void create_stream_complete(pa_stream *s) {
     pa_assert(PA_REFCNT_VALUE(s) >= 1);
     pa_assert(s->state == PA_STREAM_CREATING);
 
+    if (s->buffer_attr_not_ready || s->timing_info_not_ready)
+        return;
+
     pa_stream_set_state(s, PA_STREAM_READY);
 
     if (s->requested_bytes > 0 && s->write_callback)
@@ -426,6 +532,17 @@ static void create_stream_complete(pa_stream *s) {
     }
 }
 
+static void automatic_buffer_attr(pa_buffer_attr *attr, pa_sample_spec *ss) {
+    pa_assert(attr);
+    pa_assert(ss);
+
+    attr->tlength = pa_bytes_per_second(ss)/2;
+    attr->maxlength = (attr->tlength*3)/2;
+    attr->minreq = attr->tlength/50;
+    attr->prebuf = attr->tlength - attr->minreq;
+    attr->fragsize = attr->tlength/50;
+}
+
 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_stream *s = userdata;
 
@@ -445,13 +562,13 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     }
 
     if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
-        ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->device_index) < 0) ||
+        ((s->direction != PA_STREAM_UPLOAD) && pa_tagstruct_getu32(t, &s->stream_index) < 0) ||
         ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0)) {
         pa_context_fail(s->context, PA_ERR_PROTOCOL);
         goto finish;
     }
 
-    if (pa_context_get_server_protocol_version(s->context) >= 9) {
+    if (s->context->version >= 9) {
         if (s->direction == PA_STREAM_PLAYBACK) {
             if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
                 pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
@@ -469,6 +586,58 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
         }
     }
 
+    if (s->context->version >= 12) {
+        pa_sample_spec ss;
+        pa_channel_map cm;
+        const char *dn = NULL;
+        int suspended;
+
+        if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
+            pa_tagstruct_get_channel_map(t, &cm) < 0 ||
+            pa_tagstruct_getu32(t, &s->device_index) < 0 ||
+            pa_tagstruct_gets(t, &dn) < 0 ||
+            pa_tagstruct_get_boolean(t, &suspended) < 0) {
+            pa_context_fail(s->context, PA_ERR_PROTOCOL);
+            goto finish;
+        }
+
+        if (!dn || s->device_index == PA_INVALID_INDEX ||
+            ss.channels != cm.channels ||
+            !pa_channel_map_valid(&cm) ||
+            !pa_sample_spec_valid(&ss) ||
+            (!(s->flags & PA_STREAM_FIX_FORMAT) && ss.format != s->sample_spec.format) ||
+            (!(s->flags & PA_STREAM_FIX_RATE) && ss.rate != s->sample_spec.rate) ||
+            (!(s->flags & PA_STREAM_FIX_CHANNELS) && !pa_channel_map_equal(&cm, &s->channel_map))) {
+            pa_context_fail(s->context, PA_ERR_PROTOCOL);
+            goto finish;
+        }
+
+        pa_xfree(s->device_name);
+        s->device_name = pa_xstrdup(dn);
+        s->suspended = suspended;
+
+        if (!s->manual_buffer_attr && pa_bytes_per_second(&ss) != pa_bytes_per_second(&s->sample_spec)) {
+            pa_buffer_attr attr;
+            pa_operation *o;
+
+            automatic_buffer_attr(&attr, &ss);
+
+            /* If we need to update the buffer metrics, we wait for
+             * the the OK for that call before we go to
+             * PA_STREAM_READY */
+
+            s->state = PA_STREAM_READY;
+            pa_assert_se(o = pa_stream_set_buffer_attr(s, &attr, NULL, NULL));
+            pa_operation_unref(o);
+            s->state = PA_STREAM_CREATING;
+
+            s->buffer_attr_not_ready = TRUE;
+        }
+
+        s->channel_map = cm;
+        s->sample_spec = ss;
+    }
+
     if (!pa_tagstruct_eof(t)) {
         pa_context_fail(s->context, PA_ERR_PROTOCOL);
         goto finish;
@@ -491,15 +660,19 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
     pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
 
     if (s->direction != PA_STREAM_UPLOAD && s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
+
         /* If automatic timing updates are active, we wait for the
          * first timing update before going to PA_STREAM_READY
          * state */
+
         s->state = PA_STREAM_READY;
         request_auto_timing_update(s, 1);
         s->state = PA_STREAM_CREATING;
 
-    } else
-        create_stream_complete(s);
+        s->timing_info_not_ready = TRUE;
+    }
+
+    create_stream_complete(s);
 
 finish:
     pa_stream_unref(s);
@@ -525,7 +698,13 @@ static int create_stream(
                                                PA_STREAM_START_CORKED|
                                                PA_STREAM_INTERPOLATE_TIMING|
                                                PA_STREAM_NOT_MONOTONOUS|
-                                               PA_STREAM_AUTO_TIMING_UPDATE : 0))), PA_ERR_INVALID);
+                                               PA_STREAM_AUTO_TIMING_UPDATE|
+                                               PA_STREAM_NO_REMAP_CHANNELS|
+                                               PA_STREAM_NO_REMIX_CHANNELS|
+                                               PA_STREAM_FIX_FORMAT|
+                                               PA_STREAM_FIX_RATE|
+                                               PA_STREAM_FIX_CHANNELS|
+                                               PA_STREAM_DONT_MOVE : 0))), PA_ERR_INVALID);
     PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
     PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
 
@@ -537,15 +716,17 @@ static int create_stream(
     if (sync_stream)
         s->syncid = sync_stream->syncid;
 
-    if (attr)
+    if (attr) {
         s->buffer_attr = *attr;
-    else {
+        s->manual_buffer_attr = TRUE;
+    } else {
         /* half a second, with minimum request of 10 ms */
         s->buffer_attr.tlength = pa_bytes_per_second(&s->sample_spec)/2;
         s->buffer_attr.maxlength = (s->buffer_attr.tlength*3)/2;
         s->buffer_attr.minreq = s->buffer_attr.tlength/50;
         s->buffer_attr.prebuf = s->buffer_attr.tlength - s->buffer_attr.minreq;
         s->buffer_attr.fragsize = s->buffer_attr.tlength/50;
+        s->manual_buffer_attr = FALSE;
     }
 
     if (!dev)
@@ -585,6 +766,19 @@ static int create_stream(
     } else
         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 
+    if (s->context->version >= 12 && s->direction != PA_STREAM_UPLOAD) {
+        pa_tagstruct_put(
+                t,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMAP_CHANNELS,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMIX_CHANNELS,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_FORMAT,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_RATE,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_CHANNELS,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_DONT_MOVE,
+                PA_TAG_BOOLEAN, flags & PA_STREAM_VARIABLE_RATE,
+                PA_TAG_INVALID);
+    }
+
     pa_pstream_send_tagstruct(s->context->pstream, t);
     pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 
@@ -921,8 +1115,10 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
     }
 
     /* First, let's complete the initialization, if necessary. */
-    if (o->stream->state == PA_STREAM_CREATING)
+    if (o->stream->state == PA_STREAM_CREATING) {
+        o->stream->timing_info_not_ready = FALSE;
         create_stream_complete(o->stream);
+    }
 
     if (o->stream->latency_update_callback)
         o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
@@ -1084,6 +1280,22 @@ void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t c
     s->latency_update_userdata = userdata;
 }
 
+void pa_stream_set_moved_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    s->moved_callback = cb;
+    s->moved_userdata = userdata;
+}
+
+void pa_stream_set_suspended_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    s->suspended_callback = cb;
+    s->suspended_userdata = userdata;
+}
+
 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
     pa_operation *o = userdata;
     int success = 1;
@@ -1420,8 +1632,208 @@ const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
 
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
     PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
-    PA_CHECK_VALIDITY_RETURN_NULL(s->context,
-        pa_context_get_server_protocol_version(s->context) >= 9, PA_ERR_NODATA);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 9, PA_ERR_NODATA);
 
     return &s->buffer_attr;
 }
+
+static void stream_set_buffer_attr_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_operation *o = userdata;
+    int success = 1;
+
+    pa_assert(pd);
+    pa_assert(o);
+    pa_assert(PA_REFCNT_VALUE(o) >= 1);
+
+    if (!o->context)
+        goto finish;
+
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(o->context, command, t) < 0)
+            goto finish;
+
+        success = 0;
+    } else {
+
+        if (o->stream->direction == PA_STREAM_PLAYBACK) {
+            if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &o->stream->buffer_attr.tlength) < 0 ||
+                pa_tagstruct_getu32(t, &o->stream->buffer_attr.prebuf) < 0 ||
+                pa_tagstruct_getu32(t, &o->stream->buffer_attr.minreq) < 0) {
+                pa_context_fail(o->context, PA_ERR_PROTOCOL);
+                goto finish;
+            }
+        } else if (o->stream->direction == PA_STREAM_RECORD) {
+            if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
+                pa_tagstruct_getu32(t, &o->stream->buffer_attr.fragsize) < 0) {
+                pa_context_fail(o->context, PA_ERR_PROTOCOL);
+                goto finish;
+            }
+        }
+
+        if (!pa_tagstruct_eof(t)) {
+            pa_context_fail(o->context, PA_ERR_PROTOCOL);
+            goto finish;
+        }
+
+        o->stream->manual_buffer_attr = TRUE;
+    }
+
+    if (o->stream->state == PA_STREAM_CREATING) {
+        o->stream->buffer_attr_not_ready = FALSE;
+        create_stream_complete(o->stream);
+    }
+
+    if (o->callback) {
+        pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
+        cb(o->stream, success, o->userdata);
+    }
+
+finish:
+    pa_operation_done(o);
+    pa_operation_unref(o);
+}
+
+
+pa_operation* pa_stream_set_buffer_attr(pa_stream *s, const pa_buffer_attr *attr, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    pa_tagstruct *t;
+    uint32_t tag;
+
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+    pa_assert(attr);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
+
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR : PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR,
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+
+    pa_tagstruct_putu32(t, attr->maxlength);
+
+    if (s->direction == PA_STREAM_PLAYBACK)
+        pa_tagstruct_put(
+                t,
+                PA_TAG_U32, attr->tlength,
+                PA_TAG_U32, attr->prebuf,
+                PA_TAG_U32, attr->minreq,
+                PA_TAG_INVALID);
+    else
+        pa_tagstruct_putu32(t, attr->fragsize);
+
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_set_buffer_attr_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    return o;
+}
+
+uint32_t pa_stream_get_device_index(pa_stream *s) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, PA_INVALID_INDEX);
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
+    PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->device_index != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
+
+    return s->device_index;
+}
+
+const char *pa_stream_get_device_name(pa_stream *s) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->device_name, PA_ERR_BADSTATE);
+
+    return s->device_name;
+}
+
+int pa_stream_is_suspended(pa_stream *s) {
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
+
+    return s->suspended;
+}
+
+static void stream_update_sample_rate_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
+    pa_operation *o = userdata;
+    int success = 1;
+
+    pa_assert(pd);
+    pa_assert(o);
+    pa_assert(PA_REFCNT_VALUE(o) >= 1);
+
+    if (!o->context)
+        goto finish;
+
+    if (command != PA_COMMAND_REPLY) {
+        if (pa_context_handle_error(o->context, command, t) < 0)
+            goto finish;
+
+        success = 0;
+    } else {
+
+        if (!pa_tagstruct_eof(t)) {
+            pa_context_fail(o->context, PA_ERR_PROTOCOL);
+            goto finish;
+        }
+    }
+
+    o->stream->sample_spec.rate = PA_PTR_TO_UINT(o->private);
+    pa_assert(pa_sample_spec_valid(&o->stream->sample_spec));
+
+    if (o->callback) {
+        pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
+        cb(o->stream, success, o->userdata);
+    }
+
+finish:
+    pa_operation_done(o);
+    pa_operation_unref(o);
+}
+
+
+pa_operation *pa_stream_update_sample_rate(pa_stream *s, uint32_t rate, pa_stream_success_cb_t cb, void *userdata) {
+    pa_operation *o;
+    pa_tagstruct *t;
+    uint32_t tag;
+
+    pa_assert(s);
+    pa_assert(PA_REFCNT_VALUE(s) >= 1);
+
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, rate > 0 && rate <= PA_RATE_MAX, PA_ERR_INVALID);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->flags & PA_STREAM_VARIABLE_RATE, PA_ERR_BADSTATE);
+    PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
+
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+    o->private = PA_UINT_TO_PTR(rate);
+
+    t = pa_tagstruct_command(
+            s->context,
+            s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE : PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE,
+            &tag);
+    pa_tagstruct_putu32(t, s->channel);
+    pa_tagstruct_putu32(t, rate);
+
+    pa_pstream_send_tagstruct(s->context->pstream, t);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_update_sample_rate_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
+
+    return o;
+
+}
-- 
cgit 


From 9d2255d4e5c2fcecee4bf654cc35fba2a5b6a399 Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Thu, 22 Nov 2007 15:09:00 +0000
Subject: fix uploading of samples into PA. Problem discovered by Colin Guthrie

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@2074 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index 92420825..cd70cdcb 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -586,7 +586,7 @@ void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED
         }
     }
 
-    if (s->context->version >= 12) {
+    if (s->context->version >= 12 && s->direction != PA_STREAM_UPLOAD) {
         pa_sample_spec ss;
         pa_channel_map cm;
         const char *dn = NULL;
-- 
cgit 


From 86b9ef8c961bed9d3a65f044741bb423c26d8005 Mon Sep 17 00:00:00 2001
From: Lennart Poettering <lennart@poettering.net>
Date: Wed, 13 Feb 2008 22:13:44 +0000
Subject: deal with a possibly failing pa_channel_map_init_auto() correctly

git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@2105 fefdeb5f-60dc-0310-8127-8f9354f1896f
---
 src/pulse/stream.c | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

(limited to 'src/pulse/stream.c')

diff --git a/src/pulse/stream.c b/src/pulse/stream.c
index cd70cdcb..c44323fc 100644
--- a/src/pulse/stream.c
+++ b/src/pulse/stream.c
@@ -46,6 +46,7 @@
 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
     pa_stream *s;
     int i;
+    pa_channel_map tmap;
 
     pa_assert(c);
     pa_assert(PA_REFCNT_VALUE(c) >= 1);
@@ -54,6 +55,9 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 12 || (ss->format != PA_SAMPLE_S32LE || ss->format != PA_SAMPLE_S32NE), PA_ERR_NOTSUPPORTED);
     PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
 
+    if (!map)
+        PA_CHECK_VALIDITY_RETURN_NULL(c, map = pa_channel_map_init_auto(&tmap, ss->channels, PA_CHANNEL_MAP_DEFAULT), PA_ERR_INVALID);
+
     s = pa_xnew(pa_stream, 1);
     PA_REFCNT_INIT(s);
     s->context = c;
@@ -81,13 +85,9 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->direction = PA_STREAM_NODIRECTION;
     s->name = pa_xstrdup(name);
     s->sample_spec = *ss;
+    s->channel_map = *map;
     s->flags = 0;
 
-    if (map)
-        s->channel_map = *map;
-    else
-        pa_channel_map_init_auto(&s->channel_map, ss->channels, PA_CHANNEL_MAP_DEFAULT);
-
     s->channel = 0;
     s->channel_valid = 0;
     s->syncid = c->csyncid++;
-- 
cgit