4 This file is part of PulseAudio.
6 PulseAudio is free software; you can redistribute it and/or modify
7 it under the terms of the GNU Lesser General Public License as
8 published by the Free Software Foundation; either version 2.1 of the
9 License, or (at your option) any later version.
11 PulseAudio is distributed in the hope that it will be useful, but
12 WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 Lesser General Public License for more details.
16 You should have received a copy of the GNU Lesser General Public
17 License along with PulseAudio; if not, write to the Free Software
18 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
31 #ifdef HAVE_SYS_SOCKET_H
32 #include <sys/socket.h>
37 #ifdef HAVE_NETINET_IN_H
38 #include <netinet/in.h>
43 #include <pulse/xmalloc.h>
45 #include <pulsecore/queue.h>
46 #include <pulsecore/log.h>
47 #include <pulsecore/core-scache.h>
48 #include <pulsecore/creds.h>
49 #include <pulsecore/mutex.h>
50 #include <pulsecore/refcnt.h>
54 /* We piggyback information if audio data blocks are stored in SHM on the seek mode */
55 #define PA_FLAG_SHMDATA 0x80000000LU
56 #define PA_FLAG_SHMRELEASE 0x40000000LU
57 #define PA_FLAG_SHMREVOKE 0xC0000000LU
58 #define PA_FLAG_SHMMASK 0xFF000000LU
59 #define PA_FLAG_SEEKMASK 0x000000FFLU
61 /* The sequence descriptor header consists of 5 32bit integers: */
63 PA_PSTREAM_DESCRIPTOR_LENGTH,
64 PA_PSTREAM_DESCRIPTOR_CHANNEL,
65 PA_PSTREAM_DESCRIPTOR_OFFSET_HI,
66 PA_PSTREAM_DESCRIPTOR_OFFSET_LO,
67 PA_PSTREAM_DESCRIPTOR_FLAGS,
68 PA_PSTREAM_DESCRIPTOR_MAX
71 /* If we have an SHM block, this info follows the descriptor */
73 PA_PSTREAM_SHM_BLOCKID,
76 PA_PSTREAM_SHM_LENGTH,
80 typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX];
82 #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t))
83 #define FRAME_SIZE_MAX_ALLOW PA_SCACHE_ENTRY_SIZE_MAX /* allow uploading a single sample in one frame at max */
84 #define FRAME_SIZE_MAX_USE (1024*64)
88 PA_PSTREAM_ITEM_PACKET,
89 PA_PSTREAM_ITEM_MEMBLOCK,
90 PA_PSTREAM_ITEM_SHMRELEASE,
91 PA_PSTREAM_ITEM_SHMREVOKE
106 pa_seek_mode_t seek_mode;
108 /* release/revoke info */
115 pa_mainloop_api *mainloop;
116 pa_defer_event *defer_event;
118 pa_queue *send_queue;
124 pa_pstream_descriptor descriptor;
125 struct item_info* current;
126 uint32_t shm_info[PA_PSTREAM_SHM_MAX];
132 pa_pstream_descriptor descriptor;
133 pa_memblock *memblock;
135 uint32_t shm_info[PA_PSTREAM_SHM_MAX];
141 pa_memimport *import;
142 pa_memexport *export;
144 pa_pstream_packet_cb_t recieve_packet_callback;
145 void *recieve_packet_callback_userdata;
147 pa_pstream_memblock_cb_t recieve_memblock_callback;
148 void *recieve_memblock_callback_userdata;
150 pa_pstream_notify_cb_t drain_callback;
151 void *drain_callback_userdata;
153 pa_pstream_notify_cb_t die_callback;
154 void *die_callback_userdata;
159 pa_creds read_creds, write_creds;
160 int read_creds_valid, send_creds_now;
164 static int do_write(pa_pstream *p);
165 static int do_read(pa_pstream *p);
167 static void do_something(pa_pstream *p) {
169 assert(PA_REFCNT_VALUE(p) > 0);
173 pa_mutex_lock(p->mutex);
175 p->mainloop->defer_enable(p->defer_event, 0);
177 if (!p->dead && pa_iochannel_is_readable(p->io)) {
180 } else if (!p->dead && pa_iochannel_is_hungup(p->io))
183 if (!p->dead && pa_iochannel_is_writable(p->io)) {
188 pa_mutex_unlock(p->mutex);
198 p->die_callback(p, p->die_callback_userdata);
200 pa_mutex_unlock(p->mutex);
205 static void io_callback(pa_iochannel*io, void *userdata) {
206 pa_pstream *p = userdata;
214 static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
215 pa_pstream *p = userdata;
218 assert(p->defer_event == e);
219 assert(p->mainloop == m);
224 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata);
226 pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *pool) {
233 p = pa_xnew(pa_pstream, 1);
236 pa_iochannel_set_callback(io, io_callback, p);
239 p->mutex = pa_mutex_new(1);
242 p->defer_event = m->defer_new(m, defer_callback, p);
243 m->defer_enable(p->defer_event, 0);
245 p->send_queue = pa_queue_new();
246 assert(p->send_queue);
248 p->write.current = NULL;
250 p->read.memblock = NULL;
251 p->read.packet = NULL;
254 p->recieve_packet_callback = NULL;
255 p->recieve_packet_callback_userdata = NULL;
256 p->recieve_memblock_callback = NULL;
257 p->recieve_memblock_callback_userdata = NULL;
258 p->drain_callback = NULL;
259 p->drain_callback_userdata = NULL;
260 p->die_callback = NULL;
261 p->die_callback_userdata = NULL;
268 /* We do importing unconditionally */
269 p->import = pa_memimport_new(p->mempool, memimport_release_cb, p);
271 pa_iochannel_socket_set_rcvbuf(io, 1024*8);
272 pa_iochannel_socket_set_sndbuf(io, 1024*8);
275 p->send_creds_now = 0;
276 p->read_creds_valid = 0;
281 static void item_free(void *item, PA_GCC_UNUSED void *p) {
282 struct item_info *i = item;
285 if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) {
286 assert(i->chunk.memblock);
287 pa_memblock_unref(i->chunk.memblock);
288 } else if (i->type == PA_PSTREAM_ITEM_PACKET) {
290 pa_packet_unref(i->packet);
296 static void pstream_free(pa_pstream *p) {
301 pa_queue_free(p->send_queue, item_free, NULL);
303 if (p->write.current)
304 item_free(p->write.current, NULL);
306 if (p->read.memblock)
307 pa_memblock_unref(p->read.memblock);
310 pa_packet_unref(p->read.packet);
313 pa_mutex_free(p->mutex);
318 void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet, const pa_creds *creds) {
322 assert(PA_REFCNT_VALUE(p) > 0);
325 pa_mutex_lock(p->mutex);
330 i = pa_xnew(struct item_info, 1);
331 i->type = PA_PSTREAM_ITEM_PACKET;
332 i->packet = pa_packet_ref(packet);
335 if ((i->with_creds = !!creds))
339 pa_queue_push(p->send_queue, i);
340 p->mainloop->defer_enable(p->defer_event, 1);
344 pa_mutex_unlock(p->mutex);
347 void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) {
351 assert(PA_REFCNT_VALUE(p) > 0);
352 assert(channel != (uint32_t) -1);
355 pa_mutex_lock(p->mutex);
360 length = chunk->length;
367 i = pa_xnew(struct item_info, 1);
368 i->type = PA_PSTREAM_ITEM_MEMBLOCK;
370 n = length < FRAME_SIZE_MAX_USE ? length : FRAME_SIZE_MAX_USE;
371 i->chunk.index = chunk->index + idx;
373 i->chunk.memblock = pa_memblock_ref(chunk->memblock);
375 i->channel = channel;
377 i->seek_mode = seek_mode;
382 pa_queue_push(p->send_queue, i);
388 p->mainloop->defer_enable(p->defer_event, 1);
392 pa_mutex_unlock(p->mutex);
395 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata) {
396 struct item_info *item;
397 pa_pstream *p = userdata;
400 assert(PA_REFCNT_VALUE(p) > 0);
402 pa_mutex_lock(p->mutex);
407 /* pa_log("Releasing block %u", block_id); */
409 item = pa_xnew(struct item_info, 1);
410 item->type = PA_PSTREAM_ITEM_SHMRELEASE;
411 item->block_id = block_id;
413 item->with_creds = 0;
416 pa_queue_push(p->send_queue, item);
417 p->mainloop->defer_enable(p->defer_event, 1);
421 pa_mutex_unlock(p->mutex);
424 static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userdata) {
425 struct item_info *item;
426 pa_pstream *p = userdata;
429 assert(PA_REFCNT_VALUE(p) > 0);
431 pa_mutex_lock(p->mutex);
436 /* pa_log("Revoking block %u", block_id); */
438 item = pa_xnew(struct item_info, 1);
439 item->type = PA_PSTREAM_ITEM_SHMREVOKE;
440 item->block_id = block_id;
442 item->with_creds = 0;
445 pa_queue_push(p->send_queue, item);
446 p->mainloop->defer_enable(p->defer_event, 1);
450 pa_mutex_unlock(p->mutex);
453 static void prepare_next_write_item(pa_pstream *p) {
455 assert(PA_REFCNT_VALUE(p) > 0);
457 if (!(p->write.current = pa_queue_pop(p->send_queue)))
461 p->write.data = NULL;
463 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = 0;
464 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
465 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = 0;
466 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
467 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = 0;
469 if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
471 assert(p->write.current->packet);
472 p->write.data = p->write.current->packet->data;
473 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->packet->length);
475 } else if (p->write.current->type == PA_PSTREAM_ITEM_SHMRELEASE) {
477 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(PA_FLAG_SHMRELEASE);
478 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl(p->write.current->block_id);
480 } else if (p->write.current->type == PA_PSTREAM_ITEM_SHMREVOKE) {
482 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(PA_FLAG_SHMREVOKE);
483 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl(p->write.current->block_id);
487 int send_payload = 1;
489 assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK);
490 assert(p->write.current->chunk.memblock);
492 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel);
493 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl((uint32_t) (((uint64_t) p->write.current->offset) >> 32));
494 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = htonl((uint32_t) ((uint64_t) p->write.current->offset));
496 flags = p->write.current->seek_mode & PA_FLAG_SEEKMASK;
499 uint32_t block_id, shm_id;
500 size_t offset, length;
504 if (pa_memexport_put(p->export,
505 p->write.current->chunk.memblock,
511 flags |= PA_FLAG_SHMDATA;
514 p->write.shm_info[PA_PSTREAM_SHM_BLOCKID] = htonl(block_id);
515 p->write.shm_info[PA_PSTREAM_SHM_SHMID] = htonl(shm_id);
516 p->write.shm_info[PA_PSTREAM_SHM_INDEX] = htonl((uint32_t) (offset + p->write.current->chunk.index));
517 p->write.shm_info[PA_PSTREAM_SHM_LENGTH] = htonl((uint32_t) p->write.current->chunk.length);
519 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(sizeof(p->write.shm_info));
520 p->write.data = p->write.shm_info;
523 /* pa_log_warn("Failed to export memory block."); */
527 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(p->write.current->chunk.length);
528 p->write.data = (uint8_t*) p->write.current->chunk.memblock->data + p->write.current->chunk.index;
531 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(flags);
535 if ((p->send_creds_now = p->write.current->with_creds))
536 p->write_creds = p->write.current->creds;
540 static int do_write(pa_pstream *p) {
546 assert(PA_REFCNT_VALUE(p) > 0);
548 if (!p->write.current)
549 prepare_next_write_item(p);
551 if (!p->write.current)
554 if (p->write.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
555 d = (uint8_t*) p->write.descriptor + p->write.index;
556 l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
558 assert(p->write.data);
560 d = (uint8_t*) p->write.data + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
561 l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
567 if (p->send_creds_now) {
569 if ((r = pa_iochannel_write_with_creds(p->io, d, l, &p->write_creds)) < 0)
572 p->send_creds_now = 0;
576 if ((r = pa_iochannel_write(p->io, d, l)) < 0)
581 if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE + ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
582 assert(p->write.current);
583 item_free(p->write.current, (void *) 1);
584 p->write.current = NULL;
586 if (p->drain_callback && !pa_pstream_is_pending(p))
587 p->drain_callback(p, p->drain_callback_userdata);
593 static int do_read(pa_pstream *p) {
599 assert(PA_REFCNT_VALUE(p) > 0);
601 if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
602 d = (uint8_t*) p->read.descriptor + p->read.index;
603 l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
605 assert(p->read.data);
606 d = (uint8_t*) p->read.data + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
607 l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
614 if ((r = pa_iochannel_read_with_creds(p->io, d, l, &p->read_creds, &b)) <= 0)
617 p->read_creds_valid = p->read_creds_valid || b;
620 if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
626 if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
627 uint32_t flags, length, channel;
628 /* Reading of frame descriptor complete */
630 flags = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]);
632 if (!p->import && (flags & PA_FLAG_SHMMASK) != 0) {
633 pa_log_warn("Recieved SHM frame on a socket where SHM is disabled.");
637 if (flags == PA_FLAG_SHMRELEASE) {
639 /* This is a SHM memblock release frame with no payload */
641 /* pa_log("Got release frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */
644 pa_memexport_process_release(p->export, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI]));
648 } else if (flags == PA_FLAG_SHMREVOKE) {
650 /* This is a SHM memblock revoke frame with no payload */
652 /* pa_log("Got revoke frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */
655 pa_memimport_process_revoke(p->import, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI]));
660 length = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]);
662 if (length > FRAME_SIZE_MAX_ALLOW) {
663 pa_log_warn("Recieved invalid frame size : %lu", (unsigned long) length);
667 assert(!p->read.packet && !p->read.memblock);
669 channel = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]);
671 if (channel == (uint32_t) -1) {
674 pa_log_warn("Received packet frame with invalid flags value.");
678 /* Frame is a packet frame */
679 p->read.packet = pa_packet_new(length);
680 p->read.data = p->read.packet->data;
684 if ((flags & PA_FLAG_SEEKMASK) > PA_SEEK_RELATIVE_END) {
685 pa_log_warn("Received memblock frame with invalid seek mode.");
689 if ((flags & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA) {
691 if (length != sizeof(p->read.shm_info)) {
692 pa_log_warn("Recieved SHM memblock frame with Invalid frame length.");
696 /* Frame is a memblock frame referencing an SHM memblock */
697 p->read.data = p->read.shm_info;
699 } else if ((flags & PA_FLAG_SHMMASK) == 0) {
701 /* Frame is a memblock frame */
703 p->read.memblock = pa_memblock_new(p->mempool, length);
704 p->read.data = p->read.memblock->data;
707 pa_log_warn("Recieved memblock frame with invalid flags value.");
712 } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) {
713 /* Frame payload available */
715 if (p->read.memblock && p->recieve_memblock_callback) {
717 /* Is this memblock data? Than pass it to the user */
718 l = (p->read.index - r) < PA_PSTREAM_DESCRIPTOR_SIZE ? p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE : (size_t) r;
723 chunk.memblock = p->read.memblock;
724 chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l;
727 if (p->recieve_memblock_callback) {
731 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
732 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO]))));
734 p->recieve_memblock_callback(
736 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
738 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SEEKMASK,
740 p->recieve_memblock_callback_userdata);
743 /* Drop seek info for following callbacks */
744 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] =
745 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] =
746 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
751 if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) {
753 if (p->read.memblock) {
755 /* This was a memblock frame. We can unref the memblock now */
756 pa_memblock_unref(p->read.memblock);
758 } else if (p->read.packet) {
760 if (p->recieve_packet_callback)
762 p->recieve_packet_callback(p, p->read.packet, p->read_creds_valid ? &p->read_creds : NULL, p->recieve_packet_callback_userdata);
764 p->recieve_packet_callback(p, p->read.packet, NULL, p->recieve_packet_callback_userdata);
767 pa_packet_unref(p->read.packet);
771 assert((ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA);
775 if (!(b = pa_memimport_get(p->import,
776 ntohl(p->read.shm_info[PA_PSTREAM_SHM_BLOCKID]),
777 ntohl(p->read.shm_info[PA_PSTREAM_SHM_SHMID]),
778 ntohl(p->read.shm_info[PA_PSTREAM_SHM_INDEX]),
779 ntohl(p->read.shm_info[PA_PSTREAM_SHM_LENGTH])))) {
781 pa_log_warn("Failed to import memory block.");
785 if (p->recieve_memblock_callback) {
791 chunk.length = b->length;
794 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
795 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO]))));
797 p->recieve_memblock_callback(
799 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
801 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SEEKMASK,
803 p->recieve_memblock_callback_userdata);
806 pa_memblock_unref(b);
816 p->read.memblock = NULL;
817 p->read.packet = NULL;
821 p->read_creds_valid = 0;
827 void pa_pstream_set_die_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
829 assert(PA_REFCNT_VALUE(p) > 0);
831 pa_mutex_lock(p->mutex);
832 p->die_callback = cb;
833 p->die_callback_userdata = userdata;
834 pa_mutex_unlock(p->mutex);
837 void pa_pstream_set_drain_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
839 assert(PA_REFCNT_VALUE(p) > 0);
841 pa_mutex_lock(p->mutex);
842 p->drain_callback = cb;
843 p->drain_callback_userdata = userdata;
844 pa_mutex_unlock(p->mutex);
847 void pa_pstream_set_recieve_packet_callback(pa_pstream *p, pa_pstream_packet_cb_t cb, void *userdata) {
849 assert(PA_REFCNT_VALUE(p) > 0);
851 pa_mutex_lock(p->mutex);
852 p->recieve_packet_callback = cb;
853 p->recieve_packet_callback_userdata = userdata;
854 pa_mutex_unlock(p->mutex);
857 void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, pa_pstream_memblock_cb_t cb, void *userdata) {
859 assert(PA_REFCNT_VALUE(p) > 0);
861 pa_mutex_lock(p->mutex);
862 p->recieve_memblock_callback = cb;
863 p->recieve_memblock_callback_userdata = userdata;
864 pa_mutex_unlock(p->mutex);
867 int pa_pstream_is_pending(pa_pstream *p) {
871 assert(PA_REFCNT_VALUE(p) > 0);
873 pa_mutex_lock(p->mutex);
878 b = p->write.current || !pa_queue_is_empty(p->send_queue);
880 pa_mutex_unlock(p->mutex);
885 void pa_pstream_unref(pa_pstream*p) {
887 assert(PA_REFCNT_VALUE(p) > 0);
889 if (PA_REFCNT_DEC(p) <= 0)
893 pa_pstream* pa_pstream_ref(pa_pstream*p) {
895 assert(PA_REFCNT_VALUE(p) > 0);
901 void pa_pstream_close(pa_pstream *p) {
904 pa_mutex_lock(p->mutex);
909 pa_memimport_free(p->import);
914 pa_memexport_free(p->export);
919 pa_iochannel_free(p->io);
923 if (p->defer_event) {
924 p->mainloop->defer_free(p->defer_event);
925 p->defer_event = NULL;
928 p->die_callback = NULL;
929 p->drain_callback = NULL;
930 p->recieve_packet_callback = NULL;
931 p->recieve_memblock_callback = NULL;
933 pa_mutex_unlock(p->mutex);
936 void pa_pstream_use_shm(pa_pstream *p, int enable) {
938 assert(PA_REFCNT_VALUE(p) > 0);
940 pa_mutex_lock(p->mutex);
947 p->export = pa_memexport_new(p->mempool, memexport_revoke_cb, p);
952 pa_memexport_free(p->export);
957 pa_mutex_unlock(p->mutex);