]> code.delx.au - pulseaudio/blob - src/pulse/stream.c
reorder a few things to get rid of an uneeded comparison
[pulseaudio] / src / pulse / stream.c
1 /***
2 This file is part of PulseAudio.
3
4 Copyright 2004-2006 Lennart Poettering
5 Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
6
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 of the License,
10 or (at your option) any later version.
11
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.
16
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
20 USA.
21 ***/
22
23 #ifdef HAVE_CONFIG_H
24 #include <config.h>
25 #endif
26
27 #include <string.h>
28 #include <stdio.h>
29 #include <string.h>
30
31 #include <pulse/def.h>
32 #include <pulse/timeval.h>
33 #include <pulse/xmalloc.h>
34
35 #include <pulsecore/pstream-util.h>
36 #include <pulsecore/log.h>
37 #include <pulsecore/hashmap.h>
38 #include <pulsecore/macro.h>
39 #include <pulsecore/rtclock.h>
40
41 #include "internal.h"
42
43 #define LATENCY_IPOL_INTERVAL_USEC (333*PA_USEC_PER_MSEC)
44
45 #define SMOOTHER_ADJUST_TIME (1000*PA_USEC_PER_MSEC)
46 #define SMOOTHER_HISTORY_TIME (5000*PA_USEC_PER_MSEC)
47 #define SMOOTHER_MIN_HISTORY (4)
48
49 pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) {
50 return pa_stream_new_with_proplist(c, name, ss, map, NULL);
51 }
52
53 static void reset_callbacks(pa_stream *s) {
54 s->read_callback = NULL;
55 s->read_userdata = NULL;
56 s->write_callback = NULL;
57 s->write_userdata = NULL;
58 s->state_callback = NULL;
59 s->state_userdata = NULL;
60 s->overflow_callback = NULL;
61 s->overflow_userdata = NULL;
62 s->underflow_callback = NULL;
63 s->underflow_userdata = NULL;
64 s->latency_update_callback = NULL;
65 s->latency_update_userdata = NULL;
66 s->moved_callback = NULL;
67 s->moved_userdata = NULL;
68 s->suspended_callback = NULL;
69 s->suspended_userdata = NULL;
70 s->started_callback = NULL;
71 s->started_userdata = NULL;
72 }
73
74 pa_stream *pa_stream_new_with_proplist(
75 pa_context *c,
76 const char *name,
77 const pa_sample_spec *ss,
78 const pa_channel_map *map,
79 pa_proplist *p) {
80
81 pa_stream *s;
82 int i;
83 pa_channel_map tmap;
84
85 pa_assert(c);
86 pa_assert(PA_REFCNT_VALUE(c) >= 1);
87
88 PA_CHECK_VALIDITY_RETURN_NULL(c, ss && pa_sample_spec_valid(ss), PA_ERR_INVALID);
89 PA_CHECK_VALIDITY_RETURN_NULL(c, c->version >= 12 || (ss->format != PA_SAMPLE_S32LE || ss->format != PA_SAMPLE_S32NE), PA_ERR_NOTSUPPORTED);
90 PA_CHECK_VALIDITY_RETURN_NULL(c, !map || (pa_channel_map_valid(map) && map->channels == ss->channels), PA_ERR_INVALID);
91 PA_CHECK_VALIDITY_RETURN_NULL(c, name || (p && pa_proplist_contains(p, PA_PROP_MEDIA_NAME)), PA_ERR_INVALID);
92
93 if (!map)
94 PA_CHECK_VALIDITY_RETURN_NULL(c, map = pa_channel_map_init_auto(&tmap, ss->channels, PA_CHANNEL_MAP_DEFAULT), PA_ERR_INVALID);
95
96 s = pa_xnew(pa_stream, 1);
97 PA_REFCNT_INIT(s);
98 s->context = c;
99 s->mainloop = c->mainloop;
100
101 s->direction = PA_STREAM_NODIRECTION;
102 s->state = PA_STREAM_UNCONNECTED;
103 s->flags = 0;
104
105 s->sample_spec = *ss;
106 s->channel_map = *map;
107
108 s->direct_on_input = PA_INVALID_INDEX;
109
110 s->proplist = p ? pa_proplist_copy(p) : pa_proplist_new();
111 if (name)
112 pa_proplist_sets(s->proplist, PA_PROP_MEDIA_NAME, name);
113
114 s->channel = 0;
115 s->channel_valid = FALSE;
116 s->syncid = c->csyncid++;
117 s->stream_index = PA_INVALID_INDEX;
118
119 s->requested_bytes = 0;
120 memset(&s->buffer_attr, 0, sizeof(s->buffer_attr));
121
122 /* We initialize der target length here, so that if the user
123 * passes no explicit buffering metrics the default is similar to
124 * what older PA versions provided. */
125 s->buffer_attr.tlength = pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
126
127 s->device_index = PA_INVALID_INDEX;
128 s->device_name = NULL;
129 s->suspended = FALSE;
130
131 pa_memchunk_reset(&s->peek_memchunk);
132 s->peek_data = NULL;
133
134 s->record_memblockq = NULL;
135
136 s->corked = FALSE;
137
138 memset(&s->timing_info, 0, sizeof(s->timing_info));
139 s->timing_info_valid = FALSE;
140
141 s->previous_time = 0;
142
143 s->read_index_not_before = 0;
144 s->write_index_not_before = 0;
145 for (i = 0; i < PA_MAX_WRITE_INDEX_CORRECTIONS; i++)
146 s->write_index_corrections[i].valid = 0;
147 s->current_write_index_correction = 0;
148
149 s->auto_timing_update_event = NULL;
150 s->auto_timing_update_requested = FALSE;
151
152 reset_callbacks(s);
153
154 s->smoother = NULL;
155
156 /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
157 PA_LLIST_PREPEND(pa_stream, c->streams, s);
158 pa_stream_ref(s);
159
160 return s;
161 }
162
163 static void stream_unlink(pa_stream *s) {
164 pa_operation *o, *n;
165 pa_assert(s);
166
167 if (!s->context)
168 return;
169
170 /* Detach from context */
171
172 /* Unref all operatio object that point to us */
173 for (o = s->context->operations; o; o = n) {
174 n = o->next;
175
176 if (o->stream == s)
177 pa_operation_cancel(o);
178 }
179
180 /* Drop all outstanding replies for this stream */
181 if (s->context->pdispatch)
182 pa_pdispatch_unregister_reply(s->context->pdispatch, s);
183
184 if (s->channel_valid) {
185 pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
186 s->channel = 0;
187 s->channel_valid = FALSE;
188 }
189
190 PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
191 pa_stream_unref(s);
192
193 s->context = NULL;
194
195 if (s->auto_timing_update_event) {
196 pa_assert(s->mainloop);
197 s->mainloop->time_free(s->auto_timing_update_event);
198 }
199
200 reset_callbacks(s);
201 }
202
203 static void stream_free(pa_stream *s) {
204 pa_assert(s);
205
206 stream_unlink(s);
207
208 if (s->peek_memchunk.memblock) {
209 if (s->peek_data)
210 pa_memblock_release(s->peek_memchunk.memblock);
211 pa_memblock_unref(s->peek_memchunk.memblock);
212 }
213
214 if (s->record_memblockq)
215 pa_memblockq_free(s->record_memblockq);
216
217 if (s->proplist)
218 pa_proplist_free(s->proplist);
219
220 if (s->smoother)
221 pa_smoother_free(s->smoother);
222
223 pa_xfree(s->device_name);
224 pa_xfree(s);
225 }
226
227 void pa_stream_unref(pa_stream *s) {
228 pa_assert(s);
229 pa_assert(PA_REFCNT_VALUE(s) >= 1);
230
231 if (PA_REFCNT_DEC(s) <= 0)
232 stream_free(s);
233 }
234
235 pa_stream* pa_stream_ref(pa_stream *s) {
236 pa_assert(s);
237 pa_assert(PA_REFCNT_VALUE(s) >= 1);
238
239 PA_REFCNT_INC(s);
240 return s;
241 }
242
243 pa_stream_state_t pa_stream_get_state(pa_stream *s) {
244 pa_assert(s);
245 pa_assert(PA_REFCNT_VALUE(s) >= 1);
246
247 return s->state;
248 }
249
250 pa_context* pa_stream_get_context(pa_stream *s) {
251 pa_assert(s);
252 pa_assert(PA_REFCNT_VALUE(s) >= 1);
253
254 return s->context;
255 }
256
257 uint32_t pa_stream_get_index(pa_stream *s) {
258 pa_assert(s);
259 pa_assert(PA_REFCNT_VALUE(s) >= 1);
260
261 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
262
263 return s->stream_index;
264 }
265
266 void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
267 pa_assert(s);
268 pa_assert(PA_REFCNT_VALUE(s) >= 1);
269
270 if (s->state == st)
271 return;
272
273 pa_stream_ref(s);
274
275 s->state = st;
276
277 if (s->state_callback)
278 s->state_callback(s, s->state_userdata);
279
280 if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED))
281 stream_unlink(s);
282
283 pa_stream_unref(s);
284 }
285
286 static void request_auto_timing_update(pa_stream *s, pa_bool_t force) {
287 pa_assert(s);
288 pa_assert(PA_REFCNT_VALUE(s) >= 1);
289
290 if (!(s->flags & PA_STREAM_AUTO_TIMING_UPDATE))
291 return;
292
293 if (s->state == PA_STREAM_READY &&
294 (force || !s->auto_timing_update_requested)) {
295 pa_operation *o;
296
297 /* pa_log("automatically requesting new timing data"); */
298
299 if ((o = pa_stream_update_timing_info(s, NULL, NULL))) {
300 pa_operation_unref(o);
301 s->auto_timing_update_requested = TRUE;
302 }
303 }
304
305 if (s->auto_timing_update_event) {
306 struct timeval next;
307 pa_gettimeofday(&next);
308 pa_timeval_add(&next, LATENCY_IPOL_INTERVAL_USEC);
309 s->mainloop->time_restart(s->auto_timing_update_event, &next);
310 }
311 }
312
313 void pa_command_stream_killed(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
314 pa_context *c = userdata;
315 pa_stream *s;
316 uint32_t channel;
317
318 pa_assert(pd);
319 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_KILLED || command == PA_COMMAND_RECORD_STREAM_KILLED);
320 pa_assert(t);
321 pa_assert(c);
322 pa_assert(PA_REFCNT_VALUE(c) >= 1);
323
324 pa_context_ref(c);
325
326 if (pa_tagstruct_getu32(t, &channel) < 0 ||
327 !pa_tagstruct_eof(t)) {
328 pa_context_fail(c, PA_ERR_PROTOCOL);
329 goto finish;
330 }
331
332 if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_KILLED ? c->playback_streams : c->record_streams, channel)))
333 goto finish;
334
335 if (s->state != PA_STREAM_READY)
336 goto finish;
337
338 pa_context_set_error(c, PA_ERR_KILLED);
339 pa_stream_set_state(s, PA_STREAM_FAILED);
340
341 finish:
342 pa_context_unref(c);
343 }
344
345 static void check_smoother_status(pa_stream *s, pa_bool_t aposteriori, pa_bool_t force_start, pa_bool_t force_stop) {
346 pa_usec_t x;
347
348 pa_assert(s);
349 pa_assert(!force_start || !force_stop);
350
351 if (!s->smoother)
352 return;
353
354 x = pa_rtclock_usec();
355
356 if (s->timing_info_valid) {
357 if (aposteriori)
358 x -= s->timing_info.transport_usec;
359 else
360 x += s->timing_info.transport_usec;
361
362 if (s->direction == PA_STREAM_PLAYBACK)
363 /* it takes a while until the pause/resume is actually
364 * audible */
365 x += s->timing_info.sink_usec;
366 else
367 /* Data froma while back will be dropped */
368 x -= s->timing_info.source_usec;
369 }
370
371 if (s->suspended || s->corked || force_stop)
372 pa_smoother_pause(s->smoother, x);
373 else if (force_start || s->buffer_attr.prebuf == 0)
374 pa_smoother_resume(s->smoother, x);
375
376 /* Please note that we have no idea if playback actually started
377 * if prebuf is non-zero! */
378 }
379
380 void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
381 pa_context *c = userdata;
382 pa_stream *s;
383 uint32_t channel;
384 const char *dn;
385 pa_bool_t suspended;
386 uint32_t di;
387 pa_usec_t usec;
388 uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0;
389
390 pa_assert(pd);
391 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED);
392 pa_assert(t);
393 pa_assert(c);
394 pa_assert(PA_REFCNT_VALUE(c) >= 1);
395
396 pa_context_ref(c);
397
398 if (c->version < 12) {
399 pa_context_fail(c, PA_ERR_PROTOCOL);
400 goto finish;
401 }
402
403 if (pa_tagstruct_getu32(t, &channel) < 0 ||
404 pa_tagstruct_getu32(t, &di) < 0 ||
405 pa_tagstruct_gets(t, &dn) < 0 ||
406 pa_tagstruct_get_boolean(t, &suspended) < 0) {
407 pa_context_fail(c, PA_ERR_PROTOCOL);
408 goto finish;
409 }
410
411 if (c->version >= 13) {
412
413 if (command == PA_COMMAND_RECORD_STREAM_MOVED) {
414 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
415 pa_tagstruct_getu32(t, &fragsize) < 0 ||
416 pa_tagstruct_get_usec(t, &usec) < 0) {
417 pa_context_fail(c, PA_ERR_PROTOCOL);
418 goto finish;
419 }
420 } else {
421 if (pa_tagstruct_getu32(t, &maxlength) < 0 ||
422 pa_tagstruct_getu32(t, &tlength) < 0 ||
423 pa_tagstruct_getu32(t, &prebuf) < 0 ||
424 pa_tagstruct_getu32(t, &minreq) < 0 ||
425 pa_tagstruct_get_usec(t, &usec) < 0) {
426 pa_context_fail(c, PA_ERR_PROTOCOL);
427 goto finish;
428 }
429 }
430 }
431
432 if (!pa_tagstruct_eof(t)) {
433 pa_context_fail(c, PA_ERR_PROTOCOL);
434 goto finish;
435 }
436
437 if (!dn || di == PA_INVALID_INDEX) {
438 pa_context_fail(c, PA_ERR_PROTOCOL);
439 goto finish;
440 }
441
442 if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_MOVED ? c->playback_streams : c->record_streams, channel)))
443 goto finish;
444
445 if (s->state != PA_STREAM_READY)
446 goto finish;
447
448 if (c->version >= 13) {
449 if (s->direction == PA_STREAM_RECORD)
450 s->timing_info.configured_source_usec = usec;
451 else
452 s->timing_info.configured_sink_usec = usec;
453
454 s->buffer_attr.maxlength = maxlength;
455 s->buffer_attr.fragsize = fragsize;
456 s->buffer_attr.tlength = tlength;
457 s->buffer_attr.prebuf = prebuf;
458 s->buffer_attr.minreq = minreq;
459 }
460
461 pa_xfree(s->device_name);
462 s->device_name = pa_xstrdup(dn);
463 s->device_index = di;
464
465 s->suspended = suspended;
466
467 check_smoother_status(s, TRUE, FALSE, FALSE);
468 request_auto_timing_update(s, TRUE);
469
470 if (s->moved_callback)
471 s->moved_callback(s, s->moved_userdata);
472
473 finish:
474 pa_context_unref(c);
475 }
476
477 void pa_command_stream_suspended(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
478 pa_context *c = userdata;
479 pa_stream *s;
480 uint32_t channel;
481 pa_bool_t suspended;
482
483 pa_assert(pd);
484 pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED || command == PA_COMMAND_RECORD_STREAM_SUSPENDED);
485 pa_assert(t);
486 pa_assert(c);
487 pa_assert(PA_REFCNT_VALUE(c) >= 1);
488
489 pa_context_ref(c);
490
491 if (c->version < 12) {
492 pa_context_fail(c, PA_ERR_PROTOCOL);
493 goto finish;
494 }
495
496 if (pa_tagstruct_getu32(t, &channel) < 0 ||
497 pa_tagstruct_get_boolean(t, &suspended) < 0 ||
498 !pa_tagstruct_eof(t)) {
499 pa_context_fail(c, PA_ERR_PROTOCOL);
500 goto finish;
501 }
502
503 if (!(s = pa_dynarray_get(command == PA_COMMAND_PLAYBACK_STREAM_SUSPENDED ? c->playback_streams : c->record_streams, channel)))
504 goto finish;
505
506 if (s->state != PA_STREAM_READY)
507 goto finish;
508
509 s->suspended = suspended;
510
511 check_smoother_status(s, TRUE, FALSE, FALSE);
512 request_auto_timing_update(s, TRUE);
513
514 if (s->suspended_callback)
515 s->suspended_callback(s, s->suspended_userdata);
516
517 finish:
518 pa_context_unref(c);
519 }
520
521 void pa_command_stream_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
522 pa_context *c = userdata;
523 pa_stream *s;
524 uint32_t channel;
525
526 pa_assert(pd);
527 pa_assert(command == PA_COMMAND_STARTED);
528 pa_assert(t);
529 pa_assert(c);
530 pa_assert(PA_REFCNT_VALUE(c) >= 1);
531
532 pa_context_ref(c);
533
534 if (c->version < 13) {
535 pa_context_fail(c, PA_ERR_PROTOCOL);
536 goto finish;
537 }
538
539 if (pa_tagstruct_getu32(t, &channel) < 0 ||
540 !pa_tagstruct_eof(t)) {
541 pa_context_fail(c, PA_ERR_PROTOCOL);
542 goto finish;
543 }
544
545 if (!(s = pa_dynarray_get(c->playback_streams, channel)))
546 goto finish;
547
548 if (s->state != PA_STREAM_READY)
549 goto finish;
550
551 check_smoother_status(s, TRUE, TRUE, FALSE);
552 request_auto_timing_update(s, TRUE);
553
554 if (s->started_callback)
555 s->started_callback(s, s->suspended_userdata);
556
557 finish:
558 pa_context_unref(c);
559 }
560
561 void pa_command_request(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
562 pa_stream *s;
563 pa_context *c = userdata;
564 uint32_t bytes, channel;
565
566 pa_assert(pd);
567 pa_assert(command == PA_COMMAND_REQUEST);
568 pa_assert(t);
569 pa_assert(c);
570 pa_assert(PA_REFCNT_VALUE(c) >= 1);
571
572 pa_context_ref(c);
573
574 if (pa_tagstruct_getu32(t, &channel) < 0 ||
575 pa_tagstruct_getu32(t, &bytes) < 0 ||
576 !pa_tagstruct_eof(t)) {
577 pa_context_fail(c, PA_ERR_PROTOCOL);
578 goto finish;
579 }
580
581 if (!(s = pa_dynarray_get(c->playback_streams, channel)))
582 goto finish;
583
584 if (s->state != PA_STREAM_READY)
585 goto finish;
586
587 s->requested_bytes += bytes;
588
589 if (s->requested_bytes > 0 && s->write_callback)
590 s->write_callback(s, s->requested_bytes, s->write_userdata);
591
592 finish:
593 pa_context_unref(c);
594 }
595
596 void pa_command_overflow_or_underflow(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
597 pa_stream *s;
598 pa_context *c = userdata;
599 uint32_t channel;
600
601 pa_assert(pd);
602 pa_assert(command == PA_COMMAND_OVERFLOW || command == PA_COMMAND_UNDERFLOW);
603 pa_assert(t);
604 pa_assert(c);
605 pa_assert(PA_REFCNT_VALUE(c) >= 1);
606
607 pa_context_ref(c);
608
609 if (pa_tagstruct_getu32(t, &channel) < 0 ||
610 !pa_tagstruct_eof(t)) {
611 pa_context_fail(c, PA_ERR_PROTOCOL);
612 goto finish;
613 }
614
615 if (!(s = pa_dynarray_get(c->playback_streams, channel)))
616 goto finish;
617
618 if (s->state != PA_STREAM_READY)
619 goto finish;
620
621 if (s->buffer_attr.prebuf > 0)
622 check_smoother_status(s, TRUE, FALSE, TRUE);
623
624 request_auto_timing_update(s, TRUE);
625
626 if (command == PA_COMMAND_OVERFLOW) {
627 if (s->overflow_callback)
628 s->overflow_callback(s, s->overflow_userdata);
629 } else if (command == PA_COMMAND_UNDERFLOW) {
630 if (s->underflow_callback)
631 s->underflow_callback(s, s->underflow_userdata);
632 }
633
634 finish:
635 pa_context_unref(c);
636 }
637
638 static void invalidate_indexes(pa_stream *s, pa_bool_t r, pa_bool_t w) {
639 pa_assert(s);
640 pa_assert(PA_REFCNT_VALUE(s) >= 1);
641
642 /* pa_log("invalidate r:%u w:%u tag:%u", r, w, s->context->ctag); */
643
644 if (s->state != PA_STREAM_READY)
645 return;
646
647 if (w) {
648 s->write_index_not_before = s->context->ctag;
649
650 if (s->timing_info_valid)
651 s->timing_info.write_index_corrupt = TRUE;
652
653 /* pa_log("write_index invalidated"); */
654 }
655
656 if (r) {
657 s->read_index_not_before = s->context->ctag;
658
659 if (s->timing_info_valid)
660 s->timing_info.read_index_corrupt = TRUE;
661
662 /* pa_log("read_index invalidated"); */
663 }
664
665 request_auto_timing_update(s, TRUE);
666 }
667
668 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) {
669 pa_stream *s = userdata;
670
671 pa_assert(s);
672 pa_assert(PA_REFCNT_VALUE(s) >= 1);
673
674 pa_stream_ref(s);
675 request_auto_timing_update(s, FALSE);
676 pa_stream_unref(s);
677 }
678
679 static void create_stream_complete(pa_stream *s) {
680 pa_assert(s);
681 pa_assert(PA_REFCNT_VALUE(s) >= 1);
682 pa_assert(s->state == PA_STREAM_CREATING);
683
684 pa_stream_set_state(s, PA_STREAM_READY);
685
686 if (s->requested_bytes > 0 && s->write_callback)
687 s->write_callback(s, s->requested_bytes, s->write_userdata);
688
689 if (s->flags & PA_STREAM_AUTO_TIMING_UPDATE) {
690 struct timeval tv;
691 pa_gettimeofday(&tv);
692 tv.tv_usec += LATENCY_IPOL_INTERVAL_USEC; /* every 100 ms */
693 pa_assert(!s->auto_timing_update_event);
694 s->auto_timing_update_event = s->mainloop->time_new(s->mainloop, &tv, &auto_timing_update_callback, s);
695
696 request_auto_timing_update(s, TRUE);
697 }
698
699 check_smoother_status(s, TRUE, FALSE, FALSE);
700 }
701
702 static void automatic_buffer_attr(pa_stream *s, pa_buffer_attr *attr, const pa_sample_spec *ss) {
703 pa_assert(s);
704 pa_assert(attr);
705 pa_assert(ss);
706
707 if (s->context->version >= 13)
708 return;
709
710 /* Version older than 0.9.10 didn't do server side buffer_attr
711 * selection, hence we have to fake it on the client side. */
712
713 /* We choose fairly conservative values here, to not confuse
714 * old clients with extremely large playback buffers */
715
716 if (!attr->maxlength <= 0)
717 attr->maxlength = 4*1024*1024; /* 4MB is the maximum queue length PulseAudio <= 0.9.9 supported. */
718
719 if (!attr->tlength <= 0)
720 attr->tlength = pa_usec_to_bytes(250*PA_USEC_PER_MSEC, ss); /* 250ms of buffering */
721
722 if (!attr->minreq <= 0)
723 attr->minreq = (attr->tlength)/5; /* Ask for more data when there are only 200ms left in the playback buffer */
724
725 if (!attr->prebuf)
726 attr->prebuf = attr->tlength; /* Start to play only when the playback is fully filled up once */
727
728 if (!attr->fragsize)
729 attr->fragsize = attr->tlength; /* Pass data to the app only when the buffer is filled up once */
730 }
731
732 void pa_create_stream_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
733 pa_stream *s = userdata;
734
735 pa_assert(pd);
736 pa_assert(s);
737 pa_assert(PA_REFCNT_VALUE(s) >= 1);
738 pa_assert(s->state == PA_STREAM_CREATING);
739
740 pa_stream_ref(s);
741
742 if (command != PA_COMMAND_REPLY) {
743 if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
744 goto finish;
745
746 pa_stream_set_state(s, PA_STREAM_FAILED);
747 goto finish;
748 }
749
750 if (pa_tagstruct_getu32(t, &s->channel) < 0 ||
751 s->channel == PA_INVALID_INDEX ||
752 ((s->direction != PA_STREAM_UPLOAD) && (pa_tagstruct_getu32(t, &s->stream_index) < 0 || s->stream_index == PA_INVALID_INDEX)) ||
753 ((s->direction != PA_STREAM_RECORD) && pa_tagstruct_getu32(t, &s->requested_bytes) < 0)) {
754 pa_context_fail(s->context, PA_ERR_PROTOCOL);
755 goto finish;
756 }
757
758 if (s->context->version >= 9) {
759 if (s->direction == PA_STREAM_PLAYBACK) {
760 if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
761 pa_tagstruct_getu32(t, &s->buffer_attr.tlength) < 0 ||
762 pa_tagstruct_getu32(t, &s->buffer_attr.prebuf) < 0 ||
763 pa_tagstruct_getu32(t, &s->buffer_attr.minreq) < 0) {
764 pa_context_fail(s->context, PA_ERR_PROTOCOL);
765 goto finish;
766 }
767 } else if (s->direction == PA_STREAM_RECORD) {
768 if (pa_tagstruct_getu32(t, &s->buffer_attr.maxlength) < 0 ||
769 pa_tagstruct_getu32(t, &s->buffer_attr.fragsize) < 0) {
770 pa_context_fail(s->context, PA_ERR_PROTOCOL);
771 goto finish;
772 }
773 }
774 }
775
776 if (s->context->version >= 12 && s->direction != PA_STREAM_UPLOAD) {
777 pa_sample_spec ss;
778 pa_channel_map cm;
779 const char *dn = NULL;
780 pa_bool_t suspended;
781
782 if (pa_tagstruct_get_sample_spec(t, &ss) < 0 ||
783 pa_tagstruct_get_channel_map(t, &cm) < 0 ||
784 pa_tagstruct_getu32(t, &s->device_index) < 0 ||
785 pa_tagstruct_gets(t, &dn) < 0 ||
786 pa_tagstruct_get_boolean(t, &suspended) < 0) {
787 pa_context_fail(s->context, PA_ERR_PROTOCOL);
788 goto finish;
789 }
790
791 if (!dn || s->device_index == PA_INVALID_INDEX ||
792 ss.channels != cm.channels ||
793 !pa_channel_map_valid(&cm) ||
794 !pa_sample_spec_valid(&ss) ||
795 (!(s->flags & PA_STREAM_FIX_FORMAT) && ss.format != s->sample_spec.format) ||
796 (!(s->flags & PA_STREAM_FIX_RATE) && ss.rate != s->sample_spec.rate) ||
797 (!(s->flags & PA_STREAM_FIX_CHANNELS) && !pa_channel_map_equal(&cm, &s->channel_map))) {
798 pa_context_fail(s->context, PA_ERR_PROTOCOL);
799 goto finish;
800 }
801
802 pa_xfree(s->device_name);
803 s->device_name = pa_xstrdup(dn);
804 s->suspended = suspended;
805
806 s->channel_map = cm;
807 s->sample_spec = ss;
808 }
809
810 if (s->context->version >= 13 && s->direction != PA_STREAM_UPLOAD) {
811 pa_usec_t usec;
812
813 if (pa_tagstruct_get_usec(t, &usec) < 0) {
814 pa_context_fail(s->context, PA_ERR_PROTOCOL);
815 goto finish;
816 }
817
818 if (s->direction == PA_STREAM_RECORD)
819 s->timing_info.configured_source_usec = usec;
820 else
821 s->timing_info.configured_sink_usec = usec;
822 }
823
824 if (!pa_tagstruct_eof(t)) {
825 pa_context_fail(s->context, PA_ERR_PROTOCOL);
826 goto finish;
827 }
828
829 if (s->direction == PA_STREAM_RECORD) {
830 pa_assert(!s->record_memblockq);
831
832 s->record_memblockq = pa_memblockq_new(
833 0,
834 s->buffer_attr.maxlength,
835 0,
836 pa_frame_size(&s->sample_spec),
837 1,
838 0,
839 0,
840 NULL);
841 }
842
843 s->channel_valid = TRUE;
844 pa_dynarray_put((s->direction == PA_STREAM_RECORD) ? s->context->record_streams : s->context->playback_streams, s->channel, s);
845
846 create_stream_complete(s);
847
848 finish:
849 pa_stream_unref(s);
850 }
851
852 static int create_stream(
853 pa_stream_direction_t direction,
854 pa_stream *s,
855 const char *dev,
856 const pa_buffer_attr *attr,
857 pa_stream_flags_t flags,
858 const pa_cvolume *volume,
859 pa_stream *sync_stream) {
860
861 pa_tagstruct *t;
862 uint32_t tag;
863
864 pa_assert(s);
865 pa_assert(PA_REFCNT_VALUE(s) >= 1);
866 pa_assert(direction == PA_STREAM_PLAYBACK || direction == PA_STREAM_RECORD);
867
868 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
869 PA_CHECK_VALIDITY(s->context, s->direct_on_input == PA_INVALID_INDEX || direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
870 PA_CHECK_VALIDITY(s->context, !(flags & ~(PA_STREAM_START_CORKED|
871 PA_STREAM_INTERPOLATE_TIMING|
872 PA_STREAM_NOT_MONOTONOUS|
873 PA_STREAM_AUTO_TIMING_UPDATE|
874 PA_STREAM_NO_REMAP_CHANNELS|
875 PA_STREAM_NO_REMIX_CHANNELS|
876 PA_STREAM_FIX_FORMAT|
877 PA_STREAM_FIX_RATE|
878 PA_STREAM_FIX_CHANNELS|
879 PA_STREAM_DONT_MOVE|
880 PA_STREAM_VARIABLE_RATE|
881 PA_STREAM_PEAK_DETECT|
882 PA_STREAM_START_MUTED|
883 PA_STREAM_ADJUST_LATENCY)), PA_ERR_INVALID);
884
885 PA_CHECK_VALIDITY(s->context, s->context->version >= 12 || !(flags & PA_STREAM_VARIABLE_RATE), PA_ERR_NOTSUPPORTED);
886 PA_CHECK_VALIDITY(s->context, s->context->version >= 13 || !(flags & PA_STREAM_PEAK_DETECT), PA_ERR_NOTSUPPORTED);
887 /* Althought some of the other flags are not supported on older
888 * version, we don't check for them here, because it doesn't hurt
889 * when they are passed but actually not supported. This makes
890 * client development easier */
891
892 PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_PLAYBACK || !(flags & (PA_STREAM_START_MUTED)), PA_ERR_INVALID);
893 PA_CHECK_VALIDITY(s->context, direction == PA_STREAM_RECORD || !(flags & (PA_STREAM_PEAK_DETECT)), PA_ERR_INVALID);
894 PA_CHECK_VALIDITY(s->context, !volume || volume->channels == s->sample_spec.channels, PA_ERR_INVALID);
895 PA_CHECK_VALIDITY(s->context, !sync_stream || (direction == PA_STREAM_PLAYBACK && sync_stream->direction == PA_STREAM_PLAYBACK), PA_ERR_INVALID);
896
897 pa_stream_ref(s);
898
899 s->direction = direction;
900 s->flags = flags;
901 s->corked = !!(flags & PA_STREAM_START_CORKED);
902
903 if (sync_stream)
904 s->syncid = sync_stream->syncid;
905
906 if (attr)
907 s->buffer_attr = *attr;
908 automatic_buffer_attr(s, &s->buffer_attr, &s->sample_spec);
909
910 if (flags & PA_STREAM_INTERPOLATE_TIMING) {
911 pa_usec_t x;
912
913 if (s->smoother)
914 pa_smoother_free(s->smoother);
915
916 s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONIC), SMOOTHER_MIN_HISTORY);
917
918 x = pa_rtclock_usec();
919 pa_smoother_set_time_offset(s->smoother, x);
920 pa_smoother_pause(s->smoother, x);
921 }
922
923 if (!dev)
924 dev = s->direction == PA_STREAM_PLAYBACK ? s->context->conf->default_sink : s->context->conf->default_source;
925
926 t = pa_tagstruct_command(
927 s->context,
928 s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CREATE_PLAYBACK_STREAM : PA_COMMAND_CREATE_RECORD_STREAM,
929 &tag);
930
931 if (s->context->version < 13)
932 pa_tagstruct_puts(t, pa_proplist_gets(s->proplist, PA_PROP_MEDIA_NAME));
933
934 pa_tagstruct_put(
935 t,
936 PA_TAG_SAMPLE_SPEC, &s->sample_spec,
937 PA_TAG_CHANNEL_MAP, &s->channel_map,
938 PA_TAG_U32, PA_INVALID_INDEX,
939 PA_TAG_STRING, dev,
940 PA_TAG_U32, s->buffer_attr.maxlength,
941 PA_TAG_BOOLEAN, s->corked,
942 PA_TAG_INVALID);
943
944 if (s->direction == PA_STREAM_PLAYBACK) {
945 pa_cvolume cv;
946
947 pa_tagstruct_put(
948 t,
949 PA_TAG_U32, s->buffer_attr.tlength,
950 PA_TAG_U32, s->buffer_attr.prebuf,
951 PA_TAG_U32, s->buffer_attr.minreq,
952 PA_TAG_U32, s->syncid,
953 PA_TAG_INVALID);
954
955 if (!volume)
956 volume = pa_cvolume_reset(&cv, s->sample_spec.channels);
957
958 pa_tagstruct_put_cvolume(t, volume);
959 } else
960 pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
961
962 if (s->context->version >= 12) {
963 pa_tagstruct_put(
964 t,
965 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMAP_CHANNELS,
966 PA_TAG_BOOLEAN, flags & PA_STREAM_NO_REMIX_CHANNELS,
967 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_FORMAT,
968 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_RATE,
969 PA_TAG_BOOLEAN, flags & PA_STREAM_FIX_CHANNELS,
970 PA_TAG_BOOLEAN, flags & PA_STREAM_DONT_MOVE,
971 PA_TAG_BOOLEAN, flags & PA_STREAM_VARIABLE_RATE,
972 PA_TAG_INVALID);
973 }
974
975 if (s->context->version >= 13) {
976
977 if (s->direction == PA_STREAM_PLAYBACK)
978 pa_tagstruct_put_boolean(t, flags & PA_STREAM_START_MUTED);
979 else
980 pa_tagstruct_put_boolean(t, flags & PA_STREAM_PEAK_DETECT);
981
982 pa_tagstruct_put(
983 t,
984 PA_TAG_BOOLEAN, flags & PA_STREAM_ADJUST_LATENCY,
985 PA_TAG_PROPLIST, s->proplist,
986 PA_TAG_INVALID);
987
988 if (s->direction == PA_STREAM_RECORD)
989 pa_tagstruct_putu32(t, s->direct_on_input);
990 }
991
992 pa_pstream_send_tagstruct(s->context->pstream, t);
993 pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
994
995 pa_stream_set_state(s, PA_STREAM_CREATING);
996
997 pa_stream_unref(s);
998 return 0;
999 }
1000
1001 int pa_stream_connect_playback(
1002 pa_stream *s,
1003 const char *dev,
1004 const pa_buffer_attr *attr,
1005 pa_stream_flags_t flags,
1006 pa_cvolume *volume,
1007 pa_stream *sync_stream) {
1008
1009 pa_assert(s);
1010 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1011
1012 return create_stream(PA_STREAM_PLAYBACK, s, dev, attr, flags, volume, sync_stream);
1013 }
1014
1015 int pa_stream_connect_record(
1016 pa_stream *s,
1017 const char *dev,
1018 const pa_buffer_attr *attr,
1019 pa_stream_flags_t flags) {
1020
1021 pa_assert(s);
1022 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1023
1024 return create_stream(PA_STREAM_RECORD, s, dev, attr, flags, NULL, NULL);
1025 }
1026
1027 int pa_stream_write(
1028 pa_stream *s,
1029 const void *data,
1030 size_t length,
1031 void (*free_cb)(void *p),
1032 int64_t offset,
1033 pa_seek_mode_t seek) {
1034
1035 pa_memchunk chunk;
1036 pa_seek_mode_t t_seek;
1037 int64_t t_offset;
1038 size_t t_length;
1039 const void *t_data;
1040
1041 pa_assert(s);
1042 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1043 pa_assert(data);
1044
1045 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1046 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || s->direction == PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1047 PA_CHECK_VALIDITY(s->context, seek <= PA_SEEK_RELATIVE_END, PA_ERR_INVALID);
1048 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_PLAYBACK || (seek == PA_SEEK_RELATIVE && offset == 0), PA_ERR_INVALID);
1049
1050 if (length <= 0)
1051 return 0;
1052
1053 t_seek = seek;
1054 t_offset = offset;
1055 t_length = length;
1056 t_data = data;
1057
1058 while (t_length > 0) {
1059
1060 chunk.index = 0;
1061
1062 if (free_cb && !pa_pstream_get_shm(s->context->pstream)) {
1063 chunk.memblock = pa_memblock_new_user(s->context->mempool, (void*) t_data, t_length, free_cb, 1);
1064 chunk.length = t_length;
1065 } else {
1066 void *d;
1067
1068 chunk.length = PA_MIN(t_length, pa_mempool_block_size_max(s->context->mempool));
1069 chunk.memblock = pa_memblock_new(s->context->mempool, chunk.length);
1070
1071 d = pa_memblock_acquire(chunk.memblock);
1072 memcpy(d, t_data, chunk.length);
1073 pa_memblock_release(chunk.memblock);
1074 }
1075
1076 pa_pstream_send_memblock(s->context->pstream, s->channel, t_offset, t_seek, &chunk);
1077
1078 t_offset = 0;
1079 t_seek = PA_SEEK_RELATIVE;
1080
1081 t_data = (const uint8_t*) t_data + chunk.length;
1082 t_length -= chunk.length;
1083
1084 pa_memblock_unref(chunk.memblock);
1085 }
1086
1087 if (free_cb && pa_pstream_get_shm(s->context->pstream))
1088 free_cb((void*) data);
1089
1090 if (length < s->requested_bytes)
1091 s->requested_bytes -= length;
1092 else
1093 s->requested_bytes = 0;
1094
1095 /* FIXME!!! ^^^ will break when offset is != 0 and mode is not RELATIVE*/
1096
1097 if (s->direction == PA_STREAM_PLAYBACK) {
1098
1099 /* Update latency request correction */
1100 if (s->write_index_corrections[s->current_write_index_correction].valid) {
1101
1102 if (seek == PA_SEEK_ABSOLUTE) {
1103 s->write_index_corrections[s->current_write_index_correction].corrupt = FALSE;
1104 s->write_index_corrections[s->current_write_index_correction].absolute = TRUE;
1105 s->write_index_corrections[s->current_write_index_correction].value = offset + length;
1106 } else if (seek == PA_SEEK_RELATIVE) {
1107 if (!s->write_index_corrections[s->current_write_index_correction].corrupt)
1108 s->write_index_corrections[s->current_write_index_correction].value += offset + length;
1109 } else
1110 s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
1111 }
1112
1113 /* Update the write index in the already available latency data */
1114 if (s->timing_info_valid) {
1115
1116 if (seek == PA_SEEK_ABSOLUTE) {
1117 s->timing_info.write_index_corrupt = FALSE;
1118 s->timing_info.write_index = offset + length;
1119 } else if (seek == PA_SEEK_RELATIVE) {
1120 if (!s->timing_info.write_index_corrupt)
1121 s->timing_info.write_index += offset + length;
1122 } else
1123 s->timing_info.write_index_corrupt = TRUE;
1124 }
1125
1126 if (!s->timing_info_valid || s->timing_info.write_index_corrupt)
1127 request_auto_timing_update(s, TRUE);
1128 }
1129
1130 return 0;
1131 }
1132
1133 int pa_stream_peek(pa_stream *s, const void **data, size_t *length) {
1134 pa_assert(s);
1135 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1136 pa_assert(data);
1137 pa_assert(length);
1138
1139 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1140 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1141
1142 if (!s->peek_memchunk.memblock) {
1143
1144 if (pa_memblockq_peek(s->record_memblockq, &s->peek_memchunk) < 0) {
1145 *data = NULL;
1146 *length = 0;
1147 return 0;
1148 }
1149
1150 s->peek_data = pa_memblock_acquire(s->peek_memchunk.memblock);
1151 }
1152
1153 pa_assert(s->peek_data);
1154 *data = (uint8_t*) s->peek_data + s->peek_memchunk.index;
1155 *length = s->peek_memchunk.length;
1156 return 0;
1157 }
1158
1159 int pa_stream_drop(pa_stream *s) {
1160 pa_assert(s);
1161 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1162
1163 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1164 PA_CHECK_VALIDITY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE);
1165 PA_CHECK_VALIDITY(s->context, s->peek_memchunk.memblock, PA_ERR_BADSTATE);
1166
1167 pa_memblockq_drop(s->record_memblockq, s->peek_memchunk.length);
1168
1169 /* Fix the simulated local read index */
1170 if (s->timing_info_valid && !s->timing_info.read_index_corrupt)
1171 s->timing_info.read_index += s->peek_memchunk.length;
1172
1173 pa_assert(s->peek_data);
1174 pa_memblock_release(s->peek_memchunk.memblock);
1175 pa_memblock_unref(s->peek_memchunk.memblock);
1176 pa_memchunk_reset(&s->peek_memchunk);
1177
1178 return 0;
1179 }
1180
1181 size_t pa_stream_writable_size(pa_stream *s) {
1182 pa_assert(s);
1183 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1184
1185 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1186 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1187
1188 return s->requested_bytes;
1189 }
1190
1191 size_t pa_stream_readable_size(pa_stream *s) {
1192 pa_assert(s);
1193 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1194
1195 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, (size_t) -1);
1196 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction == PA_STREAM_RECORD, PA_ERR_BADSTATE, (size_t) -1);
1197
1198 return pa_memblockq_get_length(s->record_memblockq);
1199 }
1200
1201 pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1202 pa_operation *o;
1203 pa_tagstruct *t;
1204 uint32_t tag;
1205
1206 pa_assert(s);
1207 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1208
1209 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1210 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1211
1212 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1213
1214 t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
1215 pa_tagstruct_putu32(t, s->channel);
1216 pa_pstream_send_tagstruct(s->context->pstream, t);
1217 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);
1218
1219 return o;
1220 }
1221
1222 static pa_usec_t calc_time(pa_stream *s, pa_bool_t ignore_transport) {
1223 pa_usec_t usec;
1224
1225 pa_assert(s);
1226 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1227 pa_assert(s->state == PA_STREAM_READY);
1228 pa_assert(s->direction != PA_STREAM_UPLOAD);
1229 pa_assert(s->timing_info_valid);
1230 pa_assert(s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt);
1231 pa_assert(s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt);
1232
1233 if (s->direction == PA_STREAM_PLAYBACK) {
1234 /* The last byte that was written into the output device
1235 * had this time value associated */
1236 usec = pa_bytes_to_usec(s->timing_info.read_index < 0 ? 0 : (uint64_t) s->timing_info.read_index, &s->sample_spec);
1237
1238 if (!s->corked && !s->suspended) {
1239
1240 if (!ignore_transport)
1241 /* Because the latency info took a little time to come
1242 * to us, we assume that the real output time is actually
1243 * a little ahead */
1244 usec += s->timing_info.transport_usec;
1245
1246 /* However, the output device usually maintains a buffer
1247 too, hence the real sample currently played is a little
1248 back */
1249 if (s->timing_info.sink_usec >= usec)
1250 usec = 0;
1251 else
1252 usec -= s->timing_info.sink_usec;
1253 }
1254
1255 } else if (s->direction == PA_STREAM_RECORD) {
1256 /* The last byte written into the server side queue had
1257 * this time value associated */
1258 usec = pa_bytes_to_usec(s->timing_info.write_index < 0 ? 0 : (uint64_t) s->timing_info.write_index, &s->sample_spec);
1259
1260 if (!s->corked && !s->suspended) {
1261
1262 if (!ignore_transport)
1263 /* Add transport latency */
1264 usec += s->timing_info.transport_usec;
1265
1266 /* Add latency of data in device buffer */
1267 usec += s->timing_info.source_usec;
1268
1269 /* If this is a monitor source, we need to correct the
1270 * time by the playback device buffer */
1271 if (s->timing_info.sink_usec >= usec)
1272 usec = 0;
1273 else
1274 usec -= s->timing_info.sink_usec;
1275 }
1276 }
1277
1278 return usec;
1279 }
1280
1281 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
1282 pa_operation *o = userdata;
1283 struct timeval local, remote, now;
1284 pa_timing_info *i;
1285 pa_bool_t playing = FALSE;
1286 uint64_t underrun_for = 0, playing_for = 0;
1287
1288 pa_assert(pd);
1289 pa_assert(o);
1290 pa_assert(PA_REFCNT_VALUE(o) >= 1);
1291
1292 if (!o->context || !o->stream)
1293 goto finish;
1294
1295 i = &o->stream->timing_info;
1296
1297 o->stream->timing_info_valid = FALSE;
1298 i->write_index_corrupt = TRUE;
1299 i->read_index_corrupt = TRUE;
1300
1301 if (command != PA_COMMAND_REPLY) {
1302 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1303 goto finish;
1304
1305 } else {
1306
1307 if (pa_tagstruct_get_usec(t, &i->sink_usec) < 0 ||
1308 pa_tagstruct_get_usec(t, &i->source_usec) < 0 ||
1309 pa_tagstruct_get_boolean(t, &playing) < 0 ||
1310 pa_tagstruct_get_timeval(t, &local) < 0 ||
1311 pa_tagstruct_get_timeval(t, &remote) < 0 ||
1312 pa_tagstruct_gets64(t, &i->write_index) < 0 ||
1313 pa_tagstruct_gets64(t, &i->read_index) < 0) {
1314
1315 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1316 goto finish;
1317 }
1318
1319 if (o->context->version >= 13 &&
1320 o->stream->direction == PA_STREAM_PLAYBACK)
1321 if (pa_tagstruct_getu64(t, &underrun_for) < 0 ||
1322 pa_tagstruct_getu64(t, &playing_for) < 0) {
1323
1324 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1325 goto finish;
1326 }
1327
1328
1329 if (!pa_tagstruct_eof(t)) {
1330 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1331 goto finish;
1332 }
1333 o->stream->timing_info_valid = TRUE;
1334 i->write_index_corrupt = FALSE;
1335 i->read_index_corrupt = FALSE;
1336
1337 i->playing = (int) playing;
1338 i->since_underrun = playing ? playing_for : underrun_for;
1339
1340 pa_gettimeofday(&now);
1341
1342 /* Calculcate timestamps */
1343 if (pa_timeval_cmp(&local, &remote) <= 0 && pa_timeval_cmp(&remote, &now) <= 0) {
1344 /* local and remote seem to have synchronized clocks */
1345
1346 if (o->stream->direction == PA_STREAM_PLAYBACK)
1347 i->transport_usec = pa_timeval_diff(&remote, &local);
1348 else
1349 i->transport_usec = pa_timeval_diff(&now, &remote);
1350
1351 i->synchronized_clocks = TRUE;
1352 i->timestamp = remote;
1353 } else {
1354 /* clocks are not synchronized, let's estimate latency then */
1355 i->transport_usec = pa_timeval_diff(&now, &local)/2;
1356 i->synchronized_clocks = FALSE;
1357 i->timestamp = local;
1358 pa_timeval_add(&i->timestamp, i->transport_usec);
1359 }
1360
1361 /* Invalidate read and write indexes if necessary */
1362 if (tag < o->stream->read_index_not_before)
1363 i->read_index_corrupt = TRUE;
1364
1365 if (tag < o->stream->write_index_not_before)
1366 i->write_index_corrupt = TRUE;
1367
1368 if (o->stream->direction == PA_STREAM_PLAYBACK) {
1369 /* Write index correction */
1370
1371 int n, j;
1372 uint32_t ctag = tag;
1373
1374 /* Go through the saved correction values and add up the
1375 * total correction.*/
1376 for (n = 0, j = o->stream->current_write_index_correction+1;
1377 n < PA_MAX_WRITE_INDEX_CORRECTIONS;
1378 n++, j = (j + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS) {
1379
1380 /* Step over invalid data or out-of-date data */
1381 if (!o->stream->write_index_corrections[j].valid ||
1382 o->stream->write_index_corrections[j].tag < ctag)
1383 continue;
1384
1385 /* Make sure that everything is in order */
1386 ctag = o->stream->write_index_corrections[j].tag+1;
1387
1388 /* Now fix the write index */
1389 if (o->stream->write_index_corrections[j].corrupt) {
1390 /* A corrupting seek was made */
1391 i->write_index_corrupt = TRUE;
1392 } else if (o->stream->write_index_corrections[j].absolute) {
1393 /* An absolute seek was made */
1394 i->write_index = o->stream->write_index_corrections[j].value;
1395 i->write_index_corrupt = FALSE;
1396 } else if (!i->write_index_corrupt) {
1397 /* A relative seek was made */
1398 i->write_index += o->stream->write_index_corrections[j].value;
1399 }
1400 }
1401
1402 /* Clear old correction entries */
1403 for (n = 0; n < PA_MAX_WRITE_INDEX_CORRECTIONS; n++) {
1404 if (!o->stream->write_index_corrections[n].valid)
1405 continue;
1406
1407 if (o->stream->write_index_corrections[n].tag <= tag)
1408 o->stream->write_index_corrections[n].valid = FALSE;
1409 }
1410 }
1411
1412 if (o->stream->direction == PA_STREAM_RECORD) {
1413 /* Read index correction */
1414
1415 if (!i->read_index_corrupt)
1416 i->read_index -= pa_memblockq_get_length(o->stream->record_memblockq);
1417 }
1418
1419 /* Update smoother */
1420 if (o->stream->smoother) {
1421 pa_usec_t u, x;
1422
1423 u = x = pa_rtclock_usec() - i->transport_usec;
1424
1425 if (o->stream->direction == PA_STREAM_PLAYBACK &&
1426 o->context->version >= 13) {
1427 pa_usec_t su;
1428
1429 /* If we weren't playing then it will take some time
1430 * until the audio will actually come out through the
1431 * speakers. Since we follow that timing here, we need
1432 * to try to fix this up */
1433
1434 su = pa_bytes_to_usec(i->since_underrun, &o->stream->sample_spec);
1435
1436 if (su < i->sink_usec)
1437 x += i->sink_usec - su;
1438 }
1439
1440 if (!i->playing)
1441 pa_smoother_pause(o->stream->smoother, x);
1442
1443 /* Update the smoother */
1444 if ((o->stream->direction == PA_STREAM_PLAYBACK && !i->read_index_corrupt) ||
1445 (o->stream->direction == PA_STREAM_RECORD && !i->write_index_corrupt))
1446 pa_smoother_put(o->stream->smoother, u, calc_time(o->stream, TRUE));
1447
1448 if (i->playing)
1449 pa_smoother_resume(o->stream->smoother, x);
1450 }
1451 }
1452
1453 o->stream->auto_timing_update_requested = FALSE;
1454
1455 if (o->stream->latency_update_callback)
1456 o->stream->latency_update_callback(o->stream, o->stream->latency_update_userdata);
1457
1458 if (o->callback && o->stream && o->stream->state == PA_STREAM_READY) {
1459 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1460 cb(o->stream, o->stream->timing_info_valid, o->userdata);
1461 }
1462
1463 finish:
1464
1465 pa_operation_done(o);
1466 pa_operation_unref(o);
1467 }
1468
1469 pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1470 uint32_t tag;
1471 pa_operation *o;
1472 pa_tagstruct *t;
1473 struct timeval now;
1474 int cidx = 0;
1475
1476 pa_assert(s);
1477 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1478
1479 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1480 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1481
1482 if (s->direction == PA_STREAM_PLAYBACK) {
1483 /* Find a place to store the write_index correction data for this entry */
1484 cidx = (s->current_write_index_correction + 1) % PA_MAX_WRITE_INDEX_CORRECTIONS;
1485
1486 /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
1487 PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
1488 }
1489 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1490
1491 t = pa_tagstruct_command(
1492 s->context,
1493 s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_GET_PLAYBACK_LATENCY : PA_COMMAND_GET_RECORD_LATENCY,
1494 &tag);
1495 pa_tagstruct_putu32(t, s->channel);
1496 pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
1497
1498 pa_pstream_send_tagstruct(s->context->pstream, t);
1499 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);
1500
1501 if (s->direction == PA_STREAM_PLAYBACK) {
1502 /* Fill in initial correction data */
1503
1504 s->current_write_index_correction = cidx;
1505
1506 s->write_index_corrections[cidx].valid = TRUE;
1507 s->write_index_corrections[cidx].absolute = FALSE;
1508 s->write_index_corrections[cidx].corrupt = FALSE;
1509 s->write_index_corrections[cidx].tag = tag;
1510 s->write_index_corrections[cidx].value = 0;
1511 }
1512
1513 return o;
1514 }
1515
1516 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1517 pa_stream *s = userdata;
1518
1519 pa_assert(pd);
1520 pa_assert(s);
1521 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1522
1523 pa_stream_ref(s);
1524
1525 if (command != PA_COMMAND_REPLY) {
1526 if (pa_context_handle_error(s->context, command, t, FALSE) < 0)
1527 goto finish;
1528
1529 pa_stream_set_state(s, PA_STREAM_FAILED);
1530 goto finish;
1531 } else if (!pa_tagstruct_eof(t)) {
1532 pa_context_fail(s->context, PA_ERR_PROTOCOL);
1533 goto finish;
1534 }
1535
1536 pa_stream_set_state(s, PA_STREAM_TERMINATED);
1537
1538 finish:
1539 pa_stream_unref(s);
1540 }
1541
1542 int pa_stream_disconnect(pa_stream *s) {
1543 pa_tagstruct *t;
1544 uint32_t tag;
1545
1546 pa_assert(s);
1547 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1548
1549 PA_CHECK_VALIDITY(s->context, s->channel_valid, PA_ERR_BADSTATE);
1550 PA_CHECK_VALIDITY(s->context, s->context->state == PA_CONTEXT_READY, PA_ERR_BADSTATE);
1551
1552 pa_stream_ref(s);
1553
1554 t = pa_tagstruct_command(
1555 s->context,
1556 s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_DELETE_PLAYBACK_STREAM :
1557 (s->direction == PA_STREAM_RECORD ? PA_COMMAND_DELETE_RECORD_STREAM : PA_COMMAND_DELETE_UPLOAD_STREAM),
1558 &tag);
1559 pa_tagstruct_putu32(t, s->channel);
1560 pa_pstream_send_tagstruct(s->context->pstream, t);
1561 pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
1562
1563 pa_stream_unref(s);
1564 return 0;
1565 }
1566
1567 void pa_stream_set_read_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
1568 pa_assert(s);
1569 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1570
1571 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1572 return;
1573
1574 s->read_callback = cb;
1575 s->read_userdata = userdata;
1576 }
1577
1578 void pa_stream_set_write_callback(pa_stream *s, pa_stream_request_cb_t cb, void *userdata) {
1579 pa_assert(s);
1580 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1581
1582 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1583 return;
1584
1585 s->write_callback = cb;
1586 s->write_userdata = userdata;
1587 }
1588
1589 void pa_stream_set_state_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1590 pa_assert(s);
1591 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1592
1593 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1594 return;
1595
1596 s->state_callback = cb;
1597 s->state_userdata = userdata;
1598 }
1599
1600 void pa_stream_set_overflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1601 pa_assert(s);
1602 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1603
1604 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1605 return;
1606
1607 s->overflow_callback = cb;
1608 s->overflow_userdata = userdata;
1609 }
1610
1611 void pa_stream_set_underflow_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1612 pa_assert(s);
1613 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1614
1615 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1616 return;
1617
1618 s->underflow_callback = cb;
1619 s->underflow_userdata = userdata;
1620 }
1621
1622 void pa_stream_set_latency_update_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1623 pa_assert(s);
1624 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1625
1626 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1627 return;
1628
1629 s->latency_update_callback = cb;
1630 s->latency_update_userdata = userdata;
1631 }
1632
1633 void pa_stream_set_moved_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1634 pa_assert(s);
1635 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1636
1637 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1638 return;
1639
1640 s->moved_callback = cb;
1641 s->moved_userdata = userdata;
1642 }
1643
1644 void pa_stream_set_suspended_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1645 pa_assert(s);
1646 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1647
1648 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1649 return;
1650
1651 s->suspended_callback = cb;
1652 s->suspended_userdata = userdata;
1653 }
1654
1655 void pa_stream_set_started_callback(pa_stream *s, pa_stream_notify_cb_t cb, void *userdata) {
1656 pa_assert(s);
1657 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1658
1659 if (s->state == PA_STREAM_TERMINATED || s->state == PA_STREAM_FAILED)
1660 return;
1661
1662 s->started_callback = cb;
1663 s->started_userdata = userdata;
1664 }
1665
1666 void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1667 pa_operation *o = userdata;
1668 int success = 1;
1669
1670 pa_assert(pd);
1671 pa_assert(o);
1672 pa_assert(PA_REFCNT_VALUE(o) >= 1);
1673
1674 if (!o->context)
1675 goto finish;
1676
1677 if (command != PA_COMMAND_REPLY) {
1678 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1679 goto finish;
1680
1681 success = 0;
1682 } else if (!pa_tagstruct_eof(t)) {
1683 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1684 goto finish;
1685 }
1686
1687 if (o->callback) {
1688 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
1689 cb(o->stream, success, o->userdata);
1690 }
1691
1692 finish:
1693 pa_operation_done(o);
1694 pa_operation_unref(o);
1695 }
1696
1697 pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, void *userdata) {
1698 pa_operation *o;
1699 pa_tagstruct *t;
1700 uint32_t tag;
1701
1702 pa_assert(s);
1703 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1704
1705 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1706 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1707
1708 s->corked = b;
1709
1710 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1711
1712 t = pa_tagstruct_command(
1713 s->context,
1714 s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_CORK_PLAYBACK_STREAM : PA_COMMAND_CORK_RECORD_STREAM,
1715 &tag);
1716 pa_tagstruct_putu32(t, s->channel);
1717 pa_tagstruct_put_boolean(t, !!b);
1718 pa_pstream_send_tagstruct(s->context->pstream, t);
1719 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);
1720
1721 check_smoother_status(s, FALSE, FALSE, FALSE);
1722
1723 if (s->direction == PA_STREAM_PLAYBACK)
1724 invalidate_indexes(s, TRUE, FALSE);
1725
1726 return o;
1727 }
1728
1729 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
1730 pa_tagstruct *t;
1731 pa_operation *o;
1732 uint32_t tag;
1733
1734 pa_assert(s);
1735 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1736
1737 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1738
1739 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1740
1741 t = pa_tagstruct_command(s->context, command, &tag);
1742 pa_tagstruct_putu32(t, s->channel);
1743 pa_pstream_send_tagstruct(s->context->pstream, t);
1744 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);
1745
1746 return o;
1747 }
1748
1749 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1750 pa_operation *o;
1751
1752 pa_assert(s);
1753 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1754
1755 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1756 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1757
1758 if ((o = stream_send_simple_command(s, s->direction == PA_STREAM_PLAYBACK ? PA_COMMAND_FLUSH_PLAYBACK_STREAM : PA_COMMAND_FLUSH_RECORD_STREAM, cb, userdata))) {
1759
1760 if (s->direction == PA_STREAM_PLAYBACK) {
1761 if (s->write_index_corrections[s->current_write_index_correction].valid)
1762 s->write_index_corrections[s->current_write_index_correction].corrupt = TRUE;
1763
1764 if (s->timing_info_valid)
1765 s->timing_info.write_index_corrupt = TRUE;
1766
1767 if (s->buffer_attr.prebuf > 0)
1768 invalidate_indexes(s, TRUE, FALSE);
1769 else
1770 request_auto_timing_update(s, TRUE);
1771
1772 if (s->smoother && s->buffer_attr.prebuf > 0) {
1773 pa_usec_t x = pa_rtclock_usec();
1774
1775 if (s->timing_info_valid)
1776 x += s->timing_info.transport_usec;
1777
1778 pa_smoother_pause(s->smoother, x);
1779 }
1780
1781 } else
1782 invalidate_indexes(s, FALSE, TRUE);
1783 }
1784
1785 return o;
1786 }
1787
1788 pa_operation* pa_stream_prebuf(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1789 pa_operation *o;
1790
1791 pa_assert(s);
1792 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1793
1794 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1795 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1796 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1797
1798 if ((o = stream_send_simple_command(s, PA_COMMAND_PREBUF_PLAYBACK_STREAM, cb, userdata)))
1799 invalidate_indexes(s, TRUE, FALSE);
1800
1801 return o;
1802 }
1803
1804 pa_operation* pa_stream_trigger(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
1805 pa_operation *o;
1806
1807 pa_assert(s);
1808 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1809
1810 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1811 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction == PA_STREAM_PLAYBACK, PA_ERR_BADSTATE);
1812 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->buffer_attr.prebuf > 0, PA_ERR_BADSTATE);
1813
1814 if ((o = stream_send_simple_command(s, PA_COMMAND_TRIGGER_PLAYBACK_STREAM, cb, userdata)))
1815 invalidate_indexes(s, TRUE, FALSE);
1816
1817 return o;
1818 }
1819
1820 pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_success_cb_t cb, void *userdata) {
1821 pa_operation *o;
1822
1823 pa_assert(s);
1824 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1825 pa_assert(name);
1826
1827 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1828 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1829
1830 if (s->context->version >= 13) {
1831 pa_proplist *p = pa_proplist_new();
1832
1833 pa_proplist_sets(p, PA_PROP_APPLICATION_NAME, name);
1834 o = pa_stream_proplist_update(s, PA_UPDATE_REPLACE, p, cb, userdata);
1835 pa_proplist_free(p);
1836 } else {
1837 pa_tagstruct *t;
1838 uint32_t tag;
1839
1840 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
1841 t = pa_tagstruct_command(
1842 s->context,
1843 s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_NAME : PA_COMMAND_SET_PLAYBACK_STREAM_NAME,
1844 &tag);
1845 pa_tagstruct_putu32(t, s->channel);
1846 pa_tagstruct_puts(t, name);
1847 pa_pstream_send_tagstruct(s->context->pstream, t);
1848 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);
1849 }
1850
1851 return o;
1852 }
1853
1854 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
1855 pa_usec_t usec;
1856
1857 pa_assert(s);
1858 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1859
1860 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1861 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1862 PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1863 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1864 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1865
1866 if (s->smoother)
1867 usec = pa_smoother_get(s->smoother, pa_rtclock_usec());
1868 else
1869 usec = calc_time(s, FALSE);
1870
1871 /* Make sure the time runs monotonically */
1872 if (!(s->flags & PA_STREAM_NOT_MONOTONOUS)) {
1873 if (usec < s->previous_time)
1874 usec = s->previous_time;
1875 else
1876 s->previous_time = usec;
1877 }
1878
1879 if (r_usec)
1880 *r_usec = usec;
1881
1882 return 0;
1883 }
1884
1885 static pa_usec_t time_counter_diff(pa_stream *s, pa_usec_t a, pa_usec_t b, int *negative) {
1886 pa_assert(s);
1887 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1888
1889 if (negative)
1890 *negative = 0;
1891
1892 if (a >= b)
1893 return a-b;
1894 else {
1895 if (negative && s->direction == PA_STREAM_RECORD) {
1896 *negative = 1;
1897 return b-a;
1898 } else
1899 return 0;
1900 }
1901 }
1902
1903 int pa_stream_get_latency(pa_stream *s, pa_usec_t *r_usec, int *negative) {
1904 pa_usec_t t, c;
1905 int r;
1906 int64_t cindex;
1907
1908 pa_assert(s);
1909 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1910 pa_assert(r_usec);
1911
1912 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1913 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1914 PA_CHECK_VALIDITY(s->context, s->timing_info_valid, PA_ERR_NODATA);
1915 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_PLAYBACK || !s->timing_info.write_index_corrupt, PA_ERR_NODATA);
1916 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_RECORD || !s->timing_info.read_index_corrupt, PA_ERR_NODATA);
1917
1918 if ((r = pa_stream_get_time(s, &t)) < 0)
1919 return r;
1920
1921 if (s->direction == PA_STREAM_PLAYBACK)
1922 cindex = s->timing_info.write_index;
1923 else
1924 cindex = s->timing_info.read_index;
1925
1926 if (cindex < 0)
1927 cindex = 0;
1928
1929 c = pa_bytes_to_usec(cindex, &s->sample_spec);
1930
1931 if (s->direction == PA_STREAM_PLAYBACK)
1932 *r_usec = time_counter_diff(s, c, t, negative);
1933 else
1934 *r_usec = time_counter_diff(s, t, c, negative);
1935
1936 return 0;
1937 }
1938
1939 const pa_timing_info* pa_stream_get_timing_info(pa_stream *s) {
1940 pa_assert(s);
1941 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1942
1943 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1944 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1945 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->timing_info_valid, PA_ERR_BADSTATE);
1946
1947 return &s->timing_info;
1948 }
1949
1950 const pa_sample_spec* pa_stream_get_sample_spec(pa_stream *s) {
1951 pa_assert(s);
1952 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1953
1954 return &s->sample_spec;
1955 }
1956
1957 const pa_channel_map* pa_stream_get_channel_map(pa_stream *s) {
1958 pa_assert(s);
1959 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1960
1961 return &s->channel_map;
1962 }
1963
1964 const pa_buffer_attr* pa_stream_get_buffer_attr(pa_stream *s) {
1965 pa_assert(s);
1966 pa_assert(PA_REFCNT_VALUE(s) >= 1);
1967
1968 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
1969 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
1970 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 9, PA_ERR_NOTSUPPORTED);
1971
1972 return &s->buffer_attr;
1973 }
1974
1975 static void stream_set_buffer_attr_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
1976 pa_operation *o = userdata;
1977 int success = 1;
1978
1979 pa_assert(pd);
1980 pa_assert(o);
1981 pa_assert(PA_REFCNT_VALUE(o) >= 1);
1982
1983 if (!o->context)
1984 goto finish;
1985
1986 if (command != PA_COMMAND_REPLY) {
1987 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
1988 goto finish;
1989
1990 success = 0;
1991 } else {
1992
1993 if (o->stream->direction == PA_STREAM_PLAYBACK) {
1994 if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
1995 pa_tagstruct_getu32(t, &o->stream->buffer_attr.tlength) < 0 ||
1996 pa_tagstruct_getu32(t, &o->stream->buffer_attr.prebuf) < 0 ||
1997 pa_tagstruct_getu32(t, &o->stream->buffer_attr.minreq) < 0) {
1998 pa_context_fail(o->context, PA_ERR_PROTOCOL);
1999 goto finish;
2000 }
2001 } else if (o->stream->direction == PA_STREAM_RECORD) {
2002 if (pa_tagstruct_getu32(t, &o->stream->buffer_attr.maxlength) < 0 ||
2003 pa_tagstruct_getu32(t, &o->stream->buffer_attr.fragsize) < 0) {
2004 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2005 goto finish;
2006 }
2007 }
2008
2009 if (!pa_tagstruct_eof(t)) {
2010 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2011 goto finish;
2012 }
2013 }
2014
2015 if (o->callback) {
2016 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2017 cb(o->stream, success, o->userdata);
2018 }
2019
2020 finish:
2021 pa_operation_done(o);
2022 pa_operation_unref(o);
2023 }
2024
2025
2026 pa_operation* pa_stream_set_buffer_attr(pa_stream *s, const pa_buffer_attr *attr, pa_stream_success_cb_t cb, void *userdata) {
2027 pa_operation *o;
2028 pa_tagstruct *t;
2029 uint32_t tag;
2030
2031 pa_assert(s);
2032 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2033 pa_assert(attr);
2034
2035 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2036 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2037 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2038
2039 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2040
2041 t = pa_tagstruct_command(
2042 s->context,
2043 s->direction == PA_STREAM_RECORD ? PA_COMMAND_SET_RECORD_STREAM_BUFFER_ATTR : PA_COMMAND_SET_PLAYBACK_STREAM_BUFFER_ATTR,
2044 &tag);
2045 pa_tagstruct_putu32(t, s->channel);
2046
2047 pa_tagstruct_putu32(t, attr->maxlength);
2048
2049 if (s->direction == PA_STREAM_PLAYBACK)
2050 pa_tagstruct_put(
2051 t,
2052 PA_TAG_U32, attr->tlength,
2053 PA_TAG_U32, attr->prebuf,
2054 PA_TAG_U32, attr->minreq,
2055 PA_TAG_INVALID);
2056 else
2057 pa_tagstruct_putu32(t, attr->fragsize);
2058
2059 if (s->context->version >= 13)
2060 pa_tagstruct_put_boolean(t, !!(s->flags & PA_STREAM_ADJUST_LATENCY));
2061
2062 pa_pstream_send_tagstruct(s->context->pstream, t);
2063 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);
2064
2065 return o;
2066 }
2067
2068 uint32_t pa_stream_get_device_index(pa_stream *s) {
2069 pa_assert(s);
2070 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2071
2072 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2073 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2074 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2075 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->device_index != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2076
2077 return s->device_index;
2078 }
2079
2080 const char *pa_stream_get_device_name(pa_stream *s) {
2081 pa_assert(s);
2082 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2083
2084 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2085 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2086 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2087 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->device_name, PA_ERR_BADSTATE);
2088
2089 return s->device_name;
2090 }
2091
2092 int pa_stream_is_suspended(pa_stream *s) {
2093 pa_assert(s);
2094 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2095
2096 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2097 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2098 PA_CHECK_VALIDITY(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2099
2100 return s->suspended;
2101 }
2102
2103 int pa_stream_is_corked(pa_stream *s) {
2104 pa_assert(s);
2105 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2106
2107 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2108 PA_CHECK_VALIDITY(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2109
2110 return s->corked;
2111 }
2112
2113 static void stream_update_sample_rate_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
2114 pa_operation *o = userdata;
2115 int success = 1;
2116
2117 pa_assert(pd);
2118 pa_assert(o);
2119 pa_assert(PA_REFCNT_VALUE(o) >= 1);
2120
2121 if (!o->context)
2122 goto finish;
2123
2124 if (command != PA_COMMAND_REPLY) {
2125 if (pa_context_handle_error(o->context, command, t, FALSE) < 0)
2126 goto finish;
2127
2128 success = 0;
2129 } else {
2130
2131 if (!pa_tagstruct_eof(t)) {
2132 pa_context_fail(o->context, PA_ERR_PROTOCOL);
2133 goto finish;
2134 }
2135 }
2136
2137 o->stream->sample_spec.rate = PA_PTR_TO_UINT(o->private);
2138 pa_assert(pa_sample_spec_valid(&o->stream->sample_spec));
2139
2140 if (o->callback) {
2141 pa_stream_success_cb_t cb = (pa_stream_success_cb_t) o->callback;
2142 cb(o->stream, success, o->userdata);
2143 }
2144
2145 finish:
2146 pa_operation_done(o);
2147 pa_operation_unref(o);
2148 }
2149
2150
2151 pa_operation *pa_stream_update_sample_rate(pa_stream *s, uint32_t rate, pa_stream_success_cb_t cb, void *userdata) {
2152 pa_operation *o;
2153 pa_tagstruct *t;
2154 uint32_t tag;
2155
2156 pa_assert(s);
2157 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2158
2159 PA_CHECK_VALIDITY_RETURN_NULL(s->context, rate > 0 && rate <= PA_RATE_MAX, PA_ERR_INVALID);
2160 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2161 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2162 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->flags & PA_STREAM_VARIABLE_RATE, PA_ERR_BADSTATE);
2163 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 12, PA_ERR_NOTSUPPORTED);
2164
2165 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2166 o->private = PA_UINT_TO_PTR(rate);
2167
2168 t = pa_tagstruct_command(
2169 s->context,
2170 s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_SAMPLE_RATE : PA_COMMAND_UPDATE_PLAYBACK_STREAM_SAMPLE_RATE,
2171 &tag);
2172 pa_tagstruct_putu32(t, s->channel);
2173 pa_tagstruct_putu32(t, rate);
2174
2175 pa_pstream_send_tagstruct(s->context->pstream, t);
2176 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);
2177
2178 return o;
2179 }
2180
2181 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) {
2182 pa_operation *o;
2183 pa_tagstruct *t;
2184 uint32_t tag;
2185
2186 pa_assert(s);
2187 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2188
2189 PA_CHECK_VALIDITY_RETURN_NULL(s->context, mode == PA_UPDATE_SET || mode == PA_UPDATE_MERGE || mode == PA_UPDATE_REPLACE, PA_ERR_INVALID);
2190 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2191 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2192 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2193
2194 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2195
2196 t = pa_tagstruct_command(
2197 s->context,
2198 s->direction == PA_STREAM_RECORD ? PA_COMMAND_UPDATE_RECORD_STREAM_PROPLIST : PA_COMMAND_UPDATE_PLAYBACK_STREAM_PROPLIST,
2199 &tag);
2200 pa_tagstruct_putu32(t, s->channel);
2201 pa_tagstruct_putu32(t, (uint32_t) mode);
2202 pa_tagstruct_put_proplist(t, p);
2203
2204 pa_pstream_send_tagstruct(s->context->pstream, t);
2205 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);
2206
2207 /* Please note that we don't update s->proplist here, because we
2208 * don't export that field */
2209
2210 return o;
2211 }
2212
2213 pa_operation *pa_stream_proplist_remove(pa_stream *s, const char *const keys[], pa_stream_success_cb_t cb, void *userdata) {
2214 pa_operation *o;
2215 pa_tagstruct *t;
2216 uint32_t tag;
2217 const char * const*k;
2218
2219 pa_assert(s);
2220 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2221
2222 PA_CHECK_VALIDITY_RETURN_NULL(s->context, keys && keys[0], PA_ERR_INVALID);
2223 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->state == PA_STREAM_READY, PA_ERR_BADSTATE);
2224 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->direction != PA_STREAM_UPLOAD, PA_ERR_BADSTATE);
2225 PA_CHECK_VALIDITY_RETURN_NULL(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2226
2227 o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
2228
2229 t = pa_tagstruct_command(
2230 s->context,
2231 s->direction == PA_STREAM_RECORD ? PA_COMMAND_REMOVE_RECORD_STREAM_PROPLIST : PA_COMMAND_REMOVE_PLAYBACK_STREAM_PROPLIST,
2232 &tag);
2233 pa_tagstruct_putu32(t, s->channel);
2234
2235 for (k = keys; *k; k++)
2236 pa_tagstruct_puts(t, *k);
2237
2238 pa_tagstruct_puts(t, NULL);
2239
2240 pa_pstream_send_tagstruct(s->context->pstream, t);
2241 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);
2242
2243 /* Please note that we don't update s->proplist here, because we
2244 * don't export that field */
2245
2246 return o;
2247 }
2248
2249 int pa_stream_set_monitor_stream(pa_stream *s, uint32_t sink_input_idx) {
2250 pa_assert(s);
2251 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2252
2253 PA_CHECK_VALIDITY(s->context, sink_input_idx != PA_INVALID_INDEX, PA_ERR_INVALID);
2254 PA_CHECK_VALIDITY(s->context, s->state == PA_STREAM_UNCONNECTED, PA_ERR_BADSTATE);
2255 PA_CHECK_VALIDITY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED);
2256
2257 s->direct_on_input = sink_input_idx;
2258
2259 return 0;
2260 }
2261
2262 uint32_t pa_stream_get_monitor_stream(pa_stream *s) {
2263 pa_assert(s);
2264 pa_assert(PA_REFCNT_VALUE(s) >= 1);
2265
2266 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->direct_on_input != PA_INVALID_INDEX, PA_ERR_BADSTATE, PA_INVALID_INDEX);
2267 PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->context->version >= 13, PA_ERR_NOTSUPPORTED, PA_INVALID_INDEX);
2268
2269 return s->direct_on_input;
2270 }