]> code.delx.au - pulseaudio/blob - src/pulse/stream.c
delx.au / code / pulseaudio / blob


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