From: Hajime Fujita <crisp.fujita@xxxxxxxxx> There are two versions in the RAOP protocol; one uses TCP and the other uses UDP. Current raop implementation only supports TCP version. This patch adds an initial UDP protocol support for RAOP. It is based on Martin Blanchard's work (http://repo.or.cz/w/pulseaudio-raopUDP.git/shortlog/refs/heads/raop) which is inspired by Christophe Fergeau's work (https://github.com/zx2c4/pulseaudio-raop2). Matrin's modifications were edited by Hajime Fujita, so that it would support both TCP and UDP protocol in a single module. Also this patch includes a fix that was found thanks to Matthias, who reported that his ALAC codec support fixed the issue. https://bugs.freedesktop.org/show_bug.cgi?id=42804#c30 --- src/modules/raop/module-raop-sink.c | 457 +++++++++++++-- src/modules/raop/raop_client.c | 1063 +++++++++++++++++++++++++++++++---- src/modules/raop/raop_client.h | 39 +- 3 files changed, 1400 insertions(+), 159 deletions(-) diff --git a/src/modules/raop/module-raop-sink.c b/src/modules/raop/module-raop-sink.c index 6fc3d94..1eadde2 100644 --- a/src/modules/raop/module-raop-sink.c +++ b/src/modules/raop/module-raop-sink.c @@ -66,12 +66,13 @@ PA_MODULE_USAGE( "sink_name=<name for the sink> " "sink_properties=<properties for the sink> " "server=<address> " + "protocol=<transport protocol> " + "encryption=<encryption type> " + "codec=<audio codec> " "format=<sample format> " "rate=<sample rate> " "channels=<number of channels>"); -#define DEFAULT_SINK_NAME "raop" - struct userdata { pa_core *core; pa_module *module; @@ -82,6 +83,8 @@ struct userdata { pa_rtpoll_item *rtpoll_item; pa_thread *thread; + pa_raop_protocol_t protocol; + pa_memchunk raw_memchunk; pa_memchunk encoded_memchunk; @@ -97,7 +100,6 @@ struct userdata { int32_t rate; pa_smoother *smoother; - int fd; int64_t offset; int64_t encoding_overhead; @@ -107,12 +109,26 @@ struct userdata { pa_raop_client *raop; size_t block_size; + + /* Members only for the TCP protocol */ + int tcp_fd; + + /* Members only for the UDP protocol */ + int udp_control_fd; + int udp_timing_fd; + + /* For UDP thread wakeup clock calculation */ + pa_usec_t udp_playback_start; + uint32_t udp_sent_packets; }; static const char* const valid_modargs[] = { "sink_name", "sink_properties", "server", + "protocol", + "encryption", + "codec", "format", "rate", "channels", @@ -120,23 +136,26 @@ static const char* const valid_modargs[] = { }; enum { - SINK_MESSAGE_PASS_SOCKET = PA_SINK_MESSAGE_MAX, - SINK_MESSAGE_RIP_SOCKET + SINK_MESSAGE_TCP_PASS_SOCKET = PA_SINK_MESSAGE_MAX, + SINK_MESSAGE_TCP_RIP_SOCKET, + SINK_MESSAGE_UDP_SETUP, + SINK_MESSAGE_UDP_RECORD, + SINK_MESSAGE_UDP_DISCONNECTED, }; /* Forward declarations: */ static void sink_set_volume_cb(pa_sink *); -static void on_connection(int fd, void *userdata) { +static void tcp_on_connection(int fd, void *userdata) { int so_sndbuf = 0; socklen_t sl = sizeof(int); struct userdata *u = userdata; pa_assert(u); - pa_assert(u->fd < 0); - u->fd = fd; + pa_assert(u->tcp_fd < 0); + u->tcp_fd = fd; - if (getsockopt(u->fd, SOL_SOCKET, SO_SNDBUF, &so_sndbuf, &sl) < 0) + if (getsockopt(u->tcp_fd, SOL_SOCKET, SO_SNDBUF, &so_sndbuf, &sl) < 0) pa_log_warn("getsockopt(SO_SNDBUF) failed: %s", pa_cstrerror(errno)); else { pa_log_debug("SO_SNDBUF is %zu.", (size_t) so_sndbuf); @@ -148,19 +167,28 @@ static void on_connection(int fd, void *userdata) { pa_log_debug("Connection authenticated, handing fd to IO thread..."); - pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_PASS_SOCKET, NULL, 0, NULL, NULL); + pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_TCP_PASS_SOCKET, NULL, 0, NULL, NULL); } -static void on_close(void*userdata) { +static void tcp_on_close(void*userdata) { struct userdata *u = userdata; pa_assert(u); pa_log_debug("Connection closed, informing IO thread..."); - pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_RIP_SOCKET, NULL, 0, NULL, NULL); + pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_TCP_RIP_SOCKET, NULL, 0, NULL, NULL); +} + +static pa_usec_t sink_get_latency(const struct userdata *u) { + pa_usec_t w, r; + + r = pa_smoother_get(u->smoother, pa_rtclock_now()); + w = pa_bytes_to_usec((u->offset - u->encoding_overhead + (u->encoded_memchunk.length / u->encoding_ratio)), &u->sink->sample_spec); + + return w > r ? w - r : 0; } -static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) { +static int tcp_sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) { struct userdata *u = PA_SINK(o)->userdata; switch (code) { @@ -175,8 +203,8 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse pa_smoother_pause(u->smoother, pa_rtclock_now()); /* Issue a FLUSH if we are connected. */ - if (u->fd >= 0) { - pa_raop_flush(u->raop); + if (u->tcp_fd >= 0) { + pa_raop_client_flush(u->raop); } break; @@ -188,10 +216,10 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse /* The connection can be closed when idle, so check to * see if we need to reestablish it. */ - if (u->fd < 0) - pa_raop_connect(u->raop); + if (u->tcp_fd < 0) + pa_raop_client_connect(u->raop); else - pa_raop_flush(u->raop); + pa_raop_client_flush(u->raop); } break; @@ -205,37 +233,32 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse break; case PA_SINK_MESSAGE_GET_LATENCY: { - pa_usec_t w, r; - - r = pa_smoother_get(u->smoother, pa_rtclock_now()); - w = pa_bytes_to_usec((u->offset - u->encoding_overhead + (u->encoded_memchunk.length / u->encoding_ratio)), &u->sink->sample_spec); - - *((pa_usec_t*) data) = w > r ? w - r : 0; + *((pa_usec_t*) data) = sink_get_latency(u); return 0; } - case SINK_MESSAGE_PASS_SOCKET: { + case SINK_MESSAGE_TCP_PASS_SOCKET: { struct pollfd *pollfd; pa_assert(!u->rtpoll_item); u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1); pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); - pollfd->fd = u->fd; + pollfd->fd = u->tcp_fd; pollfd->events = POLLOUT; /*pollfd->events = */pollfd->revents = 0; if (u->sink->thread_info.state == PA_SINK_SUSPENDED) { /* Our stream has been suspended so we just flush it... */ - pa_raop_flush(u->raop); + pa_raop_client_flush(u->raop); } return 0; } - case SINK_MESSAGE_RIP_SOCKET: { - if (u->fd >= 0) { - pa_close(u->fd); - u->fd = -1; + case SINK_MESSAGE_TCP_RIP_SOCKET: { + if (u->tcp_fd >= 0) { + pa_close(u->tcp_fd); + u->tcp_fd = -1; } else /* FIXME */ pa_log("We should not get to this state. Cannot rip socket if not connected."); @@ -260,10 +283,140 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse return pa_sink_process_msg(o, code, data, offset, chunk); } +static void udp_start_wakeup_clock(struct userdata *u) { + pa_usec_t now = pa_rtclock_now(); + + u->udp_playback_start = now; + u->udp_sent_packets = 0; + pa_rtpoll_set_timer_absolute(u->rtpoll, now); +} + +static pa_usec_t udp_next_wakeup_clock(struct userdata *u) { + pa_usec_t intvl = pa_bytes_to_usec(u->block_size * u->udp_sent_packets, + &u->sink->sample_spec); + /* FIXME: how long until (u->block_size * u->udp_sent_packets) wraps?? */ + + return u->udp_playback_start + intvl; +} + +static int udp_sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) { + struct userdata *u = PA_SINK(o)->userdata; + + switch (code) { + case PA_SINK_MESSAGE_SET_STATE: + switch ((pa_sink_state_t) PA_PTR_TO_UINT(data)) { + case PA_SINK_SUSPENDED: + pa_assert(PA_SINK_IS_OPENED(u->sink->thread_info.state)); + pa_log_debug("RAOP: SUSPENDED"); + pa_smoother_pause(u->smoother, pa_rtclock_now()); + + if (pa_raop_client_udp_can_stream(u->raop)) { + /* Issue a TEARDOWN if we are still connected. */ + pa_raop_client_teardown(u->raop); + } + + break; + + case PA_SINK_IDLE: + pa_log_debug("RAOP: IDLE"); + /* Issue a flush if we're comming from running state. */ + if (u->sink->thread_info.state == PA_SINK_RUNNING) { + pa_rtpoll_set_timer_disabled(u->rtpoll); + pa_raop_client_flush(u->raop); + } + + break; + + case PA_SINK_RUNNING: + pa_log_debug("RAOP: RUNNING"); + + pa_smoother_resume(u->smoother, pa_rtclock_now(), true); + + if (!pa_raop_client_udp_can_stream(u->raop)) { + /* Connecting will trigger a RECORD */ + pa_raop_client_connect(u->raop); + } + udp_start_wakeup_clock(u); + + break; + + case PA_SINK_UNLINKED: + case PA_SINK_INIT: + case PA_SINK_INVALID_STATE: + ; + } + + break; + + case PA_SINK_MESSAGE_GET_LATENCY: { + pa_usec_t r = 0; + + if (pa_raop_client_udp_can_stream(u->raop)) + r = sink_get_latency(u); + + *((pa_usec_t*) data) = r; + + return 0; + } + + case SINK_MESSAGE_UDP_SETUP: { + struct pollfd *pollfd; + + u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 2); + pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); + + pollfd->fd = u->udp_control_fd; + pollfd->events = POLLIN | POLLPRI; + pollfd->revents = 0; + pollfd++; + pollfd->fd = u->udp_timing_fd; + pollfd->events = POLLIN | POLLPRI; + pollfd->revents = 0; + + return 0; + } + + case SINK_MESSAGE_UDP_RECORD: { + udp_start_wakeup_clock(u); + + if (u->sink->thread_info.state == PA_SINK_SUSPENDED) { + /* Our stream has been suspended so we just flush it... */ + pa_rtpoll_set_timer_disabled(u->rtpoll); + pa_raop_client_flush(u->raop); + } + + return 0; + } + + case SINK_MESSAGE_UDP_DISCONNECTED: { + if (u->sink->thread_info.state == PA_SINK_SUSPENDED) { + pa_rtpoll_set_timer_disabled(u->rtpoll); + if (u->rtpoll_item) + pa_rtpoll_item_free(u->rtpoll_item); + u->rtpoll_item = NULL; + } else { + /* Question: is this valid here: or should we do some sort of: + * return pa_sink_process_msg(PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL); ?? */ + pa_module_unload_request(u->module, true); + } + + pa_close(u->udp_control_fd); + pa_close(u->udp_timing_fd); + + u->udp_control_fd = -1; + u->udp_timing_fd = -1; + + return 0; + } + } + + return pa_sink_process_msg(o, code, data, offset, chunk); +} + static void sink_set_volume_cb(pa_sink *s) { struct userdata *u = s->userdata; pa_cvolume hw; - pa_volume_t v; + pa_volume_t v, v_orig; char t[PA_CVOLUME_SNPRINT_VERBOSE_MAX]; pa_assert(u); @@ -277,11 +430,16 @@ static void sink_set_volume_cb(pa_sink *s) { * any variation in channel volumes in software. */ v = pa_cvolume_max(&s->real_volume); + v_orig = v; + v = pa_raop_client_adjust_volume(u->raop, v_orig); + + pa_log_debug("Volume adjusted: orig=%u adjusted=%u", v_orig, v); + /* Create a pa_cvolume version of our single value. */ pa_cvolume_set(&hw, s->sample_spec.channels, v); - /* Perform any software manipulation of the volume needed. */ - pa_sw_cvolume_divide(&s->soft_volume, &s->real_volume, &hw); + /* Set the real volume based on given original volume. */ + pa_cvolume_set(&s->real_volume, s->sample_spec.channels, v_orig); pa_log_debug("Requested volume: %s", pa_cvolume_snprint_verbose(t, sizeof(t), &s->real_volume, &s->channel_map, false)); pa_log_debug("Got hardware volume: %s", pa_cvolume_snprint_verbose(t, sizeof(t), &hw, &s->channel_map, false)); @@ -305,8 +463,50 @@ static void sink_set_mute_cb(pa_sink *s) { } } -static void thread_func(void *userdata) { +static void udp_setup_cb(int control_fd, int timing_fd, void *userdata) { + struct userdata *u = userdata; + + pa_assert(control_fd); + pa_assert(timing_fd); + pa_assert(u); + + u->udp_control_fd = control_fd; + u->udp_timing_fd = timing_fd; + + pa_log_debug("Connection authenticated, syncing with server..."); + + pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_SETUP, NULL, 0, NULL, NULL); +} + +static void udp_record_cb(void *userdata) { struct userdata *u = userdata; + + pa_assert(u); + + /* Set the initial volume. */ + sink_set_volume_cb(u->sink); + + pa_log_debug("Synchronization done, pushing job to IO thread..."); + + pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_RECORD, NULL, 0, NULL, NULL); +} + +static void udp_disconnected_cb(void *userdata) { + struct userdata *u = userdata; + + pa_assert(u); + + /* This callback function is called from both STATE_TEARDOWN and + STATE_DISCONNECTED in raop_client.c */ + + pa_assert(u); + + pa_log_debug("Connection closed, informing IO thread..."); + + pa_asyncmsgq_post(u->thread_mq.inq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UDP_DISCONNECTED, NULL, 0, NULL, NULL); +} + +static void tcp_thread_func(struct userdata *u) { int write_type = 0; pa_memchunk silence; uint32_t silence_overhead = 0; @@ -314,7 +514,7 @@ static void thread_func(void *userdata) { pa_assert(u); - pa_log_debug("Thread starting up"); + pa_log_debug("TCP thread starting up"); pa_thread_mq_install(&u->thread_mq); @@ -394,7 +594,7 @@ static void thread_func(void *userdata) { pa_assert(u->encoded_memchunk.length > 0); p = pa_memblock_acquire(u->encoded_memchunk.memblock); - l = pa_write(u->fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type); + l = pa_write(u->tcp_fd, (uint8_t*) p + u->encoded_memchunk.index, u->encoded_memchunk.length, &write_type); pa_memblock_release(u->encoded_memchunk.memblock); pa_assert(l != 0); @@ -443,7 +643,7 @@ static void thread_func(void *userdata) { #ifdef SIOCOUTQ { int l; - if (ioctl(u->fd, SIOCOUTQ, &l) >= 0 && l > 0) + if (ioctl(u->tcp_fd, SIOCOUTQ, &l) >= 0 && l > 0) n -= (l / u->encoding_ratio); } #endif @@ -497,15 +697,139 @@ fail: finish: if (silence.memblock) pa_memblock_unref(silence.memblock); - pa_log_debug("Thread shutting down"); + pa_log_debug("TCP thread shutting down"); +} + +static void udp_thread_func(struct userdata *u) { + pa_assert(u); + + pa_log_debug("UDP thread starting up"); + + pa_thread_mq_install(&u->thread_mq); + pa_smoother_set_time_offset(u->smoother, pa_rtclock_now()); + + for (;;) { + pa_usec_t estimated; + int32_t overhead = 0; + ssize_t written = 0; + size_t length = 0; + int rv = 0; + + if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) { + if (u->sink->thread_info.rewind_requested) + pa_sink_process_rewind(u->sink, 0); + } + + /* Polling (audio data + control socket + timing socket). */ + if ((rv = pa_rtpoll_run(u->rtpoll, true)) < 0) + goto fail; + else if (rv == 0) + goto finish; + + if (!pa_rtpoll_timer_elapsed(u->rtpoll)) { + struct pollfd *pollfd; + uint8_t packet[32]; + ssize_t read; + + if (u->rtpoll_item) { + pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL); + + /* Event on the control socket ?? */ + if (pollfd->revents & POLLIN) { + pollfd->revents = 0; + pa_log_debug("Received control packet."); + read = pa_read(pollfd->fd, packet, sizeof(packet), NULL); + pa_raop_client_udp_handle_control_packet(u->raop, packet, read); + } + + pollfd++; + + /* Event on the timing port ?? */ + if (pollfd->revents & POLLIN) { + pollfd->revents = 0; + pa_log_debug("Received timing packet."); + read = pa_read(pollfd->fd, packet, sizeof(packet), NULL); + pa_raop_client_udp_handle_timing_packet(u->raop, packet, read); + } + } + + continue; + } + + if (!pa_raop_client_udp_can_stream(u->raop)) + continue; + if (u->sink->thread_info.state != PA_SINK_RUNNING) + continue; + + if (u->encoded_memchunk.length <= 0) { + if (u->encoded_memchunk.memblock != NULL) + pa_memblock_unref(u->encoded_memchunk.memblock); + + if (u->raw_memchunk.length <= 0) { + if (u->raw_memchunk.memblock) + pa_memblock_unref(u->raw_memchunk.memblock); + pa_memchunk_reset(&u->raw_memchunk); + + /* Grab unencoded audio data from PulseAudio. */ + pa_sink_render_full(u->sink, u->block_size, &u->raw_memchunk); + } + + pa_assert(u->raw_memchunk.length > 0); + + length = u->raw_memchunk.length; + pa_raop_client_encode_sample(u->raop, &u->raw_memchunk, &u->encoded_memchunk); + u->encoding_ratio = (double) u->encoded_memchunk.length / (double) (length - u->raw_memchunk.length); + overhead = u->encoded_memchunk.length - (length - u->raw_memchunk.length); + } + + pa_assert(u->encoded_memchunk.length > 0); + + written = pa_raop_client_udp_send_audio_packet(u->raop,&u->encoded_memchunk); + if (written < 0) { + pa_log("Failed to send UDP packet: %s", pa_cstrerror(errno)); + goto fail; + } + + u->udp_sent_packets++; + /* Sleep until next packet transmission */ + pa_rtpoll_set_timer_absolute(u->rtpoll, udp_next_wakeup_clock(u)); + + u->offset += written; + u->encoding_overhead += overhead; + + estimated = pa_bytes_to_usec(u->offset - u->encoding_overhead, &u->sink->sample_spec); + pa_smoother_put(u->smoother, pa_rtclock_now(), estimated); + } + +fail: + /* If this was no regular exit, continue processing messages until PA_MESSAGE_SHUTDOWN. */ + pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL); + pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN); + +finish: + pa_log_debug("UDP thread shutting down"); +} + +static void thread_func(void *userdata) { + struct userdata *u = userdata; + + if (u->protocol == RAOP_TCP) + tcp_thread_func(u); + else if (u->protocol == RAOP_UDP) + udp_thread_func(u); + else + pa_assert(false); + + return; } int pa__init(pa_module *m) { struct userdata *u = NULL; pa_sample_spec ss; pa_modargs *ma = NULL; - const char *server; + const char *server, *protocol, *encryption; pa_sink_new_data data; + char *t = NULL; pa_assert(m); @@ -532,7 +856,7 @@ int pa__init(pa_module *m) { u->core = m->core; u->module = m; m->userdata = u; - u->fd = -1; + u->tcp_fd = -1; u->smoother = pa_smoother_new( PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, @@ -569,15 +893,32 @@ int pa__init(pa_module *m) { goto fail; } + /* This may be overwriten if sink_name is specified in module arguments. */ + t = pa_sprintf_malloc("raop_client.%s", server); + + protocol = pa_modargs_get_value(ma, "protocol", NULL); + if (protocol == NULL || pa_streq(protocol, "TCP")) { + /* Assume TCP by default */ + u->protocol = RAOP_TCP; + } + else if (pa_streq(protocol, "UDP")) { + u->protocol = RAOP_UDP; + } else { + pa_log("Unsupported protocol argument given: %s", protocol); + goto fail; + } + pa_sink_new_data_init(&data); data.driver = __FILE__; data.module = m; - pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME)); + pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", t)); pa_sink_new_data_set_sample_spec(&data, &ss); pa_proplist_sets(data.proplist, PA_PROP_DEVICE_STRING, server); pa_proplist_sets(data.proplist, PA_PROP_DEVICE_INTENDED_ROLES, "music"); pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "RAOP sink '%s'", server); + /* RAOP discover module will eventually overwrite sink_name and others + (PA_UPDATE_REPLACE). */ if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) { pa_log("Invalid properties"); pa_sink_new_data_done(&data); @@ -585,6 +926,7 @@ int pa__init(pa_module *m) { } u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_NETWORK); + pa_xfree(t); t = NULL; pa_sink_new_data_done(&data); if (!u->sink) { @@ -592,7 +934,10 @@ int pa__init(pa_module *m) { goto fail; } - u->sink->parent.process_msg = sink_process_msg; + if (u->protocol == RAOP_TCP) + u->sink->parent.process_msg = tcp_sink_process_msg; + else + u->sink->parent.process_msg = udp_sink_process_msg; u->sink->userdata = u; pa_sink_set_set_volume_callback(u->sink, sink_set_volume_cb); pa_sink_set_set_mute_callback(u->sink, sink_set_mute_cb); @@ -601,13 +946,27 @@ int pa__init(pa_module *m) { pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq); pa_sink_set_rtpoll(u->sink, u->rtpoll); - if (!(u->raop = pa_raop_client_new(u->core, server))) { + if (!(u->raop = pa_raop_client_new(u->core, server, u->protocol))) { pa_log("Failed to connect to server."); goto fail; } - pa_raop_client_set_callback(u->raop, on_connection, u); - pa_raop_client_set_closed_callback(u->raop, on_close, u); + encryption = pa_modargs_get_value(ma, "encryption", NULL); + pa_raop_client_set_encryption(u->raop, !pa_safe_streq(encryption, "none")); + + pa_raop_client_tcp_set_callback(u->raop, tcp_on_connection, u); + pa_raop_client_tcp_set_closed_callback(u->raop, tcp_on_close, u); + + if (u->protocol == RAOP_UDP) { + /* The number of frames per blocks is not negotiable... */ + pa_raop_client_udp_get_blocks_size(u->raop, &u->block_size); + u->block_size *= pa_frame_size(&ss); + pa_sink_set_max_request(u->sink, u->block_size); + + pa_raop_client_udp_set_setup_callback(u->raop, udp_setup_cb, u); + pa_raop_client_udp_set_record_callback(u->raop, udp_record_cb, u); + pa_raop_client_udp_set_disconnected_callback(u->raop, udp_disconnected_cb, u); + } if (!(u->thread = pa_thread_new("raop-sink", thread_func, u))) { pa_log("Failed to create thread."); @@ -621,6 +980,8 @@ int pa__init(pa_module *m) { return 0; fail: + pa_xfree(t); + if (ma) pa_modargs_free(ma); @@ -679,8 +1040,8 @@ void pa__done(pa_module *m) { if (u->smoother) pa_smoother_free(u->smoother); - if (u->fd >= 0) - pa_close(u->fd); + if (u->tcp_fd >= 0) + pa_close(u->tcp_fd); pa_xfree(u); } diff --git a/src/modules/raop/raop_client.c b/src/modules/raop/raop_client.c index a5dd29c..c0be2ec 100644 --- a/src/modules/raop/raop_client.c +++ b/src/modules/raop/raop_client.c @@ -26,6 +26,7 @@ #include <errno.h> #include <unistd.h> #include <sys/ioctl.h> +#include <math.h> #ifdef HAVE_SYS_FILIO_H #include <sys/filio.h> @@ -39,10 +40,14 @@ #include <openssl/engine.h> #include <pulse/xmalloc.h> +#include <pulse/timeval.h> +#include <pulse/sample.h> #include <pulsecore/core-error.h> +#include <pulsecore/core-rtclock.h> #include <pulsecore/core-util.h> #include <pulsecore/iochannel.h> +#include <pulsecore/arpa-inet.h> #include <pulsecore/socket-util.h> #include <pulsecore/log.h> #include <pulsecore/parseaddr.h> @@ -54,6 +59,7 @@ #include "rtsp_client.h" #include "base64.h" +#define UDP_FRAMES_PER_PACKET 352 #define AES_CHUNKSIZE 16 #define JACK_STATUS_DISCONNECTED 0 @@ -66,7 +72,19 @@ #define VOLUME_MIN -144 #define VOLUME_MAX 0 -#define RAOP_PORT 5000 +#define DEFAULT_RAOP_PORT 5000 +#define UDP_DEFAULT_AUDIO_PORT 6000 +#define UDP_DEFAULT_CONTROL_PORT 6001 +#define UDP_DEFAULT_TIMING_PORT 6002 + +typedef enum { + UDP_PAYLOAD_TIMING_REQUEST = 0x52, + UDP_PAYLOAD_TIMING_RESPONSE = 0x53, + UDP_PAYLOAD_SYNCHRONIZATION = 0x54, + UDP_PAYLOAD_RETRANSMIT_REQUEST = 0x55, + UDP_PAYLOAD_RETRANSMIT_REPLY = 0x56, + UDP_PAYLOAD_AUDIO_DATA = 0x60 +} pa_raop_udp_payload_type; struct pa_raop_client { pa_core *core; @@ -74,26 +92,93 @@ struct pa_raop_client { uint16_t port; char *sid; pa_rtsp_client *rtsp; + pa_raop_protocol_t protocol; uint8_t jack_type; uint8_t jack_status; /* Encryption Related bits */ + int encryption; /* Enable encryption? */ AES_KEY aes; uint8_t aes_iv[AES_CHUNKSIZE]; /* Initialization vector for aes-cbc */ uint8_t aes_nv[AES_CHUNKSIZE]; /* Next vector for aes-cbc */ uint8_t aes_key[AES_CHUNKSIZE]; /* Key for aes-cbc */ - pa_socket_client *sc; - int fd; - uint16_t seq; uint32_t rtptime; - pa_raop_client_cb_t callback; - void *userdata; - pa_raop_client_closed_cb_t closed_callback; - void *closed_userdata; + /* Members only for the TCP protocol */ + pa_socket_client *tcp_sc; + int tcp_fd; + + pa_raop_client_cb_t tcp_callback; + void *tcp_userdata; + pa_raop_client_closed_cb_t tcp_closed_callback; + void *tcp_closed_userdata; + + /* Members only for the UDP protocol */ + uint16_t udp_my_control_port; + uint16_t udp_my_timing_port; + uint16_t udp_server_control_port; + uint16_t udp_server_timing_port; + + int udp_stream_fd; + int udp_control_fd; + int udp_timing_fd; + + uint32_t udp_ssrc; + + bool udp_first_packet; + uint32_t udp_sync_interval; + uint32_t udp_sync_count; + + pa_raop_client_setup_cb_t udp_setup_callback; + void *udp_setup_userdata; + + pa_raop_client_record_cb_t udp_record_callback; + void *udp_record_userdata; + + pa_raop_client_disconnected_cb_t udp_disconnected_callback; + void *udp_disconnected_userdata; +}; + +/* Timming packet header (8x8): + * [0] RTP v2: 0x80, + * [1] Payload type: 0x53 | marker bit: 0x80, + * [2,3] Sequence number: 0x0007, + * [4,7] Timestamp: 0x00000000 (unused). */ +static const uint8_t udp_timming_header[8] = { + 0x80, 0xd3, 0x00, 0x07, + 0x00, 0x00, 0x00, 0x00 +}; + +/* Sync packet header (8x8): + * [0] RTP v2: 0x80, + * [1] Payload type: 0x54 | marker bit: 0x80, + * [2,3] Sequence number: 0x0007, + * [4,7] Timestamp: 0x00000000 (to be set). */ +static const uint8_t udp_sync_header[8] = { + 0x80, 0xd4, 0x00, 0x07, + 0x00, 0x00, 0x00, 0x00 +}; + +static const uint8_t tcp_audio_header[16] = { + 0x24, 0x00, 0x00, 0x00, + 0xF0, 0xFF, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, +}; + +/* Audio packet header (12x8): + * [0] RTP v2: 0x80, + * [1] Payload type: 0x60, + * [2,3] Sequence number: 0x0000 (to be set), + * [4,7] Timestamp: 0x00000000 (to be set), + * [8,12] SSRC: 0x00000000 (to be set).*/ +static const uint8_t udp_audio_header[12] = { + 0x80, 0x60, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00 }; /** @@ -200,84 +285,334 @@ static inline void rtrimchar(char *str, char rc) { } } -static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) { +static void tcp_on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) { pa_raop_client *c = userdata; pa_assert(sc); pa_assert(c); - pa_assert(c->sc == sc); - pa_assert(c->fd < 0); - pa_assert(c->callback); + pa_assert(c->tcp_sc == sc); + pa_assert(c->tcp_fd < 0); + pa_assert(c->tcp_callback); - pa_socket_client_unref(c->sc); - c->sc = NULL; + pa_socket_client_unref(c->tcp_sc); + c->tcp_sc = NULL; if (!io) { pa_log("Connection failed: %s", pa_cstrerror(errno)); return; } - c->fd = pa_iochannel_get_send_fd(io); + c->tcp_fd = pa_iochannel_get_send_fd(io); pa_iochannel_set_noclose(io, true); pa_iochannel_free(io); - pa_make_tcp_socket_low_delay(c->fd); + pa_make_tcp_socket_low_delay(c->tcp_fd); pa_log_debug("Connection established"); - c->callback(c->fd, c->userdata); + c->tcp_callback(c->tcp_fd, c->tcp_userdata); } -static void rtsp_cb(pa_rtsp_client *rtsp, pa_rtsp_state state, pa_headerlist *headers, void *userdata) { - pa_raop_client *c = userdata; +static inline uint64_t timeval_to_ntp(struct timeval *tv) { + uint64_t ntp = 0; + + /* Converting micro seconds to a fraction. */ + ntp = (uint64_t) tv->tv_usec * UINT32_MAX / PA_USEC_PER_SEC; + /* Moving reference from 1 Jan 1970 to 1 Jan 1900 (seconds). */ + ntp |= (uint64_t) (tv->tv_sec + 0x83aa7e80) << 32; + + return ntp; +} + +static int connect_udp_socket(pa_raop_client *c, int fd, uint16_t port) { + struct sockaddr_in sa4; +#ifdef HAVE_IPV6 + struct sockaddr_in6 sa6; +#endif + struct sockaddr *sa; + socklen_t salen; + sa_family_t af; + + pa_zero(sa4); +#ifdef HAVE_IPV6 + pa_zero(sa6); +#endif + if (inet_pton(AF_INET, c->host, &sa4.sin_addr) > 0) { + sa4.sin_family = af = AF_INET; + sa4.sin_port = htons(port); + sa = (struct sockaddr *) &sa4; + salen = sizeof(sa4); +#ifdef HAVE_IPV6 + } else if (inet_pton(AF_INET6, c->host, &sa6.sin6_addr) > 0) { + sa6.sin6_family = af = AF_INET6; + sa6.sin6_port = htons(port); + sa = (struct sockaddr *) &sa6; + salen = sizeof(sa6); +#endif + } else { + pa_log("Invalid destination '%s'", c->host); + goto fail; + } + + if (fd < 0 && (fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) { + pa_log("socket() failed: %s", pa_cstrerror(errno)); + goto fail; + } + + /* If the socket queue is full, let's drop packets */ + pa_make_udp_socket_low_delay(fd); + pa_make_fd_nonblock(fd); + + if (connect(fd, sa, salen) < 0) { + pa_log("connect() failed: %s", pa_cstrerror(errno)); + goto fail; + } + + pa_log_debug("Connected to %s on port %d (SOCK_DGRAM)", c->host, port); + return fd; + +fail: + if (fd >= 0) + pa_close(fd); + + return -1; +} + +static int open_bind_udp_socket(pa_raop_client *c, uint16_t *actual_port) { + int fd = -1; + uint16_t port; + struct sockaddr_in sa4; +#ifdef HAVE_IPV6 + struct sockaddr_in6 sa6; +#endif + struct sockaddr *sa; + uint16_t *sa_port; + socklen_t salen; + sa_family_t af; + int one = 1; + + pa_assert(actual_port); + + port = *actual_port; + + pa_zero(sa4); +#ifdef HAVE_IPV6 + pa_zero(sa6); +#endif + if (inet_pton(AF_INET, pa_rtsp_localip(c->rtsp), &sa4.sin_addr) > 0) { + sa4.sin_family = af = AF_INET; + sa4.sin_port = htons(port); + sa = (struct sockaddr *) &sa4; + salen = sizeof(sa4); + sa_port = &sa4.sin_port; +#ifdef HAVE_IPV6 + } else if (inet_pton(AF_INET6, pa_rtsp_localip(c->rtsp), &sa6.sin6_addr) > 0) { + sa6.sin6_family = af = AF_INET6; + sa6.sin6_port = htons(port); + sa = (struct sockaddr *) &sa6; + salen = sizeof(sa6); + sa_port = &sa6.sin6_port; +#endif + } else { + pa_log("Could not determine which address family to use"); + goto fail; + } + + pa_zero(sa4); +#ifdef HAVE_IPV6 + pa_zero(sa6); +#endif + + if ((fd = pa_socket_cloexec(af, SOCK_DGRAM, 0)) < 0) { + pa_log("socket() failed: %s", pa_cstrerror(errno)); + goto fail; + } + +#ifdef SO_TIMESTAMP + if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) { + pa_log("setsockopt(SO_TIMESTAMP) failed: %s", pa_cstrerror(errno)); + goto fail; + } +#else + pa_log("SO_TIMESTAMP unsupported on this platform"); + goto fail; +#endif + + one = 1; + if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) { + pa_log("setsockopt(SO_REUSEADDR) failed: %s", pa_cstrerror(errno)); + goto fail; + } + + do { + *sa_port = htons(port); + + if (bind(fd, sa, salen) < 0 && errno != EADDRINUSE) { + pa_log("bind_socket() failed: %s", pa_cstrerror(errno)); + goto fail; + } + break; + } while (++port > 0); + + pa_log_debug("Socket bound to port %d (SOCK_DGRAM)", port); + *actual_port = port; + + return fd; + +fail: + if (fd >= 0) + pa_close(fd); + + return -1; +} + +static int udp_send_timing_packet(pa_raop_client *c, const uint32_t data[6], uint64_t received) { + uint32_t packet[8]; + struct timeval tv; + ssize_t written = 0; + uint64_t trs = 0; + int rv = 1; + + memcpy(packet, udp_timming_header, sizeof(udp_timming_header)); + /* Copying originate timestamp from the incoming request packet. */ + packet[2] = data[4]; + packet[3] = data[5]; + /* Set the receive timestamp to reception time. */ + packet[4] = htonl(received >> 32); + packet[5] = htonl(received & 0xffffffff); + /* Set the transmit timestamp to current time. */ + trs = timeval_to_ntp(pa_rtclock_get(&tv)); + packet[6] = htonl(trs >> 32); + packet[7] = htonl(trs & 0xffffffff); + + written = pa_loop_write(c->udp_timing_fd, packet, sizeof(packet), NULL); + if (written == sizeof(packet)) + rv = 0; + + return rv; +} + +static int udp_send_sync_packet(pa_raop_client *c, uint32_t stamp) { + const uint32_t delay = 88200; + uint32_t packet[5]; + struct timeval tv; + ssize_t written = 0; + uint64_t trs = 0; + int rv = 1; + + memcpy(packet, udp_sync_header, sizeof(udp_sync_header)); + if (c->udp_first_packet) + packet[0] |= 0x10; + stamp -= delay; + packet[1] = htonl(stamp); + /* Set the transmited timestamp to current time. */ + trs = timeval_to_ntp(pa_rtclock_get(&tv)); + packet[2] = htonl(trs >> 32); + packet[3] = htonl(trs & 0xffffffff); + stamp += delay; + packet[4] = htonl(stamp); + + written = pa_loop_write(c->udp_control_fd, packet, sizeof(packet), NULL); + if (written == sizeof(packet)) + rv = 0; + + return rv; +} + +static void udp_build_audio_header(pa_raop_client *c, uint32_t *buffer, size_t size) { + pa_assert(size >= sizeof(udp_audio_header)); + + memcpy(buffer, udp_audio_header, sizeof(udp_audio_header)); + if (c->udp_first_packet) + buffer[0] |= htonl((uint32_t) 0x80 << 16); + buffer[0] |= htonl((uint32_t) c->seq); + buffer[1] = htonl(c->rtptime); + buffer[2] = htonl(c->udp_ssrc); +} + +static ssize_t udp_send_audio_packet(pa_raop_client *c, uint8_t *buffer, size_t size) { + ssize_t length; + + length = pa_write(c->udp_stream_fd, buffer, size, NULL); + c->seq++; + + return length; +} + +static void do_rtsp_announce(pa_raop_client *c) { + int i; + uint8_t rsakey[512]; + char *key, *iv, *sac = NULL, *sdp; + uint16_t rand_data; + const char *ip; + char *url; + + ip = pa_rtsp_localip(c->rtsp); + /* First of all set the url properly. */ + url = pa_sprintf_malloc("rtsp://%s/%s", ip, c->sid); + pa_rtsp_set_url(c->rtsp, url); + pa_xfree(url); + + /* Now encrypt our aes_public key to send to the device. */ + i = rsa_encrypt(c->aes_key, AES_CHUNKSIZE, rsakey); + pa_base64_encode(rsakey, i, &key); + rtrimchar(key, '='); + pa_base64_encode(c->aes_iv, AES_CHUNKSIZE, &iv); + rtrimchar(iv, '='); + + /* UDP protocol does not need "Apple-Challenge" at announce. */ + if (c->protocol == RAOP_TCP) { + pa_random(&rand_data, sizeof(rand_data)); + pa_base64_encode(&rand_data, AES_CHUNKSIZE, &sac); + rtrimchar(sac, '='); + pa_rtsp_add_header(c->rtsp, "Apple-Challenge", sac); + } + + if (c->encryption) + sdp = pa_sprintf_malloc( + "v=0\r\n" + "o=iTunes %s 0 IN IP4 %s\r\n" + "s=iTunes\r\n" + "c=IN IP4 %s\r\n" + "t=0 0\r\n" + "m=audio 0 RTP/AVP 96\r\n" + "a=rtpmap:96 AppleLossless\r\n" + "a=fmtp:96 %d 0 16 40 10 14 2 255 0 0 44100\r\n" + "a=rsaaeskey:%s\r\n" + "a=aesiv:%s\r\n", + c->sid, ip, c->host, + c->protocol == RAOP_TCP ? 4096 : UDP_FRAMES_PER_PACKET, + key, iv); + else + sdp = pa_sprintf_malloc( + "v=0\r\n" + "o=iTunes %s 0 IN IP4 %s\r\n" + "s=iTunes\r\n" + "c=IN IP4 %s\r\n" + "t=0 0\r\n" + "m=audio 0 RTP/AVP 96\r\n" + "a=rtpmap:96 AppleLossless\r\n" + "a=fmtp:96 %d 0 16 40 10 14 2 255 0 0 44100\r\n", + c->sid, ip, c->host, + c->protocol == RAOP_TCP ? 4096 : UDP_FRAMES_PER_PACKET); + + pa_rtsp_announce(c->rtsp, sdp); + pa_xfree(key); + pa_xfree(iv); + pa_xfree(sac); + pa_xfree(sdp); +} + +static void tcp_rtsp_cb(pa_rtsp_client *rtsp, pa_rtsp_state state, pa_headerlist* headers, void *userdata) { + pa_raop_client* c = userdata; pa_assert(c); pa_assert(rtsp); pa_assert(rtsp == c->rtsp); switch (state) { case STATE_CONNECT: { - int i; - uint8_t rsakey[512]; - char *key, *iv, *sac, *sdp; - uint16_t rand_data; - const char *ip; - char *url; - pa_log_debug("RAOP: CONNECTED"); - ip = pa_rtsp_localip(c->rtsp); - /* First of all set the url properly. */ - url = pa_sprintf_malloc("rtsp://%s/%s", ip, c->sid); - pa_rtsp_set_url(c->rtsp, url); - pa_xfree(url); - - /* Now encrypt our aes_public key to send to the device. */ - i = rsa_encrypt(c->aes_key, AES_CHUNKSIZE, rsakey); - pa_base64_encode(rsakey, i, &key); - rtrimchar(key, '='); - pa_base64_encode(c->aes_iv, AES_CHUNKSIZE, &iv); - rtrimchar(iv, '='); - - pa_random(&rand_data, sizeof(rand_data)); - pa_base64_encode(&rand_data, AES_CHUNKSIZE, &sac); - rtrimchar(sac, '='); - pa_rtsp_add_header(c->rtsp, "Apple-Challenge", sac); - sdp = pa_sprintf_malloc( - "v=0\r\n" - "o=iTunes %s 0 IN IP4 %s\r\n" - "s=iTunes\r\n" - "c=IN IP4 %s\r\n" - "t=0 0\r\n" - "m=audio 0 RTP/AVP 96\r\n" - "a=rtpmap:96 AppleLossless\r\n" - "a=fmtp:96 4096 0 16 40 10 14 2 255 0 0 44100\r\n" - "a=rsaaeskey:%s\r\n" - "a=aesiv:%s\r\n", - c->sid, ip, c->host, key, iv); - pa_rtsp_announce(c->rtsp, sdp); - pa_xfree(key); - pa_xfree(iv); - pa_xfree(sac); - pa_xfree(sdp); + do_rtsp_announce(c); break; } @@ -325,11 +660,11 @@ static void rtsp_cb(pa_rtsp_client *rtsp, pa_rtsp_state state, pa_headerlist *he uint32_t port = pa_rtsp_serverport(c->rtsp); pa_log_debug("RAOP: RECORDED"); - if (!(c->sc = pa_socket_client_new_string(c->core->mainloop, true, c->host, port))) { + if (!(c->tcp_sc = pa_socket_client_new_string(c->core->mainloop, true, c->host, port))) { pa_log("failed to connect to server '%s:%d'", c->host, port); return; } - pa_socket_client_set_callback(c->sc, on_connection, c); + pa_socket_client_set_callback(c->tcp_sc, tcp_on_connection, c); break; } @@ -346,30 +681,328 @@ static void rtsp_cb(pa_rtsp_client *rtsp, pa_rtsp_state state, pa_headerlist *he break; case STATE_DISCONNECTED: - pa_assert(c->closed_callback); + pa_assert(c->tcp_closed_callback); pa_assert(c->rtsp); pa_log_debug("RTSP control channel closed"); pa_rtsp_client_free(c->rtsp); c->rtsp = NULL; - if (c->fd > 0) { - /* We do not close the fd, we leave it to the closed callback to do that. */ - c->fd = -1; + if (c->tcp_fd > 0) { + /* We do not close the fd, we leave it to the closed callback to do that */ + c->tcp_fd = -1; } - if (c->sc) { - pa_socket_client_unref(c->sc); - c->sc = NULL; + if (c->tcp_sc) { + pa_socket_client_unref(c->tcp_sc); + c->tcp_sc = NULL; } pa_xfree(c->sid); c->sid = NULL; - c->closed_callback(c->closed_userdata); + c->tcp_closed_callback(c->tcp_closed_userdata); break; } } -pa_raop_client* pa_raop_client_new(pa_core *core, const char *host) { - pa_parsed_address a; +static void udp_rtsp_cb(pa_rtsp_client *rtsp, pa_rtsp_state state, pa_headerlist *headers, void *userdata) { + pa_raop_client *c = userdata; + + pa_assert(c); + pa_assert(rtsp); + pa_assert(rtsp == c->rtsp); + + switch (state) { + case STATE_CONNECT: { + uint16_t rand; + char *sac; + + /* Set the Apple-Challenge key */ + pa_random(&rand, sizeof(rand)); + pa_base64_encode(&rand, AES_CHUNKSIZE, &sac); + rtrimchar(sac, '='); + pa_rtsp_add_header(c->rtsp, "Apple-Challenge", sac); + + pa_rtsp_options(c->rtsp); + + pa_xfree(sac); + break; + } + + case STATE_OPTIONS: { + pa_log_debug("RAOP: OPTIONS"); + + pa_rtsp_remove_header(c->rtsp, "Apple-Challenge"); + do_rtsp_announce(c); + break; + } + + case STATE_ANNOUNCE: { + char *trs; + + pa_assert(c->udp_control_fd < 0); + pa_assert(c->udp_timing_fd < 0); + + c->udp_control_fd = open_bind_udp_socket(c, &c->udp_my_control_port); + if (c->udp_control_fd < 0) + goto error_announce; + c->udp_timing_fd = open_bind_udp_socket(c, &c->udp_my_timing_port); + if (c->udp_timing_fd < 0) + goto error_announce; + + trs = pa_sprintf_malloc("RTP/AVP/UDP;unicast;interleaved=0-1;mode=record;control_port=%d;timing_port=%d", + c->udp_my_control_port, + c->udp_my_timing_port); + + pa_rtsp_setup(c->rtsp, trs); + + pa_xfree(trs); + break; + + error_announce: + if (c->udp_control_fd > 0) { + pa_close(c->udp_control_fd); + c->udp_control_fd = -1; + } + if (c->udp_timing_fd > 0) { + pa_close(c->udp_timing_fd); + c->udp_timing_fd = -1; + } + + pa_rtsp_client_free(c->rtsp); + c->rtsp = NULL; + + c->udp_my_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_server_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_my_timing_port = UDP_DEFAULT_TIMING_PORT; + c->udp_server_timing_port = UDP_DEFAULT_TIMING_PORT; + + pa_log_error("aborting RTSP announce, failed creating required sockets"); + } + + case STATE_SETUP: { + uint32_t stream_port = UDP_DEFAULT_AUDIO_PORT; + char *ajs, *trs, *token, *pc; + char delimiters[] = ";"; + const char *token_state = NULL; + uint32_t port = 0; + int ret; + + pa_log_debug("RAOP: SETUP"); + + ajs = pa_xstrdup(pa_headerlist_gets(headers, "Audio-Jack-Status")); + trs = pa_xstrdup(pa_headerlist_gets(headers, "Transport")); + + if (ajs) { + c->jack_type = JACK_TYPE_ANALOG; + c->jack_status = JACK_STATUS_DISCONNECTED; + + while ((token = pa_split(ajs, delimiters, &token_state))) { + if ((pc = strstr(token, "="))) { + *pc = 0; + if (pa_streq(token, "type") && pa_streq(pc + 1, "digital")) + c->jack_type = JACK_TYPE_DIGITAL; + } else { + if (pa_streq(token, "connected")) + c->jack_status = JACK_STATUS_CONNECTED; + } + pa_xfree(token); + } + + } else { + pa_log_warn("Audio-Jack-Status missing"); + } + + token_state = NULL; + + if (trs) { + /* Now parse out the server port component of the response. */ + while ((token = pa_split(trs, delimiters, &token_state))) { + if ((pc = strstr(token, "="))) { + *pc = 0; + if (pa_streq(token, "control_port")) { + port = 0; + pa_atou(pc + 1, &port); + c->udp_server_control_port = port; + } + if (pa_streq(token, "timing_port")) { + port = 0; + pa_atou(pc + 1, &port); + c->udp_server_timing_port = port; + } + *pc = '='; + } + pa_xfree(token); + } + } else { + pa_log_warn("Transport missing"); + } + + pa_xfree(ajs); + pa_xfree(trs); + + stream_port = pa_rtsp_serverport(c->rtsp); + if (stream_port == 0) + goto error; + if (c->udp_server_control_port == 0 || c->udp_server_timing_port == 0) + goto error; + + pa_log_debug("Using server_port=%d, control_port=%d & timing_port=%d", + stream_port, + c->udp_server_control_port, + c->udp_server_timing_port); + + pa_assert(c->udp_stream_fd < 0); + pa_assert(c->udp_control_fd >= 0); + pa_assert(c->udp_timing_fd >= 0); + + c->udp_stream_fd = connect_udp_socket(c, -1, stream_port); + if (c->udp_stream_fd <= 0) + goto error; + ret = connect_udp_socket(c, c->udp_control_fd, + c->udp_server_control_port); + if (ret < 0) + goto error; + ret = connect_udp_socket(c, c->udp_timing_fd, + c->udp_server_timing_port); + if (ret < 0) + goto error; + + c->udp_setup_callback(c->udp_control_fd, c->udp_timing_fd, c->udp_setup_userdata); + pa_rtsp_record(c->rtsp, &c->seq, &c->rtptime); + + break; + + error: + if (c->udp_stream_fd > 0) { + pa_close(c->udp_stream_fd); + c->udp_stream_fd = -1; + } + if (c->udp_control_fd > 0) { + pa_close(c->udp_control_fd); + c->udp_control_fd = -1; + } + if (c->udp_timing_fd > 0) { + pa_close(c->udp_timing_fd); + c->udp_timing_fd = -1; + } + + pa_rtsp_client_free(c->rtsp); + c->rtsp = NULL; + + c->udp_my_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_server_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_my_timing_port = UDP_DEFAULT_TIMING_PORT; + c->udp_server_timing_port = UDP_DEFAULT_TIMING_PORT; + + pa_log_error("aborting RTSP setup, failed creating required sockets"); + + break; + } + + case STATE_RECORD: { + int32_t latency = 0; + uint32_t rand; + char *alt; + + pa_log_debug("RAOP: RECORD"); + + alt = pa_xstrdup(pa_headerlist_gets(headers, "Audio-Latency")); + /* Generate a random synchronization source identifier from this session. */ + pa_random(&rand, sizeof(rand)); + c->udp_ssrc = rand; + + if (alt) + pa_atoi(alt, &latency); + + c->udp_first_packet = true; + c->udp_sync_count = 0; + + c->udp_record_callback(c->udp_setup_userdata); + + pa_xfree(alt); + break; + } + + case STATE_SET_PARAMETER: { + pa_log_debug("RAOP: SET_PARAMETER"); + + break; + } + + case STATE_FLUSH: { + pa_log_debug("RAOP: FLUSHED"); + + break; + } + + case STATE_TEARDOWN: { + pa_log_debug("RAOP: TEARDOWN"); + pa_assert(c->udp_disconnected_callback); + pa_assert(c->rtsp); + + pa_rtsp_disconnect(c->rtsp); + + if (c->udp_stream_fd > 0) { + pa_close(c->udp_stream_fd); + c->udp_stream_fd = -1; + } + + pa_log_debug("RTSP control channel closed (teardown)"); + + pa_rtsp_client_free(c->rtsp); + pa_xfree(c->sid); + c->rtsp = NULL; + c->sid = NULL; + + /* + Callback for cleanup -- e.g. pollfd + + Share the disconnected callback since TEARDOWN event + is essentially equivalent to DISCONNECTED. + In case some special treatment turns out to be required + for TEARDOWN in future, a new callback function may be + defined and used. + */ + c->udp_disconnected_callback(c->udp_disconnected_userdata); + + /* Control and timing fds are closed by udp_sink_process_msg, + after it disables poll */ + c->udp_control_fd = -1; + c->udp_timing_fd = -1; + + break; + } + + case STATE_DISCONNECTED: { + pa_log_debug("RAOP: DISCONNECTED"); + pa_assert(c->udp_disconnected_callback); + pa_assert(c->rtsp); + + if (c->udp_stream_fd > 0) { + pa_close(c->udp_stream_fd); + c->udp_stream_fd = -1; + } + + pa_log_debug("RTSP control channel closed (disconnected)"); + + pa_rtsp_client_free(c->rtsp); + pa_xfree(c->sid); + c->rtsp = NULL; + c->sid = NULL; + + c->udp_disconnected_callback(c->udp_disconnected_userdata); + /* Control and timing fds are closed by udp_sink_process_msg, + after it disables poll */ + c->udp_control_fd = -1; + c->udp_timing_fd = -1; + + break; + } + } +} + +pa_raop_client* pa_raop_client_new(pa_core *core, const char *host, pa_raop_protocol_t protocol) { pa_raop_client* c; + pa_parsed_address a; + pa_sample_spec ss; pa_assert(core); pa_assert(host); @@ -384,17 +1017,35 @@ pa_raop_client* pa_raop_client_new(pa_core *core, const char *host) { c = pa_xnew0(pa_raop_client, 1); c->core = core; - c->fd = -1; + c->tcp_fd = -1; + c->protocol = protocol; + c->udp_stream_fd = -1; + c->udp_control_fd = -1; + c->udp_timing_fd = -1; + + c->udp_my_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_server_control_port = UDP_DEFAULT_CONTROL_PORT; + c->udp_my_timing_port = UDP_DEFAULT_TIMING_PORT; + c->udp_server_timing_port = UDP_DEFAULT_TIMING_PORT; c->host = a.path_or_host; if (a.port) c->port = a.port; else - c->port = RAOP_PORT; + c->port = DEFAULT_RAOP_PORT; - if (pa_raop_connect(c)) { - pa_raop_client_free(c); - return NULL; + c->udp_first_packet = true; + + ss = core->default_sample_spec; + /* Packet sync interval should be around 1s. */ + c->udp_sync_interval = ss.rate / UDP_FRAMES_PER_PACKET; + c->udp_sync_count = 0; + + if (c->protocol == RAOP_TCP) { + if (pa_raop_client_connect(c)) { + pa_raop_client_free(c); + return NULL; + } } return c; @@ -411,7 +1062,7 @@ void pa_raop_client_free(pa_raop_client *c) { pa_xfree(c); } -int pa_raop_connect(pa_raop_client *c) { +int pa_raop_client_connect(pa_raop_client *c) { char *sci; struct { uint32_t a; @@ -426,7 +1077,10 @@ int pa_raop_connect(pa_raop_client *c) { return 0; } - c->rtsp = pa_rtsp_client_new(c->core->mainloop, c->host, c->port, "iTunes/4.6 (Macintosh; U; PPC Mac OS X 10.3)"); + if (c->protocol == RAOP_TCP) + c->rtsp = pa_rtsp_client_new(c->core->mainloop, c->host, c->port, "iTunes/4.6 (Macintosh; U; PPC Mac OS X 10.3)"); + else + c->rtsp = pa_rtsp_client_new(c->core->mainloop, c->host, c->port, "iTunes/7.6.2 (Windows; N;)"); /* Initialise the AES encryption system. */ pa_random(c->aes_iv, sizeof(c->aes_iv)); @@ -440,20 +1094,179 @@ int pa_raop_connect(pa_raop_client *c) { sci = pa_sprintf_malloc("%08x%08x",rand_data.b, rand_data.c); pa_rtsp_add_header(c->rtsp, "Client-Instance", sci); pa_xfree(sci); - pa_rtsp_set_callback(c->rtsp, rtsp_cb, c); + if (c->protocol == RAOP_TCP) + pa_rtsp_set_callback(c->rtsp, tcp_rtsp_cb, c); + else + pa_rtsp_set_callback(c->rtsp, udp_rtsp_cb, c); return pa_rtsp_connect(c->rtsp); } -int pa_raop_flush(pa_raop_client *c) { +int pa_raop_client_flush(pa_raop_client *c) { + int rv = 0; pa_assert(c); - pa_rtsp_flush(c->rtsp, c->seq, c->rtptime); - return 0; + if (c->rtsp != NULL) { + rv = pa_rtsp_flush(c->rtsp, c->seq, c->rtptime); + c->udp_sync_count = -1; + } + + return rv; +} + +int pa_raop_client_teardown(pa_raop_client *c) { + int rv = 0; + + pa_assert(c); + + if (c->rtsp != NULL) + rv = pa_rtsp_teardown(c->rtsp); + + return rv; +} + +int pa_raop_client_udp_can_stream(pa_raop_client *c) { + int rv = 0; + + pa_assert(c); + + if (c->udp_stream_fd > 0) + rv = 1; + + return rv; +} + +int pa_raop_client_udp_handle_timing_packet(pa_raop_client *c, const uint8_t packet[], ssize_t size) { + const uint32_t * data = NULL; + uint8_t payload = 0; + struct timeval tv; + uint64_t rci = 0; + int rv = 0; + + pa_assert(c); + pa_assert(packet); + + /* Timing packets are 32 bytes long: 1 x 8 RTP header (no ssrc) + 3 x 8 NTP timestamps. */ + if (size != 32 || packet[0] != 0x80) + { + pa_log_debug("Received an invalid timing packet."); + return 1; + } + + data = (uint32_t *) (packet + sizeof(udp_timming_header)); + rci = timeval_to_ntp(pa_rtclock_get(&tv)); + /* The market bit is always set (see rfc3550 for packet structure) ! */ + payload = packet[1] ^ 0x80; + switch (payload) { + case UDP_PAYLOAD_TIMING_REQUEST: + rv = udp_send_timing_packet(c, data, rci); + break; + case UDP_PAYLOAD_TIMING_RESPONSE: + default: + pa_log_debug("Got an unexpected payload type on timing channel !"); + return 1; + } + + return rv; +} + +int pa_raop_client_udp_handle_control_packet(pa_raop_client *c, const uint8_t packet[], ssize_t size) { + uint8_t payload = 0; + int rv = 0; + + pa_assert(c); + pa_assert(packet); + + if (size != 20 || packet[0] != 0x80) + { + pa_log_debug("Received an invalid control packet."); + return 1; + } + + /* The market bit is always set (see rfc3550 for packet structure) ! */ + + payload = packet[1] ^ 0x80; + switch (payload) { + case UDP_PAYLOAD_RETRANSMIT_REQUEST: + /* Packet retransmission not implemented yet... */ + /* rv = ... */ + break; + case UDP_PAYLOAD_RETRANSMIT_REPLY: + default: + pa_log_debug("Got an unexpected payload type on control channel !"); + return 1; + } + + return rv; +} + +int pa_raop_client_udp_get_blocks_size(pa_raop_client *c, size_t *size) { + int rv = 0; + + pa_assert(c); + pa_assert(size); + + *size = UDP_FRAMES_PER_PACKET; + + return rv; +} + +ssize_t pa_raop_client_udp_send_audio_packet(pa_raop_client *c, pa_memchunk *block) { + uint8_t *buf = NULL; + ssize_t len; + + pa_assert(c); + pa_assert(block); + + /* Sync RTP & NTP timestamp if required. */ + if (c->udp_first_packet || c->udp_sync_count >= c->udp_sync_interval) { + udp_send_sync_packet(c, c->rtptime); + c->udp_sync_count = 0; + } else { + c->udp_sync_count++; + } + + buf = pa_memblock_acquire(block->memblock); + pa_assert(buf); + pa_assert(block->length > 0); + udp_build_audio_header(c, (uint32_t *) (buf + block->index), block->length); + len = udp_send_audio_packet(c, buf + block->index, block->length); + pa_memblock_release(block->memblock); + + if (len > 0) { + pa_assert((size_t) len <= block->length); + /* UDP packet has to be sent at once, so it is meaningless to + preseve the partial data + FIXME: This won't happen at least in *NIX systems?? */ + if (block->length > (size_t) len) { + pa_log_warn("Tried to send %zu bytes but managed to send %zu bytes", block->length, len); + len = block->length; + } + block->index += block->length; + block->length = 0; + } + + if (c->udp_first_packet) + c->udp_first_packet = false; + + return len; +} + +/* Adjust volume so that it fits into VOLUME_DEF <= v <= 0 dB */ +pa_volume_t pa_raop_client_adjust_volume(pa_raop_client *c, pa_volume_t volume) { + double minv, maxv; + + if (c->protocol != RAOP_UDP) + return volume; + + maxv = pa_sw_volume_from_dB(0.0); + minv = maxv * pow(10.0, (double) VOLUME_DEF / 60.0); + + return volume - volume * (minv / maxv) + minv; } int pa_raop_client_set_volume(pa_raop_client *c, pa_volume_t volume) { - int rv; + int rv = 0; double db; char *param; @@ -465,10 +1278,13 @@ int pa_raop_client_set_volume(pa_raop_client *c, pa_volume_t volume) { else if (db > VOLUME_MAX) db = VOLUME_MAX; + pa_log_debug("volume=%u db=%.6f", volume, db); + param = pa_sprintf_malloc("volume: %0.6f\r\n", db); /* We just hit and hope, cannot wait for the callback. */ - rv = pa_rtsp_setparameter(c->rtsp, param); + if (c->rtsp != NULL && pa_rtsp_exec_ready(c->rtsp)) + rv = pa_rtsp_setparameter(c->rtsp, param); pa_xfree(param); return rv; @@ -483,21 +1299,23 @@ int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchun uint8_t *b, *p; uint32_t bsize; size_t length; - static uint8_t header[] = { - 0x24, 0x00, 0x00, 0x00, - 0xF0, 0xFF, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, - }; - int header_size = sizeof(header); + const uint8_t *header; + int header_size; pa_assert(c); - pa_assert(c->fd > 0); pa_assert(raw); pa_assert(raw->memblock); pa_assert(raw->length > 0); pa_assert(encoded); + if (c->protocol == RAOP_TCP) { + header = tcp_audio_header; + header_size = sizeof(tcp_audio_header); + } else { + header = udp_audio_header; + header_size = sizeof(udp_audio_header); + } + /* We have to send 4 byte chunks */ bsize = (int)(raw->length / 4); length = bsize * 4; @@ -526,7 +1344,9 @@ int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchun bit_writer(&bp,&bpos,&size,(bsize>>8)&0xff,8); bit_writer(&bp,&bpos,&size,(bsize)&0xff,8); - ibp = p = pa_memblock_acquire(raw->memblock); + p = pa_memblock_acquire(raw->memblock); + p += raw->index; + ibp = p; maxibp = p + raw->length - 4; while (ibp <= maxibp) { /* Byte swap stereo data. */ @@ -538,16 +1358,22 @@ int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchun raw->index += 4; raw->length -= 4; } + if (c->protocol == RAOP_UDP) + c->rtptime += (ibp - p) / 4; pa_memblock_release(raw->memblock); encoded->length = header_size + size; - /* Store the length (endian swapped: make this better). */ - len = size + header_size - 4; - *(b + 2) = len >> 8; - *(b + 3) = len & 0xff; + if (c->protocol == RAOP_TCP) { + /* Store the length (endian swapped: make this better). */ + len = size + header_size - 4; + *(b + 2) = len >> 8; + *(b + 3) = len & 0xff; + } - /* Encrypt our data. */ - aes_encrypt(c, (b + header_size), size); + if (c->encryption) { + /* Encrypt our data. */ + aes_encrypt(c, (b + header_size), size); + } /* We're done with the chunk. */ pa_memblock_release(encoded->memblock); @@ -555,16 +1381,41 @@ int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchun return 0; } -void pa_raop_client_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata) { +void pa_raop_client_tcp_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata) { + pa_assert(c); + + c->tcp_callback = callback; + c->tcp_userdata = userdata; +} + +void pa_raop_client_tcp_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata) { + pa_assert(c); + + c->tcp_closed_callback = callback; + c->tcp_closed_userdata = userdata; +} + +void pa_raop_client_set_encryption(pa_raop_client *c, int encryption) { + c->encryption = encryption; +} + +void pa_raop_client_udp_set_setup_callback(pa_raop_client *c, pa_raop_client_setup_cb_t callback, void *userdata) { + pa_assert(c); + + c->udp_setup_callback = callback; + c->udp_setup_userdata = userdata; +} + +void pa_raop_client_udp_set_record_callback(pa_raop_client *c, pa_raop_client_record_cb_t callback, void *userdata) { pa_assert(c); - c->callback = callback; - c->userdata = userdata; + c->udp_record_callback = callback; + c->udp_record_userdata = userdata; } -void pa_raop_client_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata) { +void pa_raop_client_udp_set_disconnected_callback(pa_raop_client *c, pa_raop_client_disconnected_cb_t callback, void *userdata) { pa_assert(c); - c->closed_callback = callback; - c->closed_userdata = userdata; + c->udp_disconnected_callback = callback; + c->udp_disconnected_userdata = userdata; } diff --git a/src/modules/raop/raop_client.h b/src/modules/raop/raop_client.h index 6ba32e9..36be8dc 100644 --- a/src/modules/raop/raop_client.h +++ b/src/modules/raop/raop_client.h @@ -20,23 +20,52 @@ along with PulseAudio; if not, see <http://www.gnu.org/licenses/>. ***/ +#include <pulse/volume.h> + #include <pulsecore/core.h> +#include <pulsecore/memchunk.h> + +typedef enum pa_raop_protocol { + RAOP_TCP, + RAOP_UDP, +} pa_raop_protocol_t; typedef struct pa_raop_client pa_raop_client; -pa_raop_client* pa_raop_client_new(pa_core *core, const char *host); +pa_raop_client* pa_raop_client_new(pa_core *core, const char *host, pa_raop_protocol_t protocol); void pa_raop_client_free(pa_raop_client *c); -int pa_raop_connect(pa_raop_client *c); -int pa_raop_flush(pa_raop_client *c); +int pa_raop_client_connect(pa_raop_client *c); +int pa_raop_client_flush(pa_raop_client *c); +int pa_raop_client_teardown(pa_raop_client *c); +int pa_raop_client_udp_can_stream(pa_raop_client *c); + +void pa_raop_client_set_encryption(pa_raop_client *c, int encryption); +pa_volume_t pa_raop_client_adjust_volume(pa_raop_client *c, pa_volume_t volume); int pa_raop_client_set_volume(pa_raop_client *c, pa_volume_t volume); int pa_raop_client_encode_sample(pa_raop_client *c, pa_memchunk *raw, pa_memchunk *encoded); +int pa_raop_client_udp_handle_timing_packet(pa_raop_client *c, const uint8_t packet +[], ssize_t size); +int pa_raop_client_udp_handle_control_packet(pa_raop_client *c, const uint8_t packet[], ssize_t size); +int pa_raop_client_udp_get_blocks_size(pa_raop_client *c, size_t *size); +ssize_t pa_raop_client_udp_send_audio_packet(pa_raop_client *c, pa_memchunk *block); + typedef void (*pa_raop_client_cb_t)(int fd, void *userdata); -void pa_raop_client_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata); +void pa_raop_client_tcp_set_callback(pa_raop_client *c, pa_raop_client_cb_t callback, void *userdata); typedef void (*pa_raop_client_closed_cb_t)(void *userdata); -void pa_raop_client_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata); +void pa_raop_client_tcp_set_closed_callback(pa_raop_client *c, pa_raop_client_closed_cb_t callback, void *userdata); + + +typedef void (*pa_raop_client_setup_cb_t)(int control_fd, int timing_fd, void *userdata); +void pa_raop_client_udp_set_setup_callback(pa_raop_client *c, pa_raop_client_setup_cb_t callback, void *userdata); + +typedef void (*pa_raop_client_record_cb_t)(void *userdata); +void pa_raop_client_udp_set_record_callback(pa_raop_client *c, pa_raop_client_record_cb_t callback, void *userdata); + +typedef void (*pa_raop_client_disconnected_cb_t)(void *userdata); +void pa_raop_client_udp_set_disconnected_callback(pa_raop_client *c, pa_raop_client_disconnected_cb_t callback, void *userdata); #endif -- 2.9.3
