WIP: Add support for newest Airplay2 devices

Changed rtsp_client to support response with body
Support timing packet requets earlier (required by newest AirTunes)
Set initial volume before sending RTP packets
Send progress with SET_PARAMETER (required by newest AirTunes)

src/modules/raop/raop-client.c

@@ -107,6 +107,7 @@ struct pa_raop_client {
     int udp_sfd;
     int udp_cfd;
     int udp_tfd;
+    int udp_tport;
 
     pa_raop_packet_buffer *pbuf;
 
@@ -124,6 +125,12 @@ struct pa_raop_client {
 
     pa_raop_client_state_cb_t state_callback;
     void *state_userdata;
+
+    // Airplay 2 devices require additionnal authentication
+    // It actually is useful to authenticate Airplay server, not the device (see pa_rtsp_auth_setup for more
+    // details)
+    bool require_post_auth;
+    pa_volume_t initial_volume;
 };
 
 /* Audio TCP packet header [16x8] (cf. rfc4571):
@@ -860,14 +867,37 @@ static void rtsp_stream_cb(pa_rtsp_client *rtsp, pa_rtsp_state_t state, pa_rtsp_
     pa_assert(rtsp == c->rtsp);
 
     switch (state) {
-        case STATE_CONNECT: {
+        case STATE_CONNECT:
+        case STATE_AUTH_SETUP:   {
+            if (c->require_post_auth && state == STATE_CONNECT) {
+                struct {
+                    uint32_t ci1;
+                    uint32_t ci2;
+                } rci;
+                pa_log_debug("RAOP: CONNECTED");
+
+                pa_random(&rci, sizeof(rci));
+                c->sci = pa_sprintf_malloc("%08x%08x", rci.ci1, rci.ci2);
+                pa_rtsp_add_header(c->rtsp, "Client-Instance", c->sci);
+
+                if (pa_rtsp_auth_setup(
+                        c->rtsp,
+                        "\x01\x4e\xea\xd0\x4e\xa9\x2e\x47\x69\xd2\xe1\xfb\xd0\x96\x81\xd5\x94\xa8\xef\x18\x45\x4a\x24\xae\xaf\xb3\x14\x97\x0d\xa0\xb5\xa3\x49"
+                ) < 0) {
+                    pa_log_error("RAOP: The device supports POST method but we were unable to POST auth-setup");
+                }
+
+                break;
+            }
+            if (!c->require_post_auth || (c->require_post_auth && state == STATE_AUTH_SETUP)) {
+                {
                     char *key, *iv, *sdp = NULL;
                     int frames = 0;
                     const char *ip;
                     char *url;
                     int ipv;
 
-            pa_log_debug("RAOP: CONNECTED");
+                    pa_log_debug("RAOP: AUTH SETUP");
 
                     ip = pa_rtsp_localip(c->rtsp);
                     if (pa_is_ip6_address(ip)) {
@@ -942,6 +972,10 @@ connect_finish:
                     pa_xfree(url);
                     break;
                 }
+            }
+
+            break;
+        }
 
         case STATE_OPTIONS: {
             pa_log_debug("RAOP: OPTIONS (stream cb)");
@@ -971,6 +1005,9 @@ connect_finish:
                     cport, tport);
             }
 
+            if (c->state_callback)
+                c->state_callback(PA_RAOP_CONNECTED, c->state_userdata);
+
             pa_rtsp_setup(c->rtsp, trs);
 
             pa_xfree(trs);
@@ -1076,8 +1113,6 @@ connect_finish:
 
                 pa_log_debug("Connection established (UDP;control_port=%d;timing_port=%d)", cport, tport);
 
-                if (c->state_callback)
-                    c->state_callback(PA_RAOP_CONNECTED, c->state_userdata);
             }
 
             pa_rtsp_record(c->rtsp, &c->seq, &c->rtptime);
@@ -1143,6 +1178,13 @@ connect_finish:
         case STATE_SET_PARAMETER: {
             pa_log_debug("RAOP: SET_PARAMETER");
 
+            if (c->initial_volume != 0){
+                c->initial_volume = 0;
+                // We just have set initial volume, so raise PA_RAOP_VOLUME_INIT to chain
+                if (c->state_callback)
+                    c->state_callback((int) PA_RAOP_VOLUME_INIT, c->state_userdata);
+            }
+
             break;
         }
 
@@ -1164,7 +1206,7 @@ connect_finish:
             c->udp_sfd = -1;
 
             /* Polling sockets will be closed by sink */
-            c->udp_cfd = c->udp_tfd = -1;
+            c->udp_cfd = c->udp_tfd = c->udp_tport = -1;
             c->tcp_sfd = -1;
 
             pa_rtsp_client_free(c->rtsp);
@@ -1192,8 +1234,9 @@ connect_finish:
             c->udp_sfd = -1;
 
             /* Polling sockets will be closed by sink */
-            c->udp_cfd = c->udp_tfd = -1;
+            c->udp_cfd = c->udp_tfd = c->udp_tport = -1;
             c->tcp_sfd = -1;
+            c->initial_volume = 0;
 
             pa_log_error("RTSP control channel closed (disconnected)");
 
@@ -1318,6 +1361,7 @@ static void rtsp_auth_cb(pa_rtsp_client *rtsp, pa_rtsp_state_t state, pa_rtsp_st
             if (STATUS_OK == status) {
                 publ = pa_xstrdup(pa_headerlist_gets(headers, "Public"));
                 c->sci = pa_xstrdup(pa_rtsp_get_header(c->rtsp, "Client-Instance"));
+                c->require_post_auth = true;
 
                 if (c->password)
                     pa_xfree(c->password);
@@ -1417,6 +1461,8 @@ pa_raop_client* pa_raop_client_new(pa_core *core, const char *host, pa_raop_prot
     c->udp_sfd = -1;
     c->udp_cfd = -1;
     c->udp_tfd = -1;
+    c->udp_tport = -1;
+    c->initial_volume = 0;
 
     c->secret = NULL;
     if (c->encryption != PA_RAOP_ENCRYPTION_NONE)
@@ -1597,6 +1643,10 @@ int pa_raop_client_stream(pa_raop_client *c) {
     return rv;
 }
 
+void pa_raop_client_set_initial_volume(pa_raop_client *c, pa_volume_t initial_volume) {
+    c->initial_volume = initial_volume;
+}
+
 int pa_raop_client_set_volume(pa_raop_client *c, pa_volume_t volume) {
     char *param;
     int rv = 0;
@@ -1792,3 +1842,28 @@ void pa_raop_client_set_state_callback(pa_raop_client *c, pa_raop_client_state_c
     c->state_callback = callback;
     c->state_userdata = userdata;
 }
+
+void pa_raop_client_set_tport(pa_raop_client *c, int udp_tport) {
+    pa_assert(c);
+
+    if (c->udp_tport < 0) {
+        c->udp_tport = udp_tport;
+        if ((c->udp_tfd = connect_udp_socket(c, c->udp_tfd, udp_tport)) <= 0) {
+            pa_log_error("RAOP: Error while connecting the UDP timing port");
+        }
+    }
+}
+
+void pa_raop_client_send_progress (pa_raop_client *c){
+    char *param;
+
+    pa_assert(c);
+
+    param = pa_sprintf_malloc("progress: %s/%s/%s\r\n", "0","0","0");
+    /* We just hit and hope, cannot wait for the callback. */
+    if (c->rtsp != NULL && pa_rtsp_exec_ready(c->rtsp))
+        pa_rtsp_setparameter(c->rtsp, param);
+
+    pa_xfree(param);
+
+}

src/modules/raop/raop-client.h

@@ -53,6 +53,7 @@ typedef enum pa_raop_state {
     PA_RAOP_AUTHENTICATED,
     PA_RAOP_CONNECTED,
     PA_RAOP_RECORDING,
+    PA_RAOP_VOLUME_INIT,
     PA_RAOP_DISCONNECTED
 } pa_raop_state_t;
 
@@ -79,5 +80,8 @@ ssize_t pa_raop_client_send_audio_packet(pa_raop_client *c, pa_memchunk *block,
 
 typedef void (*pa_raop_client_state_cb_t)(pa_raop_state_t state, void *userdata);
 void pa_raop_client_set_state_callback(pa_raop_client *c, pa_raop_client_state_cb_t callback, void *userdata);
+void pa_raop_client_set_tport(pa_raop_client *c, int udp_tport);
+void pa_raop_client_set_initial_volume(pa_raop_client *c, pa_volume_t initial_volume);
+void pa_raop_client_send_progress (pa_raop_client *c);
 
 #endif

src/modules/rtp/rtsp_client.c

@@ -66,6 +66,7 @@ struct pa_rtsp_client {
     char *last_header;
     pa_strbuf *header_buffer;
     pa_headerlist* response_headers;
+    int content_length;
 
     char *localip;
     char *url;
@@ -169,6 +170,12 @@ static void headers_read(pa_rtsp_client *c) {
     c->callback(c, c->state, c->status, c->response_headers, c->userdata);
 }
 
+static void stream_callback(pa_ioline *line, void *userdata) {
+    pa_rtsp_client *c = userdata;
+
+    headers_read(c);
+}
+
 static void line_callback(pa_ioline *line, const char *s, void *userdata) {
     pa_rtsp_client *c = userdata;
     char *delimpos;
@@ -200,6 +207,7 @@ static void line_callback(pa_ioline *line, const char *s, void *userdata) {
 
         c->status = STATUS_OK;
         c->waiting = 0;
+        c->content_length = 0;
         goto exit;
     } else if (c->waiting && pa_streq(s2, "RTSP/1.0 401 Unauthorized")) {
         if (c->response_headers)
@@ -209,6 +217,14 @@ static void line_callback(pa_ioline *line, const char *s, void *userdata) {
         c->status = STATUS_UNAUTHORIZED;
         c->waiting = 0;
         goto exit;
+    } else if (c->waiting && pa_streq(s2, "RTSP/1.0 403 Forbidden")) {
+        if (c->response_headers)
+            pa_headerlist_free(c->response_headers);
+        c->response_headers = pa_headerlist_new();
+
+        c->status = STATUS_FORBIDDEN;
+        c->waiting = 0;
+        goto exit;
     } else if (c->waiting) {
         pa_log_warn("Unexpected/Unhandled response: %s", s2);
 
@@ -235,7 +251,13 @@ static void line_callback(pa_ioline *line, const char *s, void *userdata) {
         }
 
         pa_log_debug("Full response received. Dispatching");
+
+        if (c->content_length!=0) {
+            pa_ioline_set_streamcallback(line, stream_callback, c->content_length, c);
+        } else {
             headers_read(c);
+        }
+
         goto exit;
     }
 
@@ -256,6 +278,8 @@ static void line_callback(pa_ioline *line, const char *s, void *userdata) {
         /* This is not a continuation header so let's dump the full
           header/value into our proplist */
         pa_headerlist_puts(c->response_headers, c->last_header, tmp);
+        if (pa_headerlist_gets(c->response_headers, "Content-Length"))
+            pa_atoi(pa_headerlist_gets(c->response_headers, "Content-Length"), &c->content_length);
         pa_xfree(tmp);
         pa_xfree(c->last_header);
         c->last_header = NULL;
@@ -501,6 +525,22 @@ int pa_rtsp_options(pa_rtsp_client *c) {
     return rv;
 }
 
+int pa_rtsp_auth_setup(pa_rtsp_client *c, const char* key) {
+    char *url;
+    int rv;
+
+    pa_assert(c);
+
+    url = c->url;
+    c->state = STATE_AUTH_SETUP;
+
+    c->url = (char *)"/auth-setup";
+    rv = rtsp_exec(c, "POST", "application/octet-stream", key, 1, NULL);
+
+    c->url = url;
+    return rv;
+}
+
 int pa_rtsp_announce(pa_rtsp_client *c, const char *sdp) {
     int rv;
 

src/modules/rtp/rtsp_client.h

@@ -37,6 +37,7 @@ typedef enum pa_rtsp_state {
   STATE_OPTIONS,
   STATE_ANNOUNCE,
   STATE_SETUP,
+  STATE_AUTH_SETUP,
   STATE_RECORD,
   STATE_SET_PARAMETER,
   STATE_FLUSH,
@@ -48,6 +49,7 @@ typedef enum pa_rtsp_status {
   STATUS_OK             = 200,
   STATUS_BAD_REQUEST    = 400,
   STATUS_UNAUTHORIZED   = 401,
+  STATUS_FORBIDDEN      = 403,
   STATUS_NO_RESPONSE    = 444,
   STATUS_INTERNAL_ERROR = 500
 } pa_rtsp_status_t;
@@ -79,5 +81,6 @@ int pa_rtsp_record(pa_rtsp_client *c, uint16_t *seq, uint32_t *rtptime);
 int pa_rtsp_setparameter(pa_rtsp_client *c, const char *param);
 int pa_rtsp_flush(pa_rtsp_client *c, uint16_t seq, uint32_t rtptime);
 int pa_rtsp_teardown(pa_rtsp_client *c);
+int pa_rtsp_auth_setup(pa_rtsp_client *c, const char* auth_key);
 
 #endif

src/pulsecore/ioline.c

@@ -51,13 +51,16 @@ struct pa_ioline {
     size_t wbuf_length, wbuf_index, wbuf_valid_length;
 
     char *rbuf;
-    size_t rbuf_length, rbuf_index, rbuf_valid_length;
+    size_t rbuf_length, rbuf_index, rbuf_valid_length, block_byte_count, rbuf_blockindex;
 
     pa_ioline_cb_t callback;
+    pa_stream_cb_t streamcallback;
+    size_t streamcallback_maxbyte;
     void *userdata;
 
     pa_ioline_drain_cb_t drain_callback;
     void *drain_userdata;
+    void *stream_userdata;
 
     bool dead:1;
     bool defer_close:1;
@@ -81,6 +84,8 @@ pa_ioline* pa_ioline_new(pa_iochannel *io) {
     l->rbuf_length = l->rbuf_index = l->rbuf_valid_length = 0;
 
     l->callback = NULL;
+    l->streamcallback = NULL;
+    l->streamcallback_maxbyte = 0;
     l->userdata = NULL;
 
     l->drain_callback = NULL;
@@ -207,6 +212,20 @@ void pa_ioline_set_callback(pa_ioline*l, pa_ioline_cb_t callback, void *userdata
     l->userdata = userdata;
 }
 
+void pa_ioline_set_streamcallback(pa_ioline*l, pa_stream_cb_t callback, size_t maxbyte, void *userdata) {
+    pa_assert(l);
+    pa_assert(PA_REFCNT_VALUE(l) >= 1);
+
+    if (l->dead)
+        return;
+
+    l->streamcallback = callback;
+    l->stream_userdata = userdata;
+    l->streamcallback_maxbyte=maxbyte;
+    l->block_byte_count = 0;
+    l->rbuf_blockindex = l->rbuf_index;
+}
+
 void pa_ioline_set_drain_callback(pa_ioline*l, pa_ioline_drain_cb_t callback, void *userdata) {
     pa_assert(l);
     pa_assert(PA_REFCNT_VALUE(l) >= 1);
@@ -268,6 +287,10 @@ static void scan_for_lines(pa_ioline *l, size_t skip) {
         if (l->callback)
             l->callback(l, pa_strip_nl(p), l->userdata);
 
+         // If we switched to block mode during the callback, exit
+         if (l->streamcallback_maxbyte !=0){
+             return;
+         }
         skip = 0;
     }
 
@@ -285,6 +308,7 @@ static int do_read(pa_ioline *l) {
     while (l->io && !l->dead && pa_iochannel_is_readable(l->io)) {
         ssize_t r;
         size_t len;
+        size_t byte_count;
 
         len = l->rbuf_length - l->rbuf_index - l->rbuf_valid_length;
 
@@ -332,9 +356,29 @@ static int do_read(pa_ioline *l) {
         }
 
         l->rbuf_valid_length += (size_t) r;
-
+        // If in block mode, append until we reach the size
+        if (l->streamcallback_maxbyte == 0){
             /* Look if a line has been terminated in the newly read data */
             scan_for_lines(l, l->rbuf_valid_length - (size_t) r);
+            byte_count = l->rbuf_valid_length;
+        } else {
+            byte_count = r;
+        }
+
+        // Not in a "else" clause, since we may have enabled block mode in callback called in scan_for_lines
+        if (l->streamcallback_maxbyte > 0) {
+            l->block_byte_count += byte_count;
+
+            if (l->block_byte_count >= l->streamcallback_maxbyte) {
+                l->rbuf_index += l->streamcallback_maxbyte;
+                l->rbuf_valid_length -= l->streamcallback_maxbyte;
+                l->streamcallback_maxbyte = 0;
+                if (l->rbuf_valid_length == 0)
+                    l->rbuf_index = 0;
+                if (l->streamcallback)
+                    l->streamcallback(l, l->stream_userdata);
+            }
+        }
     }
 
     return 0;

src/pulsecore/ioline.h

@@ -31,6 +31,7 @@
 typedef struct pa_ioline pa_ioline;
 
 typedef void (*pa_ioline_cb_t)(pa_ioline*io, const char *s, void *userdata);
+typedef void (*pa_stream_cb_t)(pa_ioline*io, void *userdata);
 typedef void (*pa_ioline_drain_cb_t)(pa_ioline *io, void *userdata);
 
 pa_ioline* pa_ioline_new(pa_iochannel *io);
@@ -47,6 +48,9 @@ void pa_ioline_printf(pa_ioline *s, const char *format, ...) PA_GCC_PRINTF_ATTR(
 /* Set the callback function that is called for every received line */
 void pa_ioline_set_callback(pa_ioline*io, pa_ioline_cb_t callback, void *userdata);
 
+/* Set the callback function that is called once maxbyte has been reached */
+void pa_ioline_set_streamcallback(pa_ioline*io, pa_stream_cb_t callback, size_t maxbyte, void *userdata);
+
 /* Set the callback function that is called when everything has been written */
 void pa_ioline_set_drain_callback(pa_ioline*io, pa_ioline_drain_cb_t callback, void *userdata);