X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~mdw/git/disorder/blobdiff_plain/23205f9cdd229343a83d8c5a6317d9187e90c3d4..af66d051899a3ac7d37095807ff14069f0815285:/clients/playrtp.c diff --git a/clients/playrtp.c b/clients/playrtp.c index 8931b6e..9757be6 100644 --- a/clients/playrtp.c +++ b/clients/playrtp.c @@ -34,11 +34,12 @@ * * The main thread is responsible for actually playing audio. In ALSA this * means it waits until ALSA says it's ready for more audio which it then - * plays. + * plays. See @ref clients/playrtp-alsa.c. * - * InCore Audio the main thread is only responsible for starting and stopping + * In Core Audio the main thread is only responsible for starting and stopping * play: the system does the actual playback in its own private thread, and - * calls adioproc() to fetch the audio data. + * calls adioproc() to fetch the audio data. See @ref + * clients/playrtp-coreaudio.c. * * Sometimes it happens that there is no audio available to play. This may * because the server went away, or a packet was dropped, or the server @@ -62,6 +63,12 @@ #include #include #include +#include +#include +#include +#include +#include +#include #include "log.h" #include "mem.h" @@ -73,16 +80,17 @@ #include "vector.h" #include "heap.h" #include "timeval.h" - -#if HAVE_COREAUDIO_AUDIOHARDWARE_H -# include -#endif -#if API_ALSA -#include -#endif +#include "client.h" +#include "playrtp.h" +#include "inputline.h" #define readahead linux_headers_are_borked +/** @brief Obsolete synonym */ +#ifndef IPV6_JOIN_GROUP +# define IPV6_JOIN_GROUP IPV6_ADD_MEMBERSHIP +#endif + /** @brief RTP socket */ static int rtpfd; @@ -90,18 +98,12 @@ static int rtpfd; static FILE *logfp; /** @brief Output device */ -static const char *device; - -/** @brief Maximum samples per packet we'll support - * - * NB that two channels = two samples in this program. - */ -#define MAXSAMPLES 2048 +const char *device; /** @brief Minimum low watermark * * We'll stop playing if there's only this many samples in the buffer. */ -static unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */ +unsigned minbuffer = 2 * 44100 / 10; /* 0.2 seconds */ /** @brief Buffer high watermark * @@ -113,79 +115,6 @@ static unsigned readahead = 2 * 2 * 44100; * We'll stop reading from the network if we have this many samples. */ static unsigned maxbuffer; -/** @brief Number of samples to infill by in one go - * - * This is an upper bound - in practice we expect the underlying audio API to - * only ask for a much smaller number of samples in any one go. - */ -#define INFILL_SAMPLES (44100 * 2) /* 1s */ - -/** @brief Received packet - * - * Received packets are kept in a binary heap (see @ref pheap) ordered by - * timestamp. - */ -struct packet { - /** @brief Next packet in @ref next_free_packet or @ref received_packets */ - struct packet *next; - - /** @brief Number of samples in this packet */ - uint32_t nsamples; - - /** @brief Timestamp from RTP packet - * - * NB that "timestamps" are really sample counters. Use lt() or lt_packet() - * to compare timestamps. - */ - uint32_t timestamp; - - /** @brief Flags - * - * Valid values are: - * - @ref IDLE - the idle bit was set in the RTP packet - */ - unsigned flags; -/** @brief idle bit set in RTP packet*/ -#define IDLE 0x0001 - - /** @brief Raw sample data - * - * Only the first @p nsamples samples are defined; the rest is uninitialized - * data. - */ - uint16_t samples_raw[MAXSAMPLES]; -}; - -/** @brief Return true iff \f$a < b\f$ in sequence-space arithmetic - * - * Specifically it returns true if \f$(a-b) mod 2^{32} < 2^{31}\f$. - * - * See also lt_packet(). - */ -static inline int lt(uint32_t a, uint32_t b) { - return (uint32_t)(a - b) & 0x80000000; -} - -/** @brief Return true iff a >= b in sequence-space arithmetic */ -static inline int ge(uint32_t a, uint32_t b) { - return !lt(a, b); -} - -/** @brief Return true iff a > b in sequence-space arithmetic */ -static inline int gt(uint32_t a, uint32_t b) { - return lt(b, a); -} - -/** @brief Return true iff a <= b in sequence-space arithmetic */ -static inline int le(uint32_t a, uint32_t b) { - return !lt(b, a); -} - -/** @brief Ordering for packets, used by @ref pheap */ -static inline int lt_packet(const struct packet *a, const struct packet *b) { - return lt(a->timestamp, b->timestamp); -} - /** @brief Received packets * Protected by @ref receive_lock * @@ -193,94 +122,73 @@ static inline int lt_packet(const struct packet *a, const struct packet *b) { * it and adds them to @ref packets. Whenever a packet is added to it, @ref * receive_cond is signalled. */ -static struct packet *received_packets; +struct packet *received_packets; /** @brief Tail of @ref received_packets * Protected by @ref receive_lock */ -static struct packet **received_tail = &received_packets; +struct packet **received_tail = &received_packets; /** @brief Lock protecting @ref received_packets * * Only listen_thread() and queue_thread() ever hold this lock. It is vital * that queue_thread() not hold it any longer than it strictly has to. */ -static pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER; +pthread_mutex_t receive_lock = PTHREAD_MUTEX_INITIALIZER; /** @brief Condition variable signalled when @ref received_packets is updated * * Used by listen_thread() to notify queue_thread() that it has added another * packet to @ref received_packets. */ -static pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER; +pthread_cond_t receive_cond = PTHREAD_COND_INITIALIZER; /** @brief Length of @ref received_packets */ -static uint32_t nreceived; - -/** @struct pheap - * @brief Binary heap of packets ordered by timestamp */ -HEAP_TYPE(pheap, struct packet *, lt_packet); +uint32_t nreceived; /** @brief Binary heap of received packets */ -static struct pheap packets; +struct pheap packets; /** @brief Total number of samples available * * We make this volatile because we inspect it without a protecting lock, * so the usual pthread_* guarantees aren't available. */ -static volatile uint32_t nsamples; +volatile uint32_t nsamples; /** @brief Timestamp of next packet to play. * * This is set to the timestamp of the last packet, plus the number of * samples it contained. Only valid if @ref active is nonzero. */ -static uint32_t next_timestamp; +uint32_t next_timestamp; /** @brief True if actively playing * * This is true when playing and false when just buffering. */ -static int active; +int active; /** @brief Lock protecting @ref packets */ -static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; +pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; /** @brief Condition variable signalled whenever @ref packets is changed */ -static pthread_cond_t cond = PTHREAD_COND_INITIALIZER; +pthread_cond_t cond = PTHREAD_COND_INITIALIZER; -/** @brief Structure of free packet list */ -union free_packet { - struct packet p; - union free_packet *next; -}; - -/** @brief Linked list of free packets - * - * This is a linked list of formerly used packets. For preference we re-use - * packets that have already been used rather than unused ones, to limit the - * size of the program's working set. If there are no free packets in the list - * we try @ref next_free_packet instead. - * - * Must hold @ref lock when accessing this. - */ -static union free_packet *free_packets; +#if HAVE_ALSA_ASOUNDLIB_H +# define DEFAULT_BACKEND playrtp_alsa +#elif HAVE_SYS_SOUNDCARD_H || EMPEG_HOST +# define DEFAULT_BACKEND playrtp_oss +#elif HAVE_COREAUDIO_AUDIOHARDWARE_H +# define DEFAULT_BACKEND playrtp_coreaudio +#else +# error No known backend +#endif -/** @brief Array of new free packets - * - * There are @ref count_free_packets ready to use at this address. If there - * are none left we allocate more memory. - * - * Must hold @ref lock when accessing this. - */ -static union free_packet *next_free_packet; +/** @brief Backend to play with */ +static void (*backend)(void) = &DEFAULT_BACKEND; -/** @brief Count of new free packets at @ref next_free_packet - * - * Must hold @ref lock when accessing this. - */ -static size_t count_free_packets; +HEAP_DEFINE(pheap, struct packet *, lt_packet); -/** @brief Lock protecting packet allocator */ -static pthread_mutex_t mem_lock = PTHREAD_MUTEX_INITIALIZER; +/** @brief Control socket or NULL */ +const char *control_socket; static const struct option options[] = { { "help", no_argument, 0, 'h' }, @@ -291,37 +199,84 @@ static const struct option options[] = { { "max", required_argument, 0, 'x' }, { "buffer", required_argument, 0, 'b' }, { "rcvbuf", required_argument, 0, 'R' }, - { "multicast", required_argument, 0, 'M' }, +#if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST + { "oss", no_argument, 0, 'o' }, +#endif +#if HAVE_ALSA_ASOUNDLIB_H + { "alsa", no_argument, 0, 'a' }, +#endif +#if HAVE_COREAUDIO_AUDIOHARDWARE_H + { "core-audio", no_argument, 0, 'c' }, +#endif + { "socket", required_argument, 0, 's' }, + { "config", required_argument, 0, 'C' }, { 0, 0, 0, 0 } }; -/** @brief Return a new packet */ -static struct packet *new_packet(void) { - struct packet *p; - - pthread_mutex_lock(&mem_lock); - if(free_packets) { - p = &free_packets->p; - free_packets = free_packets->next; - } else { - if(!count_free_packets) { - next_free_packet = xcalloc(1024, sizeof (union free_packet)); - count_free_packets = 1024; +/** @brief Control thread + * + * This thread is responsible for accepting control commands from Disobedience + * (or other controllers) over an AF_UNIX stream socket with a path specified + * by the @c --socket option. The protocol uses simple string commands and + * replies: + * + * - @c stop will shut the player down + * - @c query will send back the reply @c running + * - anything else is ignored + * + * Commands and response strings terminated by shutting down the connection or + * by a newline. No attempt is made to multiplex multiple clients so it is + * important that the command be sent as soon as the connection is made - it is + * assumed that both parties to the protocol are entirely cooperating with one + * another. + */ +static void *control_thread(void attribute((unused)) *arg) { + struct sockaddr_un sa; + int sfd, cfd; + char *line; + socklen_t salen; + FILE *fp; + + assert(control_socket); + unlink(control_socket); + memset(&sa, 0, sizeof sa); + sa.sun_family = AF_UNIX; + strcpy(sa.sun_path, control_socket); + sfd = xsocket(PF_UNIX, SOCK_STREAM, 0); + if(bind(sfd, (const struct sockaddr *)&sa, sizeof sa) < 0) + fatal(errno, "error binding to %s", control_socket); + if(listen(sfd, 128) < 0) + fatal(errno, "error calling listen on %s", control_socket); + info("listening on %s", control_socket); + for(;;) { + salen = sizeof sa; + cfd = accept(sfd, (struct sockaddr *)&sa, &salen); + if(cfd < 0) { + switch(errno) { + case EINTR: + case EAGAIN: + break; + default: + fatal(errno, "error calling accept on %s", control_socket); + } + } + if(!(fp = fdopen(cfd, "r+"))) { + error(errno, "error calling fdopen for %s connection", control_socket); + close(cfd); + continue; + } + if(!inputline(control_socket, fp, &line, '\n')) { + if(!strcmp(line, "stop")) { + info("stopped via %s", control_socket); + exit(0); /* terminate immediately */ + } + if(!strcmp(line, "query")) + fprintf(fp, "running"); + xfree(line); } - p = &(next_free_packet++)->p; - --count_free_packets; + if(fclose(fp) < 0) + error(errno, "error closing %s connection", control_socket); } - pthread_mutex_unlock(&mem_lock); - return p; -} - -/** @brief Free a packet */ -static void free_packet(struct packet *p) { - union free_packet *u = (union free_packet *)p; - pthread_mutex_lock(&mem_lock); - u->next = free_packets; - free_packets = u; - pthread_mutex_unlock(&mem_lock); } /** @brief Drop the first packet @@ -332,7 +287,7 @@ static void drop_first_packet(void) { if(pheap_count(&packets)) { struct packet *const p = pheap_remove(&packets); nsamples -= p->nsamples; - free_packet(p); + playrtp_free_packet(p); pthread_cond_broadcast(&cond); } } @@ -386,7 +341,7 @@ static void *queue_thread(void attribute((unused)) *arg) { * thread which reads packets off the list and adds them to the heap. * * We keep memory allocation (mostly) very fast by keeping pre-allocated - * packets around; see @ref new_packet(). + * packets around; see @ref playrtp_new_packet(). */ static void *listen_thread(void attribute((unused)) *arg) { struct packet *p = 0; @@ -398,7 +353,7 @@ static void *listen_thread(void attribute((unused)) *arg) { for(;;) { if(!p) - p = new_packet(); + p = playrtp_new_packet(); iov[0].iov_base = &header; iov[0].iov_len = sizeof header; iov[1].iov_base = p->samples_raw; @@ -465,23 +420,11 @@ static void *listen_thread(void attribute((unused)) *arg) { } } -/** @brief Return true if @p p contains @p timestamp - * - * Containment implies that a sample @p timestamp exists within the packet. - */ -static inline int contains(const struct packet *p, uint32_t timestamp) { - const uint32_t packet_start = p->timestamp; - const uint32_t packet_end = p->timestamp + p->nsamples; - - return (ge(timestamp, packet_start) - && lt(timestamp, packet_end)); -} - /** @brief Wait until the buffer is adequately full * * Must be called with @ref lock held. */ -static void fill_buffer(void) { +void playrtp_fill_buffer(void) { while(nsamples) drop_first_packet(); info("Buffering..."); @@ -500,7 +443,7 @@ static void fill_buffer(void) { * * Must be called with @ref lock held. */ -static struct packet *next_packet(void) { +struct packet *playrtp_next_packet(void) { while(pheap_count(&packets)) { struct packet *const p = pheap_first(&packets); if(le(p->timestamp + p->nsamples, next_timestamp)) { @@ -514,214 +457,6 @@ static struct packet *next_packet(void) { return 0; } -#if HAVE_COREAUDIO_AUDIOHARDWARE_H -/** @brief Callback from Core Audio */ -static OSStatus adioproc - (AudioDeviceID attribute((unused)) inDevice, - const AudioTimeStamp attribute((unused)) *inNow, - const AudioBufferList attribute((unused)) *inInputData, - const AudioTimeStamp attribute((unused)) *inInputTime, - AudioBufferList *outOutputData, - const AudioTimeStamp attribute((unused)) *inOutputTime, - void attribute((unused)) *inClientData) { - UInt32 nbuffers = outOutputData->mNumberBuffers; - AudioBuffer *ab = outOutputData->mBuffers; - uint32_t samples_available; - - pthread_mutex_lock(&lock); - while(nbuffers > 0) { - float *samplesOut = ab->mData; - size_t samplesOutLeft = ab->mDataByteSize / sizeof (float); - - while(samplesOutLeft > 0) { - const struct packet *p = next_packet(); - if(p && contains(p, next_timestamp)) { - /* This packet is ready to play */ - const uint32_t packet_end = p->timestamp + p->nsamples; - const uint32_t offset = next_timestamp - p->timestamp; - const uint16_t *ptr = (void *)(p->samples_raw + offset); - - samples_available = packet_end - next_timestamp; - if(samples_available > samplesOutLeft) - samples_available = samplesOutLeft; - next_timestamp += samples_available; - samplesOutLeft -= samples_available; - while(samples_available-- > 0) - *samplesOut++ = (int16_t)ntohs(*ptr++) * (0.5 / 32767); - /* We don't bother junking the packet - that'll be dealt with next time - * round */ - } else { - /* No packet is ready to play (and there might be no packet at all) */ - samples_available = p ? p->timestamp - next_timestamp - : samplesOutLeft; - if(samples_available > samplesOutLeft) - samples_available = samplesOutLeft; - //info("infill by %"PRIu32, samples_available); - /* Conveniently the buffer is 0 to start with */ - next_timestamp += samples_available; - samplesOut += samples_available; - samplesOutLeft -= samples_available; - } - } - ++ab; - --nbuffers; - } - pthread_mutex_unlock(&lock); - return 0; -} -#endif - - -#if API_ALSA -/** @brief PCM handle */ -static snd_pcm_t *pcm; - -/** @brief True when @ref pcm is up and running */ -static int alsa_prepared = 1; - -/** @brief Initialize @ref pcm */ -static void setup_alsa(void) { - snd_pcm_hw_params_t *hwparams; - snd_pcm_sw_params_t *swparams; - /* Only support one format for now */ - const int sample_format = SND_PCM_FORMAT_S16_BE; - unsigned rate = 44100; - const int channels = 2; - const int samplesize = channels * sizeof(uint16_t); - snd_pcm_uframes_t pcm_bufsize = MAXSAMPLES * samplesize * 3; - /* If we can write more than this many samples we'll get a wakeup */ - const int avail_min = 256; - int err; - - /* Open ALSA */ - if((err = snd_pcm_open(&pcm, - device ? device : "default", - SND_PCM_STREAM_PLAYBACK, - SND_PCM_NONBLOCK))) - fatal(0, "error from snd_pcm_open: %d", err); - /* Set up 'hardware' parameters */ - snd_pcm_hw_params_alloca(&hwparams); - if((err = snd_pcm_hw_params_any(pcm, hwparams)) < 0) - fatal(0, "error from snd_pcm_hw_params_any: %d", err); - if((err = snd_pcm_hw_params_set_access(pcm, hwparams, - SND_PCM_ACCESS_RW_INTERLEAVED)) < 0) - fatal(0, "error from snd_pcm_hw_params_set_access: %d", err); - if((err = snd_pcm_hw_params_set_format(pcm, hwparams, - sample_format)) < 0) - - fatal(0, "error from snd_pcm_hw_params_set_format (%d): %d", - sample_format, err); - if((err = snd_pcm_hw_params_set_rate_near(pcm, hwparams, &rate, 0)) < 0) - fatal(0, "error from snd_pcm_hw_params_set_rate (%d): %d", - rate, err); - if((err = snd_pcm_hw_params_set_channels(pcm, hwparams, - channels)) < 0) - fatal(0, "error from snd_pcm_hw_params_set_channels (%d): %d", - channels, err); - if((err = snd_pcm_hw_params_set_buffer_size_near(pcm, hwparams, - &pcm_bufsize)) < 0) - fatal(0, "error from snd_pcm_hw_params_set_buffer_size (%d): %d", - MAXSAMPLES * samplesize * 3, err); - if((err = snd_pcm_hw_params(pcm, hwparams)) < 0) - fatal(0, "error calling snd_pcm_hw_params: %d", err); - /* Set up 'software' parameters */ - snd_pcm_sw_params_alloca(&swparams); - if((err = snd_pcm_sw_params_current(pcm, swparams)) < 0) - fatal(0, "error calling snd_pcm_sw_params_current: %d", err); - if((err = snd_pcm_sw_params_set_avail_min(pcm, swparams, avail_min)) < 0) - fatal(0, "error calling snd_pcm_sw_params_set_avail_min %d: %d", - avail_min, err); - if((err = snd_pcm_sw_params(pcm, swparams)) < 0) - fatal(0, "error calling snd_pcm_sw_params: %d", err); -} - -/** @brief Wait until ALSA wants some audio */ -static void wait_alsa(void) { - struct pollfd fds[64]; - int nfds, err; - unsigned short events; - - for(;;) { - do { - if((nfds = snd_pcm_poll_descriptors(pcm, - fds, sizeof fds / sizeof *fds)) < 0) - fatal(0, "error calling snd_pcm_poll_descriptors: %d", nfds); - } while(poll(fds, nfds, -1) < 0 && errno == EINTR); - if((err = snd_pcm_poll_descriptors_revents(pcm, fds, nfds, &events))) - fatal(0, "error calling snd_pcm_poll_descriptors_revents: %d", err); - if(events & POLLOUT) - return; - } -} - -/** @brief Play some sound via ALSA - * @param s Pointer to sample data - * @param n Number of samples - * @return 0 on success, -1 on non-fatal error - */ -static int alsa_writei(const void *s, size_t n) { - /* Do the write */ - const snd_pcm_sframes_t frames_written = snd_pcm_writei(pcm, s, n / 2); - if(frames_written < 0) { - /* Something went wrong */ - switch(frames_written) { - case -EAGAIN: - return 0; - case -EPIPE: - error(0, "error calling snd_pcm_writei: %ld", - (long)frames_written); - return -1; - default: - fatal(0, "error calling snd_pcm_writei: %ld", - (long)frames_written); - } - } else { - /* Success */ - next_timestamp += frames_written * 2; - return 0; - } -} - -/** @brief Play the relevant part of a packet - * @param p Packet to play - * @return 0 on success, -1 on non-fatal error - */ -static int alsa_play(const struct packet *p) { - return alsa_writei(p->samples_raw + next_timestamp - p->timestamp, - (p->timestamp + p->nsamples) - next_timestamp); -} - -/** @brief Play some silence - * @param p Next packet or NULL - * @return 0 on success, -1 on non-fatal error - */ -static int alsa_infill(const struct packet *p) { - static const uint16_t zeros[INFILL_SAMPLES]; - size_t samples_available = INFILL_SAMPLES; - - if(p && samples_available > p->timestamp - next_timestamp) - samples_available = p->timestamp - next_timestamp; - return alsa_writei(zeros, samples_available); -} - -/** @brief Reset ALSA state after we lost synchronization */ -static void alsa_reset(int hard_reset) { - int err; - - if((err = snd_pcm_nonblock(pcm, 0))) - fatal(0, "error calling snd_pcm_nonblock: %d", err); - if(hard_reset) { - if((err = snd_pcm_drop(pcm))) - fatal(0, "error calling snd_pcm_drop: %d", err); - } else - if((err = snd_pcm_drain(pcm))) - fatal(0, "error calling snd_pcm_drain: %d", err); - if((err = snd_pcm_nonblock(pcm, 1))) - fatal(0, "error calling snd_pcm_nonblock: %d", err); - alsa_prepared = 0; -} -#endif - /** @brief Play an RTP stream * * This is the guts of the program. It is responsible for: @@ -733,120 +468,16 @@ static void alsa_reset(int hard_reset) { */ static void play_rtp(void) { pthread_t ltid; + int err; /* We receive and convert audio data in a background thread */ - pthread_create(<id, 0, listen_thread, 0); + if((err = pthread_create(<id, 0, listen_thread, 0))) + fatal(err, "pthread_create listen_thread"); /* We have a second thread to add received packets to the queue */ - pthread_create(<id, 0, queue_thread, 0); -#if API_ALSA - { - struct packet *p; - int escape, err; - - /* Open the sound device */ - setup_alsa(); - pthread_mutex_lock(&lock); - for(;;) { - /* Wait for the buffer to fill up a bit */ - fill_buffer(); - if(!alsa_prepared) { - if((err = snd_pcm_prepare(pcm))) - fatal(0, "error calling snd_pcm_prepare: %d", err); - alsa_prepared = 1; - } - escape = 0; - info("Playing..."); - /* Keep playing until the buffer empties out, or ALSA tells us to get - * lost */ - while((nsamples >= minbuffer - || (nsamples > 0 - && contains(pheap_first(&packets), next_timestamp))) - && !escape) { - /* Wait for ALSA to ask us for more data */ - pthread_mutex_unlock(&lock); - wait_alsa(); - pthread_mutex_lock(&lock); - /* ALSA is ready for more data, find something to play */ - p = next_packet(); - /* Play it or play some silence */ - if(contains(p, next_timestamp)) - escape = alsa_play(p); - else - escape = alsa_infill(p); - } - active = 0; - /* We stop playing for a bit until the buffer re-fills */ - pthread_mutex_unlock(&lock); - alsa_reset(escape); - pthread_mutex_lock(&lock); - } - - } -#elif HAVE_COREAUDIO_AUDIOHARDWARE_H - { - OSStatus status; - UInt32 propertySize; - AudioDeviceID adid; - AudioStreamBasicDescription asbd; - - /* If this looks suspiciously like libao's macosx driver there's an - * excellent reason for that... */ - - /* TODO report errors as strings not numbers */ - propertySize = sizeof adid; - status = AudioHardwareGetProperty(kAudioHardwarePropertyDefaultOutputDevice, - &propertySize, &adid); - if(status) - fatal(0, "AudioHardwareGetProperty: %d", (int)status); - if(adid == kAudioDeviceUnknown) - fatal(0, "no output device"); - propertySize = sizeof asbd; - status = AudioDeviceGetProperty(adid, 0, false, - kAudioDevicePropertyStreamFormat, - &propertySize, &asbd); - if(status) - fatal(0, "AudioHardwareGetProperty: %d", (int)status); - D(("mSampleRate %f", asbd.mSampleRate)); - D(("mFormatID %08lx", asbd.mFormatID)); - D(("mFormatFlags %08lx", asbd.mFormatFlags)); - D(("mBytesPerPacket %08lx", asbd.mBytesPerPacket)); - D(("mFramesPerPacket %08lx", asbd.mFramesPerPacket)); - D(("mBytesPerFrame %08lx", asbd.mBytesPerFrame)); - D(("mChannelsPerFrame %08lx", asbd.mChannelsPerFrame)); - D(("mBitsPerChannel %08lx", asbd.mBitsPerChannel)); - D(("mReserved %08lx", asbd.mReserved)); - if(asbd.mFormatID != kAudioFormatLinearPCM) - fatal(0, "audio device does not support kAudioFormatLinearPCM"); - status = AudioDeviceAddIOProc(adid, adioproc, 0); - if(status) - fatal(0, "AudioDeviceAddIOProc: %d", (int)status); - pthread_mutex_lock(&lock); - for(;;) { - /* Wait for the buffer to fill up a bit */ - fill_buffer(); - /* Start playing now */ - info("Playing..."); - next_timestamp = pheap_first(&packets)->timestamp; - active = 1; - status = AudioDeviceStart(adid, adioproc); - if(status) - fatal(0, "AudioDeviceStart: %d", (int)status); - /* Wait until the buffer empties out */ - while(nsamples >= minbuffer - || (nsamples > 0 - && contains(pheap_first(&packets), next_timestamp))) - pthread_cond_wait(&cond, &lock); - /* Stop playing for a bit until the buffer re-fills */ - status = AudioDeviceStop(adid, adioproc); - if(status) - fatal(0, "AudioDeviceStop: %d", (int)status); - active = 0; - /* Go back round */ - } - } -#else -# error No known audio API -#endif + if((err = pthread_create(<id, 0, queue_thread, 0))) + fatal(err, "pthread_create queue_thread"); + /* The rest of the work is backend-specific */ + backend(); } /* display usage message and terminate */ @@ -859,7 +490,16 @@ static void help(void) { " --buffer, -b FRAMES Buffer high water mark\n" " --max, -x FRAMES Buffer maximum size\n" " --rcvbuf, -R BYTES Socket receive buffer size\n" - " --multicast, -M GROUP Join multicast group\n" + " --config, -C PATH Set configuration file\n" +#if HAVE_ALSA_ASOUNDLIB_H + " --alsa, -a Use ALSA to play audio\n" +#endif +#if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST + " --oss, -o Use OSS to play audio\n" +#endif +#if HAVE_COREAUDIO_AUDIOHARDWARE_H + " --core-audio, -c Use Core Audio to play audio\n" +#endif " --help, -h Display usage message\n" " --version, -V Display version number\n" ); @@ -875,15 +515,23 @@ static void version(void) { } int main(int argc, char **argv) { - int n; + int n, err; struct addrinfo *res; struct stringlist sl; char *sockname; int rcvbuf, target_rcvbuf = 131072; socklen_t len; - char *multicast_group = 0; struct ip_mreq mreq; struct ipv6_mreq mreq6; + disorder_client *c; + char *address, *port; + int is_multicast; + union any_sockaddr { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; + }; + union any_sockaddr mgroup; static const struct addrinfo prefs = { AI_PASSIVE, @@ -898,7 +546,7 @@ int main(int argc, char **argv) { mem_init(); if(!setlocale(LC_CTYPE, "")) fatal(errno, "error calling setlocale"); - while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:", options, 0)) >= 0) { + while((n = getopt_long(argc, argv, "hVdD:m:b:x:L:R:M:aocC:", options, 0)) >= 0) { switch(n) { case 'h': help(); case 'V': version(); @@ -909,40 +557,91 @@ int main(int argc, char **argv) { case 'x': maxbuffer = 2 * atol(optarg); break; case 'L': logfp = fopen(optarg, "w"); break; case 'R': target_rcvbuf = atoi(optarg); break; - case 'M': multicast_group = optarg; break; +#if HAVE_ALSA_ASOUNDLIB_H + case 'a': backend = playrtp_alsa; break; +#endif +#if HAVE_SYS_SOUNDCARD_H || EMPEG_HOST + case 'o': backend = playrtp_oss; break; +#endif +#if HAVE_COREAUDIO_AUDIOHARDWARE_H + case 'c': backend = playrtp_coreaudio; break; +#endif + case 'C': configfile = optarg; break; + case 's': control_socket = optarg; break; default: fatal(0, "invalid option"); } } + if(config_read(0)) fatal(0, "cannot read configuration"); if(!maxbuffer) maxbuffer = 4 * readahead; argc -= optind; argv += optind; - if(argc < 1 || argc > 2) - fatal(0, "usage: disorder-playrtp [OPTIONS] ADDRESS [PORT]"); - sl.n = argc; - sl.s = argv; - /* Listen for inbound audio data */ + switch(argc) { + case 0: + /* Get configuration from server */ + if(!(c = disorder_new(1))) exit(EXIT_FAILURE); + if(disorder_connect(c)) exit(EXIT_FAILURE); + if(disorder_rtp_address(c, &address, &port)) exit(EXIT_FAILURE); + sl.n = 2; + sl.s = xcalloc(2, sizeof *sl.s); + sl.s[0] = address; + sl.s[1] = port; + break; + case 1: + case 2: + /* Use command-line ADDRESS+PORT or just PORT */ + sl.n = argc; + sl.s = argv; + break; + default: + fatal(0, "usage: disorder-playrtp [OPTIONS] [[ADDRESS] PORT]"); + } + /* Look up address and port */ if(!(res = get_address(&sl, &prefs, &sockname))) exit(1); + /* Create the socket */ if((rtpfd = socket(res->ai_family, res->ai_socktype, res->ai_protocol)) < 0) fatal(errno, "error creating socket"); + /* Stash the multicast group address */ + if((is_multicast = multicast(res->ai_addr))) { + memcpy(&mgroup, res->ai_addr, res->ai_addrlen); + switch(res->ai_addr->sa_family) { + case AF_INET: + mgroup.in.sin_port = 0; + break; + case AF_INET6: + mgroup.in6.sin6_port = 0; + break; + } + } + /* Bind to 0/port */ + switch(res->ai_addr->sa_family) { + case AF_INET: + memset(&((struct sockaddr_in *)res->ai_addr)->sin_addr, 0, + sizeof (struct in_addr)); + break; + case AF_INET6: + memset(&((struct sockaddr_in6 *)res->ai_addr)->sin6_addr, 0, + sizeof (struct in6_addr)); + break; + default: + fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family); + } if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0) fatal(errno, "error binding socket to %s", sockname); - if(multicast_group) { - if((n = getaddrinfo(multicast_group, 0, &prefs, &res))) - fatal(0, "getaddrinfo %s: %s", multicast_group, gai_strerror(n)); - switch(res->ai_family) { + if(is_multicast) { + switch(mgroup.sa.sa_family) { case PF_INET: - mreq.imr_multiaddr = ((struct sockaddr_in *)res->ai_addr)->sin_addr; + mreq.imr_multiaddr = mgroup.in.sin_addr; mreq.imr_interface.s_addr = 0; /* use primary interface */ if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof mreq) < 0) fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP"); break; case PF_INET6: - mreq6.ipv6mr_multiaddr = ((struct sockaddr_in6 *)res->ai_addr)->sin6_addr; + mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr; memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface); if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq6, sizeof mreq6) < 0) @@ -951,7 +650,10 @@ int main(int argc, char **argv) { default: fatal(0, "unsupported address family %d", res->ai_family); } - } + info("listening on %s multicast group %s", + format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa)); + } else + info("listening on %s", format_sockaddr(res->ai_addr)); len = sizeof rcvbuf; if(getsockopt(rtpfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, &len) < 0) fatal(errno, "error calling getsockopt SO_RCVBUF"); @@ -968,6 +670,12 @@ int main(int argc, char **argv) { info("default socket receive buffer %d", rcvbuf); if(logfp) info("WARNING: -L option can impact performance"); + if(control_socket) { + pthread_t tid; + + if((err = pthread_create(&tid, 0, control_thread, 0))) + fatal(err, "pthread_create control_thread"); + } play_rtp(); return 0; }