2 This file is part of PulseAudio.
4 Copyright 2004-2006 Lennart Poettering
5 Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
7 PulseAudio is free software; you can redistribute it and/or modify
8 it under the terms of the GNU Lesser General Public License as published
9 by the Free Software Foundation; either version 2.1 of the License,
10 or (at your option) any later version.
12 PulseAudio is distributed in the hope that it will be useful, but
13 WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 General Public License for more details.
17 You should have received a copy of the GNU Lesser General Public License
18 along with PulseAudio; if not, write to the Free Software
19 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
31 #include <pulse/def.h>
32 #include <pulse/timeval.h>
33 #include <pulse/rtclock.h>
34 #include <pulse/xmalloc.h>
35 #include <pulse/fork-detect.h>
37 #include <pulsecore/pstream-util.h>
38 #include <pulsecore/log.h>
39 #include <pulsecore/hashmap.h>
40 #include <pulsecore/macro.h>
41 #include <pulsecore/core-rtclock.h>
42 #include <pulsecore/core-util.h>
47 #define AUTO_TIMING_INTERVAL_START_USEC (10*PA_USEC_PER_MSEC)
48 #define AUTO_TIMING_INTERVAL_END_USEC (1500*PA_USEC_PER_MSEC)
50 #define SMOOTHER_ADJUST_TIME (1000*PA_USEC_PER_MSEC)
51 #define SMOOTHER_HISTORY_TIME (5000*PA_USEC_PER_MSEC)
52 #define SMOOTHER_MIN_HISTORY (4)
54 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
55 return pa_stream_new_with_proplist(c, name, ss, map, NULL);
58 static void reset_callbacks(pa_stream *s) {
59 s->read_callback = NULL;
60 s->read_userdata = NULL;
61 s->write_callback = NULL;
62 s->write_userdata = NULL;
63 s->state_callback = NULL;
64 s->state_userdata = NULL;
65 s->overflow_callback = NULL;
66 s->overflow_userdata = NULL;
67 s->underflow_callback = NULL;
68 s->underflow_userdata = NULL;
69 s->latency_update_callback = NULL;
70 s->latency_update_userdata = NULL;
71 s->moved_callback = NULL;
72 s->moved_userdata = NULL;
73 s->suspended_callback = NULL;
74 s->suspended_userdata = NULL;
75 s->started_callback = NULL;
76 s->started_userdata = NULL;
77 s->event_callback = NULL;
78 s->event_userdata = NULL;
79 s->buffer_attr_callback = NULL;
80 s->buffer_attr_userdata = NULL;
83 static pa_stream *pa_stream_new_with_proplist_internal(
86 const pa_sample_spec *ss,
87 const pa_channel_map *map,
88 pa_format_info * const *formats,
89 unsigned int n_formats,
96 pa_assert(PA_REFCNT_VALUE(c) >= 1);
97 pa_assert((ss == NULL && map == NULL) || (formats == NULL && n_formats == 0));
98 pa_assert(n_formats < PA_MAX_FORMATS);
100 PA_CHECK_VALIDITY_RETURN_NULL(c, !pa_detect_fork(), PA_ERR_FORKED);
101 PA_CHECK_VALIDITY_RETURN_NULL(c, name || (p && pa_proplist_contains(p, PA_PROP_MEDIA_NAME)), PA_ERR_INVALID);
103 s = pa_xnew(pa_stream, 1);
106 s->mainloop = c->mainloop;
108 s->direction = PA_STREAM_NODIRECTION;
109 s->state = PA_STREAM_UNCONNECTED;
113 s->sample_spec = *ss;
115 s->sample_spec.format = PA_SAMPLE_INVALID;
118 s->channel_map = *map;
120 pa_channel_map_init(&s->channel_map);
124 s->n_formats = n_formats;
125 for (i = 0; i < n_formats; i++)
126 s->req_formats[i] = pa_format_info_copy(formats[i]);
129 /* We'll get the final negotiated format after connecting */
132 s->direct_on_input = PA_INVALID_INDEX;
134 s->proplist = p ? pa_proplist_copy(p) : pa_proplist_new();
136 pa_proplist_sets(s->proplist, PA_PROP_MEDIA_NAME, name);
139 s->channel_valid = FALSE;
140 s->syncid = c->csyncid++;
141 s->stream_index = PA_INVALID_INDEX;
143 s->requested_bytes = 0;
144 memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
146 /* We initialize der target length here, so that if the user
147 * passes no explicit buffering metrics the default is similar to
148 * what older PA versions provided. */
150 s->buffer_attr.maxlength = (uint32_t) -1;
152 s->buffer_attr.tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
154 /* FIXME: We assume a worst-case compressed format corresponding to
155 * 48000 Hz, 2 ch, S16 PCM, but this can very well be incorrect */
156 pa_sample_spec tmp_ss = {
157 .format = PA_SAMPLE_S16NE,
161 s->buffer_attr.tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, &tmp_ss); /* 250ms of buffering */
163 s->buffer_attr.minreq = (uint32_t) -1;
164 s->buffer_attr.prebuf = (uint32_t) -1;
165 s->buffer_attr.fragsize = (uint32_t) -1;
167 s->device_index = PA_INVALID_INDEX;
168 s->device_name = NULL;
169 s->suspended = FALSE;
172 s->write_memblock = NULL;
173 s->write_data = NULL;
175 pa_memchunk_reset(&s->peek_memchunk);
177 s->record_memblockq = NULL;
179 memset(&s->timing_info, 0, sizeof(s->timing_info));
180 s->timing_info_valid = FALSE;
182 s->previous_time = 0;
183 s->latest_underrun_at_index = -1;
185 s->read_index_not_before = 0;
186 s->write_index_not_before = 0;
187 for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
188 s->write_index_corrections[i].valid = 0;
189 s->current_write_index_correction = 0;
191 s->auto_timing_update_event = NULL;
192 s->auto_timing_update_requested = FALSE;
193 s->auto_timing_interval_usec = AUTO_TIMING_INTERVAL_START_USEC;
199 /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
200 PA_LLIST_PREPEND(pa_stream, c->streams, s);
206 pa_stream *pa_stream_new_with_proplist(
209 const pa_sample_spec *ss,
210 const pa_channel_map *map,
215 PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
216 PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 12 || (ss->format != PA_SAMPLE_S32LE && ss->format != PA_SAMPLE_S32BE), PA_ERR_NOTSUPPORTED);
217 PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 15 || (ss->format != PA_SAMPLE_S24LE && ss->format != PA_SAMPLE_S24BE), PA_ERR_NOTSUPPORTED);
218 PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 15 || (ss->format != PA_SAMPLE_S24_32LE && ss->format != PA_SAMPLE_S24_32BE), PA_ERR_NOTSUPPORTED);
219 PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
222 PA_CHECK_VALIDITY_RETURN_NULL(c, map = pa_channel_map_init_auto(&tmap, ss->channels, PA_CHANNEL_MAP_DEFAULT), PA_ERR_INVALID);
224 return pa_stream_new_with_proplist_internal(c, name, ss, map, NULL, 0, p);
227 pa_stream *pa_stream_new_extended(
230 pa_format_info * const *formats,
231 unsigned int n_formats,
234 PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 21, PA_ERR_NOTSUPPORTED);
236 return pa_stream_new_with_proplist_internal(c, name, NULL, NULL, formats, n_formats, p);
239 static void stream_unlink(pa_stream *s) {
246 /* Detach from context */
248 /* Unref all operation object that point to us */
249 for (o = s->context->operations; o; o = n) {
253 pa_operation_cancel(o);
256 /* Drop all outstanding replies for this stream */
257 if (s->context->pdispatch)
258 pa_pdispatch_unregister_reply(s->context->pdispatch, s);
260 if (s->channel_valid) {
261 pa_hashmap_remove((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, PA_UINT32_TO_PTR(s->channel));
263 s->channel_valid = FALSE;
266 PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
271 if (s->auto_timing_update_event) {
272 pa_assert(s->mainloop);
273 s->mainloop->time_free(s->auto_timing_update_event);
279 static void stream_free(pa_stream *s) {
286 if (s->write_memblock) {
287 pa_memblock_release(s->write_memblock);
288 pa_memblock_unref(s->write_data);
291 if (s->peek_memchunk.memblock) {
293 pa_memblock_release(s->peek_memchunk.memblock);
294 pa_memblock_unref(s->peek_memchunk.memblock);
297 if (s->record_memblockq)
298 pa_memblockq_free(s->record_memblockq);
301 pa_proplist_free(s->proplist);
304 pa_smoother_free(s->smoother);
306 for (i = 0; i < s->n_formats; i++)
307 pa_format_info_free(s->req_formats[i]);
310 pa_format_info_free(s->format);
312 pa_xfree(s->device_name);
316 void pa_stream_unref(pa_stream *s) {
318 pa_assert(PA_REFCNT_VALUE(s) >= 1);
320 if (PA_REFCNT_DEC(s) <= 0)
324 pa_stream* pa_stream_ref(pa_stream *s) {
326 pa_assert(PA_REFCNT_VALUE(s) >= 1);
332 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
334 pa_assert(PA_REFCNT_VALUE(s) >= 1);
339 pa_context* pa_stream_get_context(pa_stream *s) {
341 pa_assert(PA_REFCNT_VALUE(s) >= 1);
346 uint32_t pa_stream_get_index(pa_stream *s) {
348 pa_assert(PA_REFCNT_VALUE(s) >= 1);
350 PA_CHECK_VALIDITY_RETURN_ANY(s->context, !pa_detect_fork(), PA_ERR_FORKED, PA_INVALID_INDEX);
351 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
353 return s->stream_index;
356 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
358 pa_assert(PA_REFCNT_VALUE(s) >= 1);
367 if (s->state_callback)
368 s->state_callback(s, s->state_userdata);
370 if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED))
376 static void request_auto_timing_update(pa_stream *s, pa_bool_t force) {
378 pa_assert(PA_REFCNT_VALUE(s) >= 1);
380 if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
383 if (s->state == PA_STREAM_READY &&
384 (force || !s->auto_timing_update_requested)) {
387 /* pa_log("Automatically requesting new timing data"); */
389 if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
390 pa_operation_unref(o);
391 s->auto_timing_update_requested = TRUE;
395 if (s->auto_timing_update_event) {
396 if (s->suspended && !force) {
397 pa_assert(s->mainloop);
398 s->mainloop->time_free(s->auto_timing_update_event);
399 s->auto_timing_update_event = NULL;
402 s->auto_timing_interval_usec = AUTO_TIMING_INTERVAL_START_USEC;
404 pa_context_rttime_restart(s->context, s->auto_timing_update_event, pa_rtclock_now() + s->auto_timing_interval_usec);
406 s->auto_timing_interval_usec = PA_MIN(AUTO_TIMING_INTERVAL_END_USEC, s->auto_timing_interval_usec*2);
411 void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
412 pa_context *c = userdata;
417 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
420 pa_assert(PA_REFCNT_VALUE(c) >= 1);
424 if (pa_tagstruct_getu32(t, &channel) < 0 ||
425 !pa_tagstruct_eof(t)) {
426 pa_context_fail(c, PA_ERR_PROTOCOL);
430 if (!(s = pa_hashmap_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, PA_UINT32_TO_PTR(channel))))
433 if (s->state != PA_STREAM_READY)
436 pa_context_set_error(c, PA_ERR_KILLED);
437 pa_stream_set_state(s, PA_STREAM_FAILED);
443 static void check_smoother_status(pa_stream *s, pa_bool_t aposteriori, pa_bool_t force_start, pa_bool_t force_stop) {
447 pa_assert(!force_start || !force_stop);
452 x = pa_rtclock_now();
454 if (s->timing_info_valid) {
456 x -= s->timing_info.transport_usec;
458 x += s->timing_info.transport_usec;
461 if (s->suspended || s->corked || force_stop)
462 pa_smoother_pause(s->smoother, x);
463 else if (force_start || s->buffer_attr.prebuf == 0) {
465 if (!s->timing_info_valid &&
469 s->context->version >= 13) {
471 /* If the server supports STARTED events we take them as
472 * indications when audio really starts/stops playing, if
473 * we don't have any timing info yet -- instead of trying
474 * to be smart and guessing the server time. Otherwise the
475 * unknown transport delay add too much noise to our time
481 pa_smoother_resume(s->smoother, x, TRUE);
484 /* Please note that we have no idea if playback actually started
485 * if prebuf is non-zero! */
488 static void auto_timing_update_callback(pa_mainloop_api *m, pa_time_event *e, const struct timeval *t, void *userdata);
490 void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
491 pa_context *c = userdata;
498 uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0;
501 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED);
504 pa_assert(PA_REFCNT_VALUE(c) >= 1);
508 if (c->version < 12) {
509 pa_context_fail(c, PA_ERR_PROTOCOL);
513 if (pa_tagstruct_getu32(t, &channel) < 0 ||
514 pa_tagstruct_getu32(t, &di) < 0 ||
515 pa_tagstruct_gets(t, &dn) < 0 ||
516 pa_tagstruct_get_boolean(t, &suspended) < 0) {
517 pa_context_fail(c, PA_ERR_PROTOCOL);
521 if (c->version >= 13) {
523 if (command == PA_COMMAND_RECORD_STREAM_MOVED) {
524 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
525 pa_tagstruct_getu32(t, &fragsize) < 0 ||
526 pa_tagstruct_get_usec(t, &usec) < 0) {
527 pa_context_fail(c, PA_ERR_PROTOCOL);
531 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
532 pa_tagstruct_getu32(t, &tlength) < 0 ||
533 pa_tagstruct_getu32(t, &prebuf) < 0 ||
534 pa_tagstruct_getu32(t, &minreq) < 0 ||
535 pa_tagstruct_get_usec(t, &usec) < 0) {
536 pa_context_fail(c, PA_ERR_PROTOCOL);
542 if (!pa_tagstruct_eof(t)) {
543 pa_context_fail(c, PA_ERR_PROTOCOL);
547 if (!dn || di == PA_INVALID_INDEX) {
548 pa_context_fail(c, PA_ERR_PROTOCOL);
552 if (!(s = pa_hashmap_get(command == PA_COMMAND_PLAYBACK_STREAM_MOVED ? c->playback_streams : c->record_streams, PA_UINT32_TO_PTR(channel))))
555 if (s->state != PA_STREAM_READY)
558 if (c->version >= 13) {
559 if (s->direction == PA_STREAM_RECORD)
560 s->timing_info.configured_source_usec = usec;
562 s->timing_info.configured_sink_usec = usec;
564 s->buffer_attr.maxlength = maxlength;
565 s->buffer_attr.fragsize = fragsize;
566 s->buffer_attr.tlength = tlength;
567 s->buffer_attr.prebuf = prebuf;
568 s->buffer_attr.minreq = minreq;
571 pa_xfree(s->device_name);
572 s->device_name = pa_xstrdup(dn);
573 s->device_index = di;
575 s->suspended = suspended;
577 if ((s->flags & PA_STREAM_AUTO_TIMING_UPDATE) && !suspended && !s->auto_timing_update_event) {
578 s->auto_timing_interval_usec = AUTO_TIMING_INTERVAL_START_USEC;
579 s->auto_timing_update_event = pa_context_rttime_new(s->context, pa_rtclock_now() + s->auto_timing_interval_usec, &auto_timing_update_callback, s);
580 request_auto_timing_update(s, TRUE);
583 check_smoother_status(s, TRUE, FALSE, FALSE);
584 request_auto_timing_update(s, TRUE);
586 if (s->moved_callback)
587 s->moved_callback(s, s->moved_userdata);
593 void pa_command_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
594 pa_context *c = userdata;
598 uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0;
601 pa_assert(command == PA_COMMAND_PLAYBACK_BUFFER_ATTR_CHANGED || command == PA_COMMAND_RECORD_BUFFER_ATTR_CHANGED);
604 pa_assert(PA_REFCNT_VALUE(c) >= 1);
608 if (c->version < 15) {
609 pa_context_fail(c, PA_ERR_PROTOCOL);
613 if (pa_tagstruct_getu32(t, &channel) < 0) {
614 pa_context_fail(c, PA_ERR_PROTOCOL);
618 if (command == PA_COMMAND_RECORD_STREAM_MOVED) {
619 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
620 pa_tagstruct_getu32(t, &fragsize) < 0 ||
621 pa_tagstruct_get_usec(t, &usec) < 0) {
622 pa_context_fail(c, PA_ERR_PROTOCOL);
626 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
627 pa_tagstruct_getu32(t, &tlength) < 0 ||
628 pa_tagstruct_getu32(t, &prebuf) < 0 ||
629 pa_tagstruct_getu32(t, &minreq) < 0 ||
630 pa_tagstruct_get_usec(t, &usec) < 0) {
631 pa_context_fail(c, PA_ERR_PROTOCOL);
636 if (!pa_tagstruct_eof(t)) {
637 pa_context_fail(c, PA_ERR_PROTOCOL);
641 if (!(s = pa_hashmap_get(command == PA_COMMAND_PLAYBACK_BUFFER_ATTR_CHANGED ? c->playback_streams : c->record_streams, PA_UINT32_TO_PTR(channel))))
644 if (s->state != PA_STREAM_READY)
647 if (s->direction == PA_STREAM_RECORD)
648 s->timing_info.configured_source_usec = usec;
650 s->timing_info.configured_sink_usec = usec;
652 s->buffer_attr.maxlength = maxlength;
653 s->buffer_attr.fragsize = fragsize;
654 s->buffer_attr.tlength = tlength;
655 s->buffer_attr.prebuf = prebuf;
656 s->buffer_attr.minreq = minreq;
658 request_auto_timing_update(s, TRUE);
660 if (s->buffer_attr_callback)
661 s->buffer_attr_callback(s, s->buffer_attr_userdata);
667 void pa_command_stream_suspended(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
668 pa_context *c = userdata;
674 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED || command == PA_COMMAND_RECORD_STREAM_SUSPENDED);
677 pa_assert(PA_REFCNT_VALUE(c) >= 1);
681 if (c->version < 12) {
682 pa_context_fail(c, PA_ERR_PROTOCOL);
686 if (pa_tagstruct_getu32(t, &channel) < 0 ||
687 pa_tagstruct_get_boolean(t, &suspended) < 0 ||
688 !pa_tagstruct_eof(t)) {
689 pa_context_fail(c, PA_ERR_PROTOCOL);
693 if (!(s = pa_hashmap_get(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED ? c->playback_streams : c->record_streams, PA_UINT32_TO_PTR(channel))))
696 if (s->state != PA_STREAM_READY)
699 s->suspended = suspended;
701 if ((s->flags & PA_STREAM_AUTO_TIMING_UPDATE) && !suspended && !s->auto_timing_update_event) {
702 s->auto_timing_interval_usec = AUTO_TIMING_INTERVAL_START_USEC;
703 s->auto_timing_update_event = pa_context_rttime_new(s->context, pa_rtclock_now() + s->auto_timing_interval_usec, &auto_timing_update_callback, s);
704 request_auto_timing_update(s, TRUE);
707 check_smoother_status(s, TRUE, FALSE, FALSE);
708 request_auto_timing_update(s, TRUE);
710 if (s->suspended_callback)
711 s->suspended_callback(s, s->suspended_userdata);
717 void pa_command_stream_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
718 pa_context *c = userdata;
723 pa_assert(command == PA_COMMAND_STARTED);
726 pa_assert(PA_REFCNT_VALUE(c) >= 1);
730 if (c->version < 13) {
731 pa_context_fail(c, PA_ERR_PROTOCOL);
735 if (pa_tagstruct_getu32(t, &channel) < 0 ||
736 !pa_tagstruct_eof(t)) {
737 pa_context_fail(c, PA_ERR_PROTOCOL);
741 if (!(s = pa_hashmap_get(c->playback_streams, PA_UINT32_TO_PTR(channel))))
744 if (s->state != PA_STREAM_READY)
747 check_smoother_status(s, TRUE, TRUE, FALSE);
748 request_auto_timing_update(s, TRUE);
750 if (s->started_callback)
751 s->started_callback(s, s->started_userdata);
757 void pa_command_stream_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
758 pa_context *c = userdata;
761 pa_proplist *pl = NULL;
765 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_EVENT || command == PA_COMMAND_RECORD_STREAM_EVENT);
768 pa_assert(PA_REFCNT_VALUE(c) >= 1);
772 if (c->version < 15) {
773 pa_context_fail(c, PA_ERR_PROTOCOL);
777 pl = pa_proplist_new();
779 if (pa_tagstruct_getu32(t, &channel) < 0 ||
780 pa_tagstruct_gets(t, &event) < 0 ||
781 pa_tagstruct_get_proplist(t, pl) < 0 ||
782 !pa_tagstruct_eof(t) || !event) {
783 pa_context_fail(c, PA_ERR_PROTOCOL);
787 if (!(s = pa_hashmap_get(command == PA_COMMAND_PLAYBACK_STREAM_EVENT ? c->playback_streams : c->record_streams, PA_UINT32_TO_PTR(channel))))
790 if (s->state != PA_STREAM_READY)
793 if (pa_streq(event, PA_STREAM_EVENT_FORMAT_LOST)) {
794 /* Let client know what the running time was when the stream had to be killed */
795 pa_usec_t stream_time;
796 if (pa_stream_get_time(s, &stream_time) == 0)
797 pa_proplist_setf(pl, "stream-time", "%llu", (unsigned long long) stream_time);
800 if (s->event_callback)
801 s->event_callback(s, event, pl, s->event_userdata);
807 pa_proplist_free(pl);
810 void pa_command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
812 pa_context *c = userdata;
813 uint32_t bytes, channel;
816 pa_assert(command == PA_COMMAND_REQUEST);
819 pa_assert(PA_REFCNT_VALUE(c) >= 1);
823 if (pa_tagstruct_getu32(t, &channel) < 0 ||
824 pa_tagstruct_getu32(t, &bytes) < 0 ||
825 !pa_tagstruct_eof(t)) {
826 pa_context_fail(c, PA_ERR_PROTOCOL);
830 if (!(s = pa_hashmap_get(c->playback_streams, PA_UINT32_TO_PTR(channel))))
833 if (s->state != PA_STREAM_READY)
836 s->requested_bytes += bytes;
838 /* pa_log("got request for %lli, now at %lli", (long long) bytes, (long long) s->requested_bytes); */
840 if (s->requested_bytes > 0 && s->write_callback)
841 s->write_callback(s, (size_t) s->requested_bytes, s->write_userdata);
847 int64_t pa_stream_get_underflow_index(pa_stream *p)
850 return p->latest_underrun_at_index;
853 void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
855 pa_context *c = userdata;
860 pa_assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
863 pa_assert(PA_REFCNT_VALUE(c) >= 1);
867 if (pa_tagstruct_getu32(t, &channel) < 0) {
868 pa_context_fail(c, PA_ERR_PROTOCOL);
872 if (c->version >= 23 && command == PA_COMMAND_UNDERFLOW) {
873 if (pa_tagstruct_gets64(t, &offset) < 0) {
874 pa_context_fail(c, PA_ERR_PROTOCOL);
879 if (!pa_tagstruct_eof(t)) {
880 pa_context_fail(c, PA_ERR_PROTOCOL);
884 if (!(s = pa_hashmap_get(c->playback_streams, PA_UINT32_TO_PTR(channel))))
887 if (s->state != PA_STREAM_READY)
891 s->latest_underrun_at_index = offset;
893 if (s->buffer_attr.prebuf > 0)
894 check_smoother_status(s, TRUE, FALSE, TRUE);
896 request_auto_timing_update(s, TRUE);
898 if (command == PA_COMMAND_OVERFLOW) {
899 if (s->overflow_callback)
900 s->overflow_callback(s, s->overflow_userdata);
901 } else if (command == PA_COMMAND_UNDERFLOW) {
902 if (s->underflow_callback)
903 s->underflow_callback(s, s->underflow_userdata);
910 static void invalidate_indexes(pa_stream *s, pa_bool_t r, pa_bool_t w) {
912 pa_assert(PA_REFCNT_VALUE(s) >= 1);
914 /* pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
916 if (s->state != PA_STREAM_READY)
920 s->write_index_not_before = s->context->ctag;
922 if (s->timing_info_valid)
923 s->timing_info.write_index_corrupt = TRUE;
925 /* pa_log("write_index invalidated"); */
929 s->read_index_not_before = s->context->ctag;
931 if (s->timing_info_valid)
932 s->timing_info.read_index_corrupt = TRUE;
934 /* pa_log("read_index invalidated"); */
937 request_auto_timing_update(s, TRUE);
940 static void auto_timing_update_callback(pa_mainloop_api *m, pa_time_event *e, const struct timeval *t, void *userdata) {
941 pa_stream *s = userdata;
944 pa_assert(PA_REFCNT_VALUE(s) >= 1);
947 request_auto_timing_update(s, FALSE);
951 static void create_stream_complete(pa_stream *s) {
953 pa_assert(PA_REFCNT_VALUE(s) >= 1);
954 pa_assert(s->state == PA_STREAM_CREATING);
956 pa_stream_set_state(s, PA_STREAM_READY);
958 if (s->requested_bytes > 0 && s->write_callback)
959 s->write_callback(s, (size_t) s->requested_bytes, s->write_userdata);
961 if (s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
962 s->auto_timing_interval_usec = AUTO_TIMING_INTERVAL_START_USEC;
963 pa_assert(!s->auto_timing_update_event);
964 s->auto_timing_update_event = pa_context_rttime_new(s->context, pa_rtclock_now() + s->auto_timing_interval_usec, &auto_timing_update_callback, s);
966 request_auto_timing_update(s, TRUE);
969 check_smoother_status(s, TRUE, FALSE, FALSE);
972 static void patch_buffer_attr(pa_stream *s, pa_buffer_attr *attr, pa_stream_flags_t *flags) {
978 if ((e = getenv("PULSE_LATENCY_MSEC"))) {
981 if (pa_atou(e, &ms) < 0 || ms <= 0)
982 pa_log_debug("Failed to parse $PULSE_LATENCY_MSEC: %s", e);
984 attr->maxlength = (uint32_t) -1;
985 attr->tlength = pa_usec_to_bytes(ms * PA_USEC_PER_MSEC, &s->sample_spec);
986 attr->minreq = (uint32_t) -1;
987 attr->prebuf = (uint32_t) -1;
988 attr->fragsize = attr->tlength;
992 *flags |= PA_STREAM_ADJUST_LATENCY;
995 if (s->context->version >= 13)
998 /* Version older than 0.9.10 didn't do server side buffer_attr
999 * selection, hence we have to fake it on the client side. */
1001 /* We choose fairly conservative values here, to not confuse
1002 * old clients with extremely large playback buffers */
1004 if (attr->maxlength == (uint32_t) -1)
1005 attr->maxlength = 4*1024*1024; /* 4MB is the maximum queue length PulseAudio <= 0.9.9 supported. */
1007 if (attr->tlength == (uint32_t) -1)
1008 attr->tlength = (uint32_t) pa_usec_to_bytes(250*PA_USEC_PER_MSEC, &s->sample_spec); /* 250ms of buffering */
1010 if (attr->minreq == (uint32_t) -1)
1011 attr->minreq = (attr->tlength)/5; /* Ask for more data when there are only 200ms left in the playback buffer */
1013 if (attr->prebuf == (uint32_t) -1)
1014 attr->prebuf = attr->tlength; /* Start to play only when the playback is fully filled up once */
1016 if (attr->fragsize == (uint32_t) -1)
1017 attr->fragsize = attr->tlength; /* Pass data to the app only when the buffer is filled up once */
1020 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1021 pa_stream *s = userdata;
1022 uint32_t requested_bytes = 0;
1026 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1027 pa_assert(s->state == PA_STREAM_CREATING);
1031 if (command != PA_COMMAND_REPLY) {
1032 if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
1035 pa_stream_set_state(s, PA_STREAM_FAILED);
1039 if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
1040 s->channel == PA_INVALID_INDEX ||
1041 ((s->direction != PA_STREAM_UPLOAD) && (pa_tagstruct_getu32(t, &s->stream_index) < 0 || s->stream_index == PA_INVALID_INDEX)) ||
1042 ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &requested_bytes) < 0)) {
1043 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1047 s->requested_bytes = (int64_t) requested_bytes;
1049 if (s->context->version >= 9) {
1050 if (s->direction == PA_STREAM_PLAYBACK) {
1051 if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
1052 pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
1053 pa_tagstruct_getu32(t, &s->buffer_attr.prebuf) < 0 ||
1054 pa_tagstruct_getu32(t, &s->buffer_attr.minreq) < 0) {
1055 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1058 } else if (s->direction == PA_STREAM_RECORD) {
1059 if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
1060 pa_tagstruct_getu32(t, &s->buffer_attr.fragsize) < 0) {
1061 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1067 if (s->context->version >= 12 && s->direction != PA_STREAM_UPLOAD) {
1070 const char *dn = NULL;
1071 pa_bool_t suspended;
1073 if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
1074 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
1075 pa_tagstruct_getu32(t, &s->device_index) < 0 ||
1076 pa_tagstruct_gets(t, &dn) < 0 ||
1077 pa_tagstruct_get_boolean(t, &suspended) < 0) {
1078 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1082 if (!dn || s->device_index == PA_INVALID_INDEX ||
1083 ss.channels != cm.channels ||
1084 !pa_channel_map_valid(&cm) ||
1085 !pa_sample_spec_valid(&ss) ||
1086 (s->n_formats == 0 && (
1087 (!(s->flags & PA_STREAM_FIX_FORMAT) && ss.format != s->sample_spec.format) ||
1088 (!(s->flags & PA_STREAM_FIX_RATE) && ss.rate != s->sample_spec.rate) ||
1089 (!(s->flags & PA_STREAM_FIX_CHANNELS) && !pa_channel_map_equal(&cm, &s->channel_map))))) {
1090 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1094 pa_xfree(s->device_name);
1095 s->device_name = pa_xstrdup(dn);
1096 s->suspended = suspended;
1098 s->channel_map = cm;
1099 s->sample_spec = ss;
1102 if (s->context->version >= 13 && s->direction != PA_STREAM_UPLOAD) {
1105 if (pa_tagstruct_get_usec(t, &usec) < 0) {
1106 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1110 if (s->direction == PA_STREAM_RECORD)
1111 s->timing_info.configured_source_usec = usec;
1113 s->timing_info.configured_sink_usec = usec;
1116 if ((s->context->version >= 21 && s->direction == PA_STREAM_PLAYBACK)
1117 || s->context->version >= 22) {
1119 pa_format_info *f = pa_format_info_new();
1120 pa_tagstruct_get_format_info(t, f);
1122 if (pa_format_info_valid(f))
1125 pa_format_info_free(f);
1126 if (s->n_formats > 0) {
1127 /* We used the extended API, so we should have got back a proper format */
1128 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1134 if (!pa_tagstruct_eof(t)) {
1135 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1139 if (s->direction == PA_STREAM_RECORD) {
1140 pa_assert(!s->record_memblockq);
1142 s->record_memblockq = pa_memblockq_new(
1144 s->buffer_attr.maxlength,
1146 pa_frame_size(&s->sample_spec),
1153 s->channel_valid = TRUE;
1154 pa_hashmap_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, PA_UINT32_TO_PTR(s->channel), s);
1156 create_stream_complete(s);
1162 static int create_stream(
1163 pa_stream_direction_t direction,
1166 const pa_buffer_attr *attr,
1167 pa_stream_flags_t flags,
1168 const pa_cvolume *volume,
1169 pa_stream *sync_stream) {
1173 pa_bool_t volume_set = !!volume;
1178 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1179 pa_assert(direction == PA_STREAM_PLAYBACK || direction == PA_STREAM_RECORD);
1181 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1182 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
1183 PA_CHECK_VALIDITY(s->context, s->direct_on_input == PA_INVALID_INDEX || direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1184 PA_CHECK_VALIDITY(s->context, !(flags & ~(PA_STREAM_START_CORKED|
1185 PA_STREAM_INTERPOLATE_TIMING|
1186 PA_STREAM_NOT_MONOTONIC|
1187 PA_STREAM_AUTO_TIMING_UPDATE|
1188 PA_STREAM_NO_REMAP_CHANNELS|
1189 PA_STREAM_NO_REMIX_CHANNELS|
1190 PA_STREAM_FIX_FORMAT|
1192 PA_STREAM_FIX_CHANNELS|
1193 PA_STREAM_DONT_MOVE|
1194 PA_STREAM_VARIABLE_RATE|
1195 PA_STREAM_PEAK_DETECT|
1196 PA_STREAM_START_MUTED|
1197 PA_STREAM_ADJUST_LATENCY|
1198 PA_STREAM_EARLY_REQUESTS|
1199 PA_STREAM_DONT_INHIBIT_AUTO_SUSPEND|
1200 PA_STREAM_START_UNMUTED|
1201 PA_STREAM_FAIL_ON_SUSPEND|
1202 PA_STREAM_RELATIVE_VOLUME|
1203 PA_STREAM_PASSTHROUGH)), PA_ERR_INVALID);
1206 PA_CHECK_VALIDITY(s->context, s->context->version >= 12 || !(flags & PA_STREAM_VARIABLE_RATE), PA_ERR_NOTSUPPORTED);
1207 PA_CHECK_VALIDITY(s->context, s->context->version >= 13 || !(flags & PA_STREAM_PEAK_DETECT), PA_ERR_NOTSUPPORTED);
1208 PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
1209 /* Although some of the other flags are not supported on older
1210 * version, we don't check for them here, because it doesn't hurt
1211 * when they are passed but actually not supported. This makes
1212 * client development easier */
1214 PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_PLAYBACK || !(flags & (PA_STREAM_START_MUTED)), PA_ERR_INVALID);
1215 PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_RECORD || !(flags & (PA_STREAM_PEAK_DETECT)), PA_ERR_INVALID);
1216 PA_CHECK_VALIDITY(s->context, !volume || (pa_sample_spec_valid(&s->sample_spec) && volume->channels == s->sample_spec.channels), PA_ERR_INVALID);
1217 PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
1218 PA_CHECK_VALIDITY(s->context, (flags & (PA_STREAM_ADJUST_LATENCY|PA_STREAM_EARLY_REQUESTS)) != (PA_STREAM_ADJUST_LATENCY|PA_STREAM_EARLY_REQUESTS), PA_ERR_INVALID);
1222 s->direction = direction;
1225 s->syncid = sync_stream->syncid;
1228 s->buffer_attr = *attr;
1229 patch_buffer_attr(s, &s->buffer_attr, &flags);
1232 s->corked = !!(flags & PA_STREAM_START_CORKED);
1234 if (flags & PA_STREAM_INTERPOLATE_TIMING) {
1237 x = pa_rtclock_now();
1239 pa_assert(!s->smoother);
1240 s->smoother = pa_smoother_new(
1241 SMOOTHER_ADJUST_TIME,
1242 SMOOTHER_HISTORY_TIME,
1243 !(flags & PA_STREAM_NOT_MONOTONIC),
1245 SMOOTHER_MIN_HISTORY,
1251 dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
1253 t = pa_tagstruct_command(
1255 (uint32_t) (s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM),
1258 if (s->context->version < 13)
1259 pa_tagstruct_puts(t, pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME));
1263 PA_TAG_SAMPLE_SPEC, &s->sample_spec,
1264 PA_TAG_CHANNEL_MAP, &s->channel_map,
1265 PA_TAG_U32, PA_INVALID_INDEX,
1267 PA_TAG_U32, s->buffer_attr.maxlength,
1268 PA_TAG_BOOLEAN, s->corked,
1272 if (pa_sample_spec_valid(&s->sample_spec))
1273 volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
1275 /* This is not really relevant, since no volume was set, and
1276 * the real number of channels is embedded in the format_info
1278 volume = pa_cvolume_reset(&cv, PA_CHANNELS_MAX);
1282 if (s->direction == PA_STREAM_PLAYBACK) {
1285 PA_TAG_U32, s->buffer_attr.tlength,
1286 PA_TAG_U32, s->buffer_attr.prebuf,
1287 PA_TAG_U32, s->buffer_attr.minreq,
1288 PA_TAG_U32, s->syncid,
1291 pa_tagstruct_put_cvolume(t, volume);
1293 pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
1295 if (s->context->version >= 12) {
1298 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMAP_CHANNELS,
1299 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMIX_CHANNELS,
1300 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_FORMAT,
1301 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_RATE,
1302 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_CHANNELS,
1303 PA_TAG_BOOLEAN, flags & PA_STREAM_DONT_MOVE,
1304 PA_TAG_BOOLEAN, flags & PA_STREAM_VARIABLE_RATE,
1308 if (s->context->version >= 13) {
1310 if (s->direction == PA_STREAM_PLAYBACK)
1311 pa_tagstruct_put_boolean(t, flags & PA_STREAM_START_MUTED);
1313 pa_tagstruct_put_boolean(t, flags & PA_STREAM_PEAK_DETECT);
1317 PA_TAG_BOOLEAN, flags & PA_STREAM_ADJUST_LATENCY,
1318 PA_TAG_PROPLIST, s->proplist,
1321 if (s->direction == PA_STREAM_RECORD)
1322 pa_tagstruct_putu32(t, s->direct_on_input);
1325 if (s->context->version >= 14) {
1327 if (s->direction == PA_STREAM_PLAYBACK)
1328 pa_tagstruct_put_boolean(t, volume_set);
1330 pa_tagstruct_put_boolean(t, flags & PA_STREAM_EARLY_REQUESTS);
1333 if (s->context->version >= 15) {
1335 if (s->direction == PA_STREAM_PLAYBACK)
1336 pa_tagstruct_put_boolean(t, flags & (PA_STREAM_START_MUTED|PA_STREAM_START_UNMUTED));
1338 pa_tagstruct_put_boolean(t, flags & PA_STREAM_DONT_INHIBIT_AUTO_SUSPEND);
1339 pa_tagstruct_put_boolean(t, flags & PA_STREAM_FAIL_ON_SUSPEND);
1342 if (s->context->version >= 17 && s->direction == PA_STREAM_PLAYBACK)
1343 pa_tagstruct_put_boolean(t, flags & PA_STREAM_RELATIVE_VOLUME);
1345 if (s->context->version >= 18 && s->direction == PA_STREAM_PLAYBACK)
1346 pa_tagstruct_put_boolean(t, flags & (PA_STREAM_PASSTHROUGH));
1348 if ((s->context->version >= 21 && s->direction == PA_STREAM_PLAYBACK)
1349 || s->context->version >= 22) {
1351 pa_tagstruct_putu8(t, s->n_formats);
1352 for (i = 0; i < s->n_formats; i++)
1353 pa_tagstruct_put_format_info(t, s->req_formats[i]);
1356 if (s->context->version >= 22 && s->direction == PA_STREAM_RECORD) {
1357 pa_tagstruct_put_cvolume(t, volume);
1358 pa_tagstruct_put_boolean(t, flags & PA_STREAM_START_MUTED);
1359 pa_tagstruct_put_boolean(t, volume_set);
1360 pa_tagstruct_put_boolean(t, flags & (PA_STREAM_START_MUTED|PA_STREAM_START_UNMUTED));
1361 pa_tagstruct_put_boolean(t, flags & PA_STREAM_RELATIVE_VOLUME);
1362 pa_tagstruct_put_boolean(t, flags & (PA_STREAM_PASSTHROUGH));
1365 pa_pstream_send_tagstruct(s->context->pstream, t);
1366 pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
1368 pa_stream_set_state(s, PA_STREAM_CREATING);
1374 int pa_stream_connect_playback(
1377 const pa_buffer_attr *attr,
1378 pa_stream_flags_t flags,
1379 const pa_cvolume *volume,
1380 pa_stream *sync_stream) {
1383 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1385 return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
1388 int pa_stream_connect_record(
1391 const pa_buffer_attr *attr,
1392 pa_stream_flags_t flags) {
1395 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1397 return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
1400 int pa_stream_begin_write(
1406 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1408 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1409 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1410 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1411 PA_CHECK_VALIDITY(s->context, data, PA_ERR_INVALID);
1412 PA_CHECK_VALIDITY(s->context, nbytes && *nbytes != 0, PA_ERR_INVALID);
1414 if (*nbytes != (size_t) -1) {
1417 m = pa_mempool_block_size_max(s->context->mempool);
1418 fs = pa_frame_size(&s->sample_spec);
1425 if (!s->write_memblock) {
1426 s->write_memblock = pa_memblock_new(s->context->mempool, *nbytes);
1427 s->write_data = pa_memblock_acquire(s->write_memblock);
1430 *data = s->write_data;
1431 *nbytes = pa_memblock_get_length(s->write_memblock);
1436 int pa_stream_cancel_write(
1440 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1442 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1443 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1444 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1445 PA_CHECK_VALIDITY(s->context, s->write_memblock, PA_ERR_BADSTATE);
1447 pa_assert(s->write_data);
1449 pa_memblock_release(s->write_memblock);
1450 pa_memblock_unref(s->write_memblock);
1451 s->write_memblock = NULL;
1452 s->write_data = NULL;
1457 int pa_stream_write(
1461 pa_free_cb_t free_cb,
1463 pa_seek_mode_t seek) {
1466 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1469 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1470 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1471 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1472 PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
1473 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
1474 PA_CHECK_VALIDITY(s->context,
1475 !s->write_memblock ||
1476 ((data >= s->write_data) &&
1477 ((const char*) data + length <= (const char*) s->write_data + pa_memblock_get_length(s->write_memblock))),
1479 PA_CHECK_VALIDITY(s->context, !free_cb || !s->write_memblock, PA_ERR_INVALID);
1481 if (s->write_memblock) {
1484 /* pa_stream_write_begin() was called before */
1486 pa_memblock_release(s->write_memblock);
1488 chunk.memblock = s->write_memblock;
1489 chunk.index = (const char *) data - (const char *) s->write_data;
1490 chunk.length = length;
1492 s->write_memblock = NULL;
1493 s->write_data = NULL;
1495 pa_pstream_send_memblock(s->context->pstream, s->channel, offset, seek, &chunk);
1496 pa_memblock_unref(chunk.memblock);
1499 pa_seek_mode_t t_seek = seek;
1500 int64_t t_offset = offset;
1501 size_t t_length = length;
1502 const void *t_data = data;
1504 /* pa_stream_write_begin() was not called before */
1506 while (t_length > 0) {
1511 if (free_cb && !pa_pstream_get_shm(s->context->pstream)) {
1512 chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) t_data, t_length, free_cb, 1);
1513 chunk.length = t_length;
1517 chunk.length = PA_MIN(t_length, pa_mempool_block_size_max(s->context->mempool));
1518 chunk.memblock = pa_memblock_new(s->context->mempool, chunk.length);
1520 d = pa_memblock_acquire(chunk.memblock);
1521 memcpy(d, t_data, chunk.length);
1522 pa_memblock_release(chunk.memblock);
1525 pa_pstream_send_memblock(s->context->pstream, s->channel, t_offset, t_seek, &chunk);
1528 t_seek = PA_SEEK_RELATIVE;
1530 t_data = (const uint8_t*) t_data + chunk.length;
1531 t_length -= chunk.length;
1533 pa_memblock_unref(chunk.memblock);
1536 if (free_cb && pa_pstream_get_shm(s->context->pstream))
1537 free_cb((void*) data);
1540 /* This is obviously wrong since we ignore the seeking index . But
1541 * that's OK, the server side applies the same error */
1542 s->requested_bytes -= (seek == PA_SEEK_RELATIVE ? offset : 0) + (int64_t) length;
1544 /* pa_log("wrote %lli, now at %lli", (long long) length, (long long) s->requested_bytes); */
1546 if (s->direction == PA_STREAM_PLAYBACK) {
1548 /* Update latency request correction */
1549 if (s->write_index_corrections[s->current_write_index_correction].valid) {
1551 if (seek == PA_SEEK_ABSOLUTE) {
1552 s->write_index_corrections[s->current_write_index_correction].corrupt = FALSE;
1553 s->write_index_corrections[s->current_write_index_correction].absolute = TRUE;
1554 s->write_index_corrections[s->current_write_index_correction].value = offset + (int64_t) length;
1555 } else if (seek == PA_SEEK_RELATIVE) {
1556 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
1557 s->write_index_corrections[s->current_write_index_correction].value += offset + (int64_t) length;
1559 s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
1562 /* Update the write index in the already available latency data */
1563 if (s->timing_info_valid) {
1565 if (seek == PA_SEEK_ABSOLUTE) {
1566 s->timing_info.write_index_corrupt = FALSE;
1567 s->timing_info.write_index = offset + (int64_t) length;
1568 } else if (seek == PA_SEEK_RELATIVE) {
1569 if (!s->timing_info.write_index_corrupt)
1570 s->timing_info.write_index += offset + (int64_t) length;
1572 s->timing_info.write_index_corrupt = TRUE;
1575 if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
1576 request_auto_timing_update(s, TRUE);
1582 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
1584 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1588 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1589 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1590 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1592 if (!s->peek_memchunk.memblock) {
1594 if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
1600 s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
1603 pa_assert(s->peek_data);
1604 *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
1605 *length = s->peek_memchunk.length;
1609 int pa_stream_drop(pa_stream *s) {
1611 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1613 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1614 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1615 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1616 PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
1618 pa_memblockq_drop(s->record_memblockq, s->peek_memchunk.length);
1620 /* Fix the simulated local read index */
1621 if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
1622 s->timing_info.read_index += (int64_t) s->peek_memchunk.length;
1624 pa_assert(s->peek_data);
1625 pa_memblock_release(s->peek_memchunk.memblock);
1626 pa_memblock_unref(s->peek_memchunk.memblock);
1627 pa_memchunk_reset(&s->peek_memchunk);
1632 size_t pa_stream_writable_size(pa_stream *s) {
1634 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1636 PA_CHECK_VALIDITY_RETURN_ANY(s->context, !pa_detect_fork(), PA_ERR_FORKED, (size_t) -1);
1637 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1638 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1640 return s->requested_bytes > 0 ? (size_t) s->requested_bytes : 0;
1643 size_t pa_stream_readable_size(pa_stream *s) {
1645 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1647 PA_CHECK_VALIDITY_RETURN_ANY(s->context, !pa_detect_fork(), PA_ERR_FORKED, (size_t) -1);
1648 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1649 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1651 return pa_memblockq_get_length(s->record_memblockq);
1654 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1660 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1662 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1663 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1664 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1666 /* Ask for a timing update before we cork/uncork to get the best
1667 * accuracy for the transport latency suitable for the
1668 * check_smoother_status() call in the started callback */
1669 request_auto_timing_update(s, TRUE);
1671 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1673 t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
1674 pa_tagstruct_putu32(t, s->channel);
1675 pa_pstream_send_tagstruct(s->context->pstream, t);
1676 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);
1678 /* This might cause the read index to continue again, hence
1679 * let's request a timing update */
1680 request_auto_timing_update(s, TRUE);
1685 static pa_usec_t calc_time(pa_stream *s, pa_bool_t ignore_transport) {
1689 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1690 pa_assert(s->state == PA_STREAM_READY);
1691 pa_assert(s->direction != PA_STREAM_UPLOAD);
1692 pa_assert(s->timing_info_valid);
1693 pa_assert(s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt);
1694 pa_assert(s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt);
1696 if (s->direction == PA_STREAM_PLAYBACK) {
1697 /* The last byte that was written into the output device
1698 * had this time value associated */
1699 usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1701 if (!s->corked && !s->suspended) {
1703 if (!ignore_transport)
1704 /* Because the latency info took a little time to come
1705 * to us, we assume that the real output time is actually
1707 usec += s->timing_info.transport_usec;
1709 /* However, the output device usually maintains a buffer
1710 too, hence the real sample currently played is a little
1712 if (s->timing_info.sink_usec >= usec)
1715 usec -= s->timing_info.sink_usec;
1719 pa_assert(s->direction == PA_STREAM_RECORD);
1721 /* The last byte written into the server side queue had
1722 * this time value associated */
1723 usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1725 if (!s->corked && !s->suspended) {
1727 if (!ignore_transport)
1728 /* Add transport latency */
1729 usec += s->timing_info.transport_usec;
1731 /* Add latency of data in device buffer */
1732 usec += s->timing_info.source_usec;
1734 /* If this is a monitor source, we need to correct the
1735 * time by the playback device buffer */
1736 if (s->timing_info.sink_usec >= usec)
1739 usec -= s->timing_info.sink_usec;
1746 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1747 pa_operation *o = userdata;
1748 struct timeval local, remote, now;
1750 pa_bool_t playing = FALSE;
1751 uint64_t underrun_for = 0, playing_for = 0;
1755 pa_assert(PA_REFCNT_VALUE(o) >= 1);
1757 if (!o->context || !o->stream)
1760 i = &o->stream->timing_info;
1762 o->stream->timing_info_valid = FALSE;
1763 i->write_index_corrupt = TRUE;
1764 i->read_index_corrupt = TRUE;
1766 if (command != PA_COMMAND_REPLY) {
1767 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1772 if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
1773 pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
1774 pa_tagstruct_get_boolean(t, &playing) < 0 ||
1775 pa_tagstruct_get_timeval(t, &local) < 0 ||
1776 pa_tagstruct_get_timeval(t, &remote) < 0 ||
1777 pa_tagstruct_gets64(t, &i->write_index) < 0 ||
1778 pa_tagstruct_gets64(t, &i->read_index) < 0) {
1780 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1784 if (o->context->version >= 13 &&
1785 o->stream->direction == PA_STREAM_PLAYBACK)
1786 if (pa_tagstruct_getu64(t, &underrun_for) < 0 ||
1787 pa_tagstruct_getu64(t, &playing_for) < 0) {
1789 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1794 if (!pa_tagstruct_eof(t)) {
1795 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1798 o->stream->timing_info_valid = TRUE;
1799 i->write_index_corrupt = FALSE;
1800 i->read_index_corrupt = FALSE;
1802 i->playing = (int) playing;
1803 i->since_underrun = (int64_t) (playing ? playing_for : underrun_for);
1805 pa_gettimeofday(&now);
1807 /* Calculate timestamps */
1808 if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
1809 /* local and remote seem to have synchronized clocks */
1811 if (o->stream->direction == PA_STREAM_PLAYBACK)
1812 i->transport_usec = pa_timeval_diff(&remote, &local);
1814 i->transport_usec = pa_timeval_diff(&now, &remote);
1816 i->synchronized_clocks = TRUE;
1817 i->timestamp = remote;
1819 /* clocks are not synchronized, let's estimate latency then */
1820 i->transport_usec = pa_timeval_diff(&now, &local)/2;
1821 i->synchronized_clocks = FALSE;
1822 i->timestamp = local;
1823 pa_timeval_add(&i->timestamp, i->transport_usec);
1826 /* Invalidate read and write indexes if necessary */
1827 if (tag < o->stream->read_index_not_before)
1828 i->read_index_corrupt = TRUE;
1830 if (tag < o->stream->write_index_not_before)
1831 i->write_index_corrupt = TRUE;
1833 if (o->stream->direction == PA_STREAM_PLAYBACK) {
1834 /* Write index correction */
1837 uint32_t ctag = tag;
1839 /* Go through the saved correction values and add up the
1840 * total correction.*/
1841 for (n = 0, j = o->stream->current_write_index_correction+1;
1842 n < PA_MAX_WRITE_INDEX_CORRECTIONS;
1843 n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
1845 /* Step over invalid data or out-of-date data */
1846 if (!o->stream->write_index_corrections[j].valid ||
1847 o->stream->write_index_corrections[j].tag < ctag)
1850 /* Make sure that everything is in order */
1851 ctag = o->stream->write_index_corrections[j].tag+1;
1853 /* Now fix the write index */
1854 if (o->stream->write_index_corrections[j].corrupt) {
1855 /* A corrupting seek was made */
1856 i->write_index_corrupt = TRUE;
1857 } else if (o->stream->write_index_corrections[j].absolute) {
1858 /* An absolute seek was made */
1859 i->write_index = o->stream->write_index_corrections[j].value;
1860 i->write_index_corrupt = FALSE;
1861 } else if (!i->write_index_corrupt) {
1862 /* A relative seek was made */
1863 i->write_index += o->stream->write_index_corrections[j].value;
1867 /* Clear old correction entries */
1868 for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
1869 if (!o->stream->write_index_corrections[n].valid)
1872 if (o->stream->write_index_corrections[n].tag <= tag)
1873 o->stream->write_index_corrections[n].valid = FALSE;
1877 if (o->stream->direction == PA_STREAM_RECORD) {
1878 /* Read index correction */
1880 if (!i->read_index_corrupt)
1881 i->read_index -= (int64_t) pa_memblockq_get_length(o->stream->record_memblockq);
1884 /* Update smoother if we're not corked */
1885 if (o->stream->smoother && !o->stream->corked) {
1888 u = x = pa_rtclock_now() - i->transport_usec;
1890 if (o->stream->direction == PA_STREAM_PLAYBACK && o->context->version >= 13) {
1893 /* If we weren't playing then it will take some time
1894 * until the audio will actually come out through the
1895 * speakers. Since we follow that timing here, we need
1896 * to try to fix this up */
1898 su = pa_bytes_to_usec((uint64_t) i->since_underrun, &o->stream->sample_spec);
1900 if (su < i->sink_usec)
1901 x += i->sink_usec - su;
1905 pa_smoother_pause(o->stream->smoother, x);
1907 /* Update the smoother */
1908 if ((o->stream->direction == PA_STREAM_PLAYBACK && !i->read_index_corrupt) ||
1909 (o->stream->direction == PA_STREAM_RECORD && !i->write_index_corrupt))
1910 pa_smoother_put(o->stream->smoother, u, calc_time(o->stream, TRUE));
1913 pa_smoother_resume(o->stream->smoother, x, TRUE);
1917 o->stream->auto_timing_update_requested = FALSE;
1919 if (o->stream->latency_update_callback)
1920 o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
1922 if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
1923 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1924 cb(o->stream, o->stream->timing_info_valid, o->userdata);
1929 pa_operation_done(o);
1930 pa_operation_unref(o);
1933 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1941 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1943 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
1944 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1945 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1947 if (s->direction == PA_STREAM_PLAYBACK) {
1948 /* Find a place to store the write_index correction data for this entry */
1949 cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
1951 /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
1952 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
1954 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1956 t = pa_tagstruct_command(
1958 (uint32_t) (s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY),
1960 pa_tagstruct_putu32(t, s->channel);
1961 pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
1963 pa_pstream_send_tagstruct(s->context->pstream, t);
1964 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);
1966 if (s->direction == PA_STREAM_PLAYBACK) {
1967 /* Fill in initial correction data */
1969 s->current_write_index_correction = cidx;
1971 s->write_index_corrections[cidx].valid = TRUE;
1972 s->write_index_corrections[cidx].absolute = FALSE;
1973 s->write_index_corrections[cidx].corrupt = FALSE;
1974 s->write_index_corrections[cidx].tag = tag;
1975 s->write_index_corrections[cidx].value = 0;
1981 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1982 pa_stream *s = userdata;
1986 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1990 if (command != PA_COMMAND_REPLY) {
1991 if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
1994 pa_stream_set_state(s, PA_STREAM_FAILED);
1996 } else if (!pa_tagstruct_eof(t)) {
1997 pa_context_fail(s->context, PA_ERR_PROTOCOL);
2001 pa_stream_set_state(s, PA_STREAM_TERMINATED);
2007 int pa_stream_disconnect(pa_stream *s) {
2012 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2014 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2015 PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
2016 PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
2020 t = pa_tagstruct_command(
2022 (uint32_t) (s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
2023 (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM)),
2025 pa_tagstruct_putu32(t, s->channel);
2026 pa_pstream_send_tagstruct(s->context->pstream, t);
2027 pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
2033 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
2035 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2037 if (pa_detect_fork())
2040 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2043 s->read_callback = cb;
2044 s->read_userdata = userdata;
2047 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
2049 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2051 if (pa_detect_fork())
2054 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2057 s->write_callback = cb;
2058 s->write_userdata = userdata;
2061 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2063 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2065 if (pa_detect_fork())
2068 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2071 s->state_callback = cb;
2072 s->state_userdata = userdata;
2075 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2077 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2079 if (pa_detect_fork())
2082 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2085 s->overflow_callback = cb;
2086 s->overflow_userdata = userdata;
2089 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2091 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2093 if (pa_detect_fork())
2096 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2099 s->underflow_callback = cb;
2100 s->underflow_userdata = userdata;
2103 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2105 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2107 if (pa_detect_fork())
2110 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2113 s->latency_update_callback = cb;
2114 s->latency_update_userdata = userdata;
2117 void pa_stream_set_moved_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2119 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2121 if (pa_detect_fork())
2124 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2127 s->moved_callback = cb;
2128 s->moved_userdata = userdata;
2131 void pa_stream_set_suspended_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2133 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2135 if (pa_detect_fork())
2138 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2141 s->suspended_callback = cb;
2142 s->suspended_userdata = userdata;
2145 void pa_stream_set_started_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2147 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2149 if (pa_detect_fork())
2152 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2155 s->started_callback = cb;
2156 s->started_userdata = userdata;
2159 void pa_stream_set_event_callback(pa_stream *s, pa_stream_event_cb_t cb, void *userdata) {
2161 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2163 if (pa_detect_fork())
2166 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2169 s->event_callback = cb;
2170 s->event_userdata = userdata;
2173 void pa_stream_set_buffer_attr_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
2175 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2177 if (pa_detect_fork())
2180 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
2183 s->buffer_attr_callback = cb;
2184 s->buffer_attr_userdata = userdata;
2187 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2188 pa_operation *o = userdata;
2193 pa_assert(PA_REFCNT_VALUE(o) >= 1);
2198 if (command != PA_COMMAND_REPLY) {
2199 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
2203 } else if (!pa_tagstruct_eof(t)) {
2204 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2209 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2210 cb(o->stream, success, o->userdata);
2214 pa_operation_done(o);
2215 pa_operation_unref(o);
2218 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
2224 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2226 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2227 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2228 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2230 /* Ask for a timing update before we cork/uncork to get the best
2231 * accuracy for the transport latency suitable for the
2232 * check_smoother_status() call in the started callback */
2233 request_auto_timing_update(s, TRUE);
2237 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2239 t = pa_tagstruct_command(
2241 (uint32_t) (s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM),
2243 pa_tagstruct_putu32(t, s->channel);
2244 pa_tagstruct_put_boolean(t, !!b);
2245 pa_pstream_send_tagstruct(s->context->pstream, t);
2246 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);
2248 check_smoother_status(s, FALSE, FALSE, FALSE);
2250 /* This might cause the indexes to hang/start again, hence let's
2251 * request a timing update, after the cork/uncork, too */
2252 request_auto_timing_update(s, TRUE);
2257 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
2263 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2265 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2266 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2268 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2270 t = pa_tagstruct_command(s->context, command, &tag);
2271 pa_tagstruct_putu32(t, s->channel);
2272 pa_pstream_send_tagstruct(s->context->pstream, t);
2273 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);
2278 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
2282 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2284 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2285 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2286 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2288 /* Ask for a timing update *before* the flush, so that the
2289 * transport usec is as up to date as possible when we get the
2290 * underflow message and update the smoother status*/
2291 request_auto_timing_update(s, TRUE);
2293 if (!(o = stream_send_simple_command(s, (uint32_t) (s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM), cb, userdata)))
2296 if (s->direction == PA_STREAM_PLAYBACK) {
2298 if (s->write_index_corrections[s->current_write_index_correction].valid)
2299 s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
2301 if (s->buffer_attr.prebuf > 0)
2302 check_smoother_status(s, FALSE, FALSE, TRUE);
2304 /* This will change the write index, but leave the
2305 * read index untouched. */
2306 invalidate_indexes(s, FALSE, TRUE);
2309 /* For record streams this has no influence on the write
2310 * index, but the read index might jump. */
2311 invalidate_indexes(s, TRUE, FALSE);
2313 /* Note that we do not update requested_bytes here. This is
2314 * because we cannot really know how data actually was dropped
2315 * from the write index due to this. This 'error' will be applied
2316 * by both client and server and hence we should be fine. */
2321 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
2325 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2327 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2328 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2329 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
2330 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
2332 /* Ask for a timing update before we cork/uncork to get the best
2333 * accuracy for the transport latency suitable for the
2334 * check_smoother_status() call in the started callback */
2335 request_auto_timing_update(s, TRUE);
2337 if (!(o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
2340 /* This might cause the read index to hang again, hence
2341 * let's request a timing update */
2342 request_auto_timing_update(s, TRUE);
2347 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
2351 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2353 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2354 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2355 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
2356 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
2358 /* Ask for a timing update before we cork/uncork to get the best
2359 * accuracy for the transport latency suitable for the
2360 * check_smoother_status() call in the started callback */
2361 request_auto_timing_update(s, TRUE);
2363 if (!(o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
2366 /* This might cause the read index to start moving again, hence
2367 * let's request a timing update */
2368 request_auto_timing_update(s, TRUE);
2373 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
2377 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2380 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2381 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2382 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2384 if (s->context->version >= 13) {
2385 pa_proplist *p = pa_proplist_new();
2387 pa_proplist_sets(p, PA_PROP_MEDIA_NAME, name);
2388 o = pa_stream_proplist_update(s, PA_UPDATE_REPLACE, p, cb, userdata);
2389 pa_proplist_free(p);
2394 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2395 t = pa_tagstruct_command(
2397 (uint32_t) (s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME),
2399 pa_tagstruct_putu32(t, s->channel);
2400 pa_tagstruct_puts(t, name);
2401 pa_pstream_send_tagstruct(s->context->pstream, t);
2402 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);
2408 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
2412 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2414 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2415 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2416 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2417 PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
2418 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
2419 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
2422 usec = pa_smoother_get(s->smoother, pa_rtclock_now());
2424 usec = calc_time(s, FALSE);
2426 /* Make sure the time runs monotonically */
2427 if (!(s->flags & PA_STREAM_NOT_MONOTONIC)) {
2428 if (usec < s->previous_time)
2429 usec = s->previous_time;
2431 s->previous_time = usec;
2440 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
2442 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2450 if (negative && s->direction == PA_STREAM_RECORD) {
2458 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
2464 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2467 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2468 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2469 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2470 PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
2471 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
2472 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
2474 if ((r = pa_stream_get_time(s, &t)) < 0)
2477 if (s->direction == PA_STREAM_PLAYBACK)
2478 cindex = s->timing_info.write_index;
2480 cindex = s->timing_info.read_index;
2485 c = pa_bytes_to_usec((uint64_t) cindex, &s->sample_spec);
2487 if (s->direction == PA_STREAM_PLAYBACK)
2488 *r_usec = time_counter_diff(s, c, t, negative);
2490 *r_usec = time_counter_diff(s, t, c, negative);
2495 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
2497 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2499 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2500 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2501 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2502 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_NODATA);
2504 return &s->timing_info;
2507 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
2509 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2511 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2513 return &s->sample_spec;
2516 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
2518 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2520 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2522 return &s->channel_map;
2525 const pa_format_info* pa_stream_get_format_info(pa_stream *s) {
2527 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2529 /* We don't have the format till routing is done */
2530 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2531 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2535 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
2537 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2539 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2540 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2541 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 9, PA_ERR_NOTSUPPORTED);
2543 return &s->buffer_attr;
2546 static void stream_set_buffer_attr_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2547 pa_operation *o = userdata;
2552 pa_assert(PA_REFCNT_VALUE(o) >= 1);
2557 if (command != PA_COMMAND_REPLY) {
2558 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
2563 if (o->stream->direction == PA_STREAM_PLAYBACK) {
2564 if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
2565 pa_tagstruct_getu32(t, &o->stream->buffer_attr.tlength) < 0 ||
2566 pa_tagstruct_getu32(t, &o->stream->buffer_attr.prebuf) < 0 ||
2567 pa_tagstruct_getu32(t, &o->stream->buffer_attr.minreq) < 0) {
2568 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2571 } else if (o->stream->direction == PA_STREAM_RECORD) {
2572 if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
2573 pa_tagstruct_getu32(t, &o->stream->buffer_attr.fragsize) < 0) {
2574 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2579 if (o->stream->context->version >= 13) {
2582 if (pa_tagstruct_get_usec(t, &usec) < 0) {
2583 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2587 if (o->stream->direction == PA_STREAM_RECORD)
2588 o->stream->timing_info.configured_source_usec = usec;
2590 o->stream->timing_info.configured_sink_usec = usec;
2593 if (!pa_tagstruct_eof(t)) {
2594 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2600 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2601 cb(o->stream, success, o->userdata);
2605 pa_operation_done(o);
2606 pa_operation_unref(o);
2610 pa_operation* pa_stream_set_buffer_attr(pa_stream *s, const pa_buffer_attr *attr, pa_stream_success_cb_t cb, void *userdata) {
2614 pa_buffer_attr copy;
2617 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2620 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2621 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2622 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2623 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2625 /* Ask for a timing update before we cork/uncork to get the best
2626 * accuracy for the transport latency suitable for the
2627 * check_smoother_status() call in the started callback */
2628 request_auto_timing_update(s, TRUE);
2630 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2632 t = pa_tagstruct_command(
2634 (uint32_t) (s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR : PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR),
2636 pa_tagstruct_putu32(t, s->channel);
2639 patch_buffer_attr(s, ©, NULL);
2642 pa_tagstruct_putu32(t, attr->maxlength);
2644 if (s->direction == PA_STREAM_PLAYBACK)
2647 PA_TAG_U32, attr->tlength,
2648 PA_TAG_U32, attr->prebuf,
2649 PA_TAG_U32, attr->minreq,
2652 pa_tagstruct_putu32(t, attr->fragsize);
2654 if (s->context->version >= 13)
2655 pa_tagstruct_put_boolean(t, !!(s->flags & PA_STREAM_ADJUST_LATENCY));
2657 if (s->context->version >= 14)
2658 pa_tagstruct_put_boolean(t, !!(s->flags & PA_STREAM_EARLY_REQUESTS));
2660 pa_pstream_send_tagstruct(s->context->pstream, t);
2661 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);
2663 /* This might cause changes in the read/write index, hence let's
2664 * request a timing update */
2665 request_auto_timing_update(s, TRUE);
2670 uint32_t pa_stream_get_device_index(pa_stream *s) {
2672 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2674 PA_CHECK_VALIDITY_RETURN_ANY(s->context, !pa_detect_fork(), PA_ERR_FORKED, PA_INVALID_INDEX);
2675 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2676 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2677 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2678 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->device_index != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2680 return s->device_index;
2683 const char *pa_stream_get_device_name(pa_stream *s) {
2685 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2687 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2688 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2689 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2690 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2691 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->device_name, PA_ERR_BADSTATE);
2693 return s->device_name;
2696 int pa_stream_is_suspended(pa_stream *s) {
2698 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2700 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2701 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2702 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2703 PA_CHECK_VALIDITY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2705 return s->suspended;
2708 int pa_stream_is_corked(pa_stream *s) {
2710 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2712 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2713 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2714 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2719 static void stream_update_sample_rate_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
2720 pa_operation *o = userdata;
2725 pa_assert(PA_REFCNT_VALUE(o) >= 1);
2730 if (command != PA_COMMAND_REPLY) {
2731 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
2737 if (!pa_tagstruct_eof(t)) {
2738 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2743 o->stream->sample_spec.rate = PA_PTR_TO_UINT(o->private);
2744 pa_assert(pa_sample_spec_valid(&o->stream->sample_spec));
2747 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2748 cb(o->stream, success, o->userdata);
2752 pa_operation_done(o);
2753 pa_operation_unref(o);
2757 pa_operation *pa_stream_update_sample_rate(pa_stream *s, uint32_t rate, pa_stream_success_cb_t cb, void *userdata) {
2763 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2765 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2766 PA_CHECK_VALIDITY_RETURN_NULL(s->context, rate > 0 && rate <= PA_RATE_MAX, PA_ERR_INVALID);
2767 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2768 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2769 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->flags & PA_STREAM_VARIABLE_RATE, PA_ERR_BADSTATE);
2770 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2772 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2773 o->private = PA_UINT_TO_PTR(rate);
2775 t = pa_tagstruct_command(
2777 (uint32_t) (s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE : PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE),
2779 pa_tagstruct_putu32(t, s->channel);
2780 pa_tagstruct_putu32(t, rate);
2782 pa_pstream_send_tagstruct(s->context->pstream, t);
2783 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);
2788 pa_operation *pa_stream_proplist_update(pa_stream *s, pa_update_mode_t mode, pa_proplist *p, pa_stream_success_cb_t cb, void *userdata) {
2794 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2796 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2797 PA_CHECK_VALIDITY_RETURN_NULL(s->context, mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE, PA_ERR_INVALID);
2798 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2799 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2800 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2802 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2804 t = pa_tagstruct_command(
2806 (uint32_t) (s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST : PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST),
2808 pa_tagstruct_putu32(t, s->channel);
2809 pa_tagstruct_putu32(t, (uint32_t) mode);
2810 pa_tagstruct_put_proplist(t, p);
2812 pa_pstream_send_tagstruct(s->context->pstream, t);
2813 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);
2815 /* Please note that we don't update s->proplist here, because we
2816 * don't export that field */
2821 pa_operation *pa_stream_proplist_remove(pa_stream *s, const char *const keys[], pa_stream_success_cb_t cb, void *userdata) {
2825 const char * const*k;
2828 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2830 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2831 PA_CHECK_VALIDITY_RETURN_NULL(s->context, keys && keys[0], PA_ERR_INVALID);
2832 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2833 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2834 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2836 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2838 t = pa_tagstruct_command(
2840 (uint32_t) (s->direction == PA_STREAM_RECORD ? PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST : PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST),
2842 pa_tagstruct_putu32(t, s->channel);
2844 for (k = keys; *k; k++)
2845 pa_tagstruct_puts(t, *k);
2847 pa_tagstruct_puts(t, NULL);
2849 pa_pstream_send_tagstruct(s->context->pstream, t);
2850 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);
2852 /* Please note that we don't update s->proplist here, because we
2853 * don't export that field */
2858 int pa_stream_set_monitor_stream(pa_stream *s, uint32_t sink_input_idx) {
2860 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2862 PA_CHECK_VALIDITY(s->context, !pa_detect_fork(), PA_ERR_FORKED);
2863 PA_CHECK_VALIDITY(s->context, sink_input_idx != PA_INVALID_INDEX, PA_ERR_INVALID);
2864 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
2865 PA_CHECK_VALIDITY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2867 s->direct_on_input = sink_input_idx;
2872 uint32_t pa_stream_get_monitor_stream(pa_stream *s) {
2874 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2876 PA_CHECK_VALIDITY_RETURN_ANY(s->context, !pa_detect_fork(), PA_ERR_FORKED, PA_INVALID_INDEX);
2877 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direct_on_input != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2878 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2880 return s->direct_on_input;