Move closer to an asynchronous structure (still some parsing code to be converted).

Move type definition into .c file to keep it private
Add more utility functions to add/remove headers and return the serverport now the structure is private.
This commit will break the test application but I will fix that in due course

git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/coling@2365 fefdeb5f-60dc-0310-8127-8f9354f1896f

src/modules/rtp/rtsp.c

@@ -49,6 +49,22 @@
 
 #include "rtsp.h"
 
+struct pa_rtsp_context {
+    pa_socket_client *sc;
+    pa_iochannel *io;
+    pa_rtsp_cb_t callback;
+    void* userdata;
+    const char* useragent;
+    pa_headerlist* headers;
+    char* localip;
+    char* url;
+    uint32_t port;
+    uint32_t cseq;
+    char* session;
+    char* transport;
+    pa_rtsp_state state;
+};
+
 /*
  * read one line from the file descriptor
  * timeout: msec unit, -1 for infinite
@@ -112,18 +128,10 @@ static int pa_read_line(pa_iochannel* io, char *line, int maxlen, int timeout)
 static int pa_rtsp_exec(pa_rtsp_context* c, const char* cmd,
                         const char* content_type, const char* content,
                         int expect_response,
-                        pa_headerlist* headers, pa_headerlist** response_headers) {
+                        pa_headerlist* headers) {
     pa_strbuf* buf;
     char* hdrs;
     ssize_t l;
-    char response[1024];
-    int timeout;
-    char* token;
-    const char* token_state;
-    char delimiters[2];
-    char* header;
-    char* delimpos;
-
 
     pa_assert(c);
     pa_assert(c->url);
@@ -167,91 +175,6 @@ static int pa_rtsp_exec(pa_rtsp_context* c, const char* cmd,
     l = pa_iochannel_write(c->io, hdrs, strlen(hdrs));
     pa_xfree(hdrs);
 
-    /* Do we expect a response? */
-    if (!expect_response)
-        return 0;
-
-    timeout = 5000;
-    if (pa_read_line(c->io, response, sizeof(response), timeout) <= 0) {
-        /*ERRMSG("%s: request failed\n",__func__);*/
-        return -1;
-    }
-
-    delimiters[0] = ' ';
-    delimiters[1] = '\0';
-    token_state = NULL;
-    pa_xfree(pa_split(response, delimiters, &token_state));
-    token = pa_split(response, delimiters, &token_state);
-    if (!token || strcmp(token, "200")) {
-        pa_xfree(token);
-        /*ERRMSG("%s: request failed, error %s\n",__func__,token);*/
-        return -1;
-    }
-    pa_xfree(token);
-
-    /* We want to return the headers? */
-    if (!response_headers)
-    {
-        /* We have no storage, so just clear out the response. */
-        while (pa_read_line(c->io, response, sizeof(response), timeout) > 0) {
-            /* Reduce timeout for future requests */
-            timeout = 1000;
-        }
-        return 0;
-    }
-
-    /* TODO: Move header reading into the headerlist. */
-    header = NULL;
-    buf = pa_strbuf_new();
-    while (pa_read_line(c->io, response, sizeof(response), timeout) > 0) {
-        /* Reduce timeout for future requests */
-        timeout = 1000;
-
-        /* If the first character is a space, it's a continuation header */
-        if (header && ' ' == response[0]) {
-            /* Add this line to the buffer (sans the space. */
-            pa_strbuf_puts(buf, &(response[1]));
-            continue;
-        }
-
-        if (header) {
-            /* This is not a continuation header so let's dump the full
-               header/value into our proplist */
-            pa_headerlist_puts(*response_headers, header, pa_strbuf_tostring_free(buf));
-            pa_xfree(header);
-            buf = pa_strbuf_new();
-        }
-
-        delimpos = strstr(response, ":");
-        if (!delimpos) {
-            /*ERRMSG("%s: Request failed, bad header\n",__func__);*/
-            return -1;
-        }
-
-        if (strlen(delimpos) > 1) {
-            /* Cut our line off so we can copy the header name out */
-            *delimpos++ = '\0';
-
-            /* Trim the front of any spaces */
-            while (' ' == *delimpos)
-                ++delimpos;
-
-            pa_strbuf_puts(buf, delimpos);
-        } else {
-            /* Cut our line off so we can copy the header name out */
-            *delimpos = '\0';
-        }
-
-        /* Save the header name */
-        header = pa_xstrdup(response);
-    }
-    /* We will have a header left from our looping itteration, so add it in :) */
-    if (header) {
-        /* This is not a continuation header so let's dump it into our proplist */
-        pa_headerlist_puts(*response_headers, header, pa_strbuf_tostring(buf));
-    }
-    pa_strbuf_free(buf);
-
     return 0;
 }
 
@@ -286,6 +209,146 @@ void pa_rtsp_context_free(pa_rtsp_context* c) {
 }
 
 
+static void io_callback(PA_GCC_UNUSED pa_iochannel *io, void *userdata) {
+    pa_strbuf* buf;
+    pa_headerlist* response_headers = NULL;
+    char response[1024];
+    int timeout;
+    char* token;
+    char* header;
+    char* delimpos;
+    char delimiters[] = " ";
+    pa_rtsp_context *c = userdata;
+    pa_assert(c);
+
+    /* TODO: convert this to a pa_ioline based reader */
+    if (STATE_CONNECT == c->state) {
+        response_headers = pa_headerlist_new();
+    }
+    timeout = 5000;
+    /* read in any response headers */
+    if (pa_read_line(c->io, response, sizeof(response), timeout) > 0) {
+        const char* token_state = NULL;
+
+        timeout = 1000;
+        pa_xfree(pa_split(response, delimiters, &token_state));
+        token = pa_split(response, delimiters, &token_state);
+        if (!token || strcmp(token, "200")) {
+            pa_xfree(token);
+            pa_log("Invalid Response");
+            /* TODO: Bail out completely */
+            return;
+        }
+        pa_xfree(token);
+
+        /* We want to return the headers? */
+        if (!response_headers) {
+            /* We have no storage, so just clear out the response. */
+            while (pa_read_line(c->io, response, sizeof(response), timeout) > 0);
+        } else {
+            /* TODO: Move header reading into the headerlist. */
+            header = NULL;
+            buf = pa_strbuf_new();
+            while (pa_read_line(c->io, response, sizeof(response), timeout) > 0) {
+                /* If the first character is a space, it's a continuation header */
+                if (header && ' ' == response[0]) {
+                    /* Add this line to the buffer (sans the space. */
+                    pa_strbuf_puts(buf, &(response[1]));
+                    continue;
+                }
+
+                if (header) {
+                    /* This is not a continuation header so let's dump the full
+                      header/value into our proplist */
+                    pa_headerlist_puts(response_headers, header, pa_strbuf_tostring_free(buf));
+                    pa_xfree(header);
+                    buf = pa_strbuf_new();
+                }
+
+                delimpos = strstr(response, ":");
+                if (!delimpos) {
+                    pa_log("Invalid response header");
+                    return;
+                }
+
+                if (strlen(delimpos) > 1) {
+                    /* Cut our line off so we can copy the header name out */
+                    *delimpos++ = '\0';
+
+                    /* Trim the front of any spaces */
+                    while (' ' == *delimpos)
+                        ++delimpos;
+
+                    pa_strbuf_puts(buf, delimpos);
+                } else {
+                    /* Cut our line off so we can copy the header name out */
+                    *delimpos = '\0';
+                }
+
+                /* Save the header name */
+                header = pa_xstrdup(response);
+            }
+            /* We will have a header left from our looping itteration, so add it in :) */
+            if (header) {
+                /* This is not a continuation header so let's dump it into our proplist */
+                pa_headerlist_puts(response_headers, header, pa_strbuf_tostring(buf));
+            }
+            pa_strbuf_free(buf);
+        }
+    }
+
+    /* Deal with a CONNECT response */
+    if (STATE_CONNECT == c->state) {
+        const char* token_state = NULL;
+        const char* pc = NULL;
+        c->session = pa_xstrdup(pa_headerlist_gets(response_headers, "Session"));
+        c->transport = pa_xstrdup(pa_headerlist_gets(response_headers, "Transport"));
+
+        if (!c->session || !c->transport) {
+            pa_headerlist_free(response_headers);
+            return;
+        }
+
+        /* Now parse out the server port component of the response. */
+        c->port = 0;
+        delimiters[0] = ';';
+        while ((token = pa_split(c->transport, delimiters, &token_state))) {
+            if ((pc = strstr(token, "="))) {
+                if (0 == strncmp(token, "server_port", 11)) {
+                    pa_atou(pc+1, &c->port);
+                    pa_xfree(token);
+                    break;
+                }
+            }
+            pa_xfree(token);
+        }
+        if (0 == c->port) {
+            /* Error no server_port in response */
+            pa_headerlist_free(response_headers);
+            return;
+        }
+    }
+
+    /* Call our callback */
+    if (c->callback)
+        c->callback(c, c->state, response_headers, c->userdata);
+
+
+    if (response_headers)
+        pa_headerlist_free(response_headers);
+
+    /*
+    if (do_read(u) < 0 || do_write(u) < 0) {
+
+        if (u->io) {
+            pa_iochannel_free(u->io);
+            u->io = NULL;
+        }
+
+       pa_module_unload_request(u->module);
+    }
+    */
+}
 
 static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
     pa_rtsp_context *c = userdata;
@@ -309,6 +372,7 @@ static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata
     }
     pa_assert(!c->io);
     c->io = io;
+    pa_iochannel_set_callback(c->io, io_callback, c);
 
     /* Get the local IP address for use externally */
     if (0 == getsockname(pa_iochannel_get_recv_fd(io), &sa.sa, &sa_len)) {
@@ -337,9 +401,16 @@ int pa_rtsp_connect(pa_rtsp_context *c, pa_mainloop_api *mainloop, const char* h
     }
 
     pa_socket_client_set_callback(c->sc, on_connection, c);
+    c->state = STATE_CONNECT;
     return 0;
 }
 
+void pa_rtsp_set_callback(pa_rtsp_context *c, pa_rtsp_cb_t callback, void *userdata) {
+    pa_assert(c);
+
+    c->callback = callback;
+    c->userdata = userdata;
+}
 
 void pa_rtsp_disconnect(pa_rtsp_context *c) {
     pa_assert(c);
@@ -356,6 +427,11 @@ const char* pa_rtsp_localip(pa_rtsp_context* c) {
     return c->localip;
 }
 
+uint32_t pa_rtsp_serverport(pa_rtsp_context* c) {
+    pa_assert(c);
+
+    return c->port;
+}
 
 void pa_rtsp_set_url(pa_rtsp_context* c, const char* url) {
     pa_assert(c);
@@ -363,67 +439,46 @@ void pa_rtsp_set_url(pa_rtsp_context* c, const char* url) {
     c->url = pa_xstrdup(url);
 }
 
+void pa_rtsp_add_header(pa_rtsp_context *c, const char* key, const char* value)
+{
+    pa_assert(c);
+    pa_assert(key);
+    pa_assert(value);
+
+    pa_headerlist_puts(c->headers, key, value);
+}
+
+void pa_rtsp_remove_header(pa_rtsp_context *c, const char* key)
+{
+    pa_assert(c);
+    pa_assert(key);
+
+    pa_headerlist_remove(c->headers, key);
+}
+
 int pa_rtsp_announce(pa_rtsp_context *c, const char* sdp) {
     pa_assert(c);
     if (!sdp)
         return -1;
 
-    return pa_rtsp_exec(c, "ANNOUNCE", "application/sdp", sdp, 1, NULL, NULL);
+    c->state = STATE_ANNOUNCE;
+    return pa_rtsp_exec(c, "ANNOUNCE", "application/sdp", sdp, 1, NULL);
 }
 
 
-int pa_rtsp_setup(pa_rtsp_context* c, pa_headerlist** response_headers) {
+int pa_rtsp_setup(pa_rtsp_context* c) {
     pa_headerlist* headers;
-    pa_headerlist* rheaders;
+    int rv;
-    char delimiters[2];
-    char* token;
-    const char* token_state;
-    const char* pc;
 
     pa_assert(c);
 
     headers = pa_headerlist_new();
-    rheaders = pa_headerlist_new();
     pa_headerlist_puts(headers, "Transport", "RTP/AVP/TCP;unicast;interleaved=0-1;mode=record");
 
-    if (pa_rtsp_exec(c, "SETUP", NULL, NULL, 1, headers, &rheaders)) {
+    c->state = STATE_SETUP;
-        pa_headerlist_free(headers);
+    rv = pa_rtsp_exec(c, "SETUP", NULL, NULL, 1, headers);
-        pa_headerlist_free(rheaders);
-        return -1;
-    }
     pa_headerlist_free(headers);
-
+    return rv;
-    c->session = pa_xstrdup(pa_headerlist_gets(rheaders, "Session"));
-    c->transport = pa_xstrdup(pa_headerlist_gets(rheaders, "Transport"));
-
-    if (!c->session || !c->transport) {
-        pa_headerlist_free(rheaders);
-        return -1;
-    }
-
-    /* Now parse out the server port component of the response. */
-    c->port = 0;
-    delimiters[0] = ';';
-    delimiters[1] = '\0';
-    token_state = NULL;
-    while ((token = pa_split(c->transport, delimiters, &token_state))) {
-        if ((pc = strstr(token, "="))) {
-            if (0 == strncmp(token, "server_port", 11)) {
-                pa_atou(pc+1, &c->port);
-                pa_xfree(token);
-                break;
-            }
-        }
-        pa_xfree(token);
-    }
-    if (0 == c->port) {
-        /* Error no server_port in response */
-        pa_headerlist_free(rheaders);
-        return -1;
-    }
-
-    *response_headers = rheaders;
-    return 0;
 }
 
 
@@ -441,7 +496,8 @@ int pa_rtsp_record(pa_rtsp_context* c) {
     pa_headerlist_puts(headers, "Range", "npt=0-");
     pa_headerlist_puts(headers, "RTP-Info", "seq=0;rtptime=0");
 
-    rv = pa_rtsp_exec(c, "RECORD", NULL, NULL, 1, headers, NULL);
+    c->state = STATE_RECORD;
+    rv = pa_rtsp_exec(c, "RECORD", NULL, NULL, 1, headers);
     pa_headerlist_free(headers);
     return rv;
 }
@@ -450,7 +506,8 @@ int pa_rtsp_record(pa_rtsp_context* c) {
 int pa_rtsp_teardown(pa_rtsp_context *c) {
     pa_assert(c);
 
-    return pa_rtsp_exec(c, "TEARDOWN", NULL, NULL, 0, NULL, NULL);
+    c->state = STATE_TEARDOWN;
+    return pa_rtsp_exec(c, "TEARDOWN", NULL, NULL, 0, NULL);
 }
 
 
@@ -459,7 +516,8 @@ int pa_rtsp_setparameter(pa_rtsp_context *c, const char* param) {
     if (!param)
         return -1;
 
-    return pa_rtsp_exec(c, "SET_PARAMETER", "text/parameters", param, 1, NULL, NULL);
+    c->state = STATE_SET_PARAMETER;
+    return pa_rtsp_exec(c, "SET_PARAMETER", "text/parameters", param, 1, NULL);
 }
 
 
@@ -472,7 +530,8 @@ int pa_rtsp_flush(pa_rtsp_context *c) {
     headers = pa_headerlist_new();
     pa_headerlist_puts(headers, "RTP-Info", "seq=0;rtptime=0");
 
-    rv = pa_rtsp_exec(c, "FLUSH", NULL, NULL, 1, headers, NULL);
+    c->state = STATE_FLUSH;
+    rv = pa_rtsp_exec(c, "FLUSH", NULL, NULL, 1, headers);
     pa_headerlist_free(headers);
     return rv;
 }

src/modules/rtp/rtsp.h

@@ -36,30 +36,35 @@
 
 #include "headerlist.h"
 
-typedef struct pa_rtsp_context {
+typedef struct pa_rtsp_context pa_rtsp_context;
-    pa_socket_client *sc;
+typedef enum {
-    pa_iochannel *io;
+  STATE_CONNECT,
-    const char* useragent;
+  STATE_ANNOUNCE,
-    pa_headerlist* headers;
+  STATE_SETUP,
-    char* localip;
+  STATE_RECORD,
-    char* url;
+  STATE_TEARDOWN,
-    uint32_t port;
+  STATE_SET_PARAMETER,
-    uint32_t cseq;
+  STATE_FLUSH
-    char* session;
+} pa_rtsp_state;
-    char* transport;
+typedef void (*pa_rtsp_cb_t)(pa_rtsp_context *c, pa_rtsp_state state, pa_headerlist* hl, void *userdata);
-} pa_rtsp_context;
 
 pa_rtsp_context* pa_rtsp_context_new(const char* useragent);
 void pa_rtsp_context_free(pa_rtsp_context* c);
 
 int pa_rtsp_connect(pa_rtsp_context* c, pa_mainloop_api *mainloop, const char* hostname, uint16_t port);
+void pa_rtsp_set_callback(pa_rtsp_context *c, pa_rtsp_cb_t callback, void *userdata);
+
 void pa_rtsp_disconnect(pa_rtsp_context* c);
 
 const char* pa_rtsp_localip(pa_rtsp_context* c);
+uint32_t pa_rtsp_serverport(pa_rtsp_context* c);
 void pa_rtsp_set_url(pa_rtsp_context* c, const char* url);
+void pa_rtsp_add_header(pa_rtsp_context *c, const char* key, const char* value);
+void pa_rtsp_remove_header(pa_rtsp_context *c, const char* key);
+
 int pa_rtsp_announce(pa_rtsp_context* c, const char* sdp);
 
-int pa_rtsp_setup(pa_rtsp_context* c, pa_headerlist** response_headers);
+int pa_rtsp_setup(pa_rtsp_context* c);
 int pa_rtsp_record(pa_rtsp_context* c);
 int pa_rtsp_teardown(pa_rtsp_context* c);