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