most complete pulse driver

git-svn-id: file:///home/lennart/svn/public/libcanberra/trunk@9 01b60673-d06a-42c0-afdd-89cb8e0f78ac

8 files changed, 812 insertions, 137 deletions

diff --git a/canberra.h b/canberra.h
index 110c4b9..2d48cd8 100644
--- a/canberra.h
+++ b/canberra.h
@@ -79,7 +79,10 @@
 #define CA_GCC_SENTINEL
 #endif
 
-/** Context, event, and playback properties */
+/** Properties for the media that is being played, about the event
+ * that caused the media to play, the window on which behalf this
+ * media is being played, the application that this window belongs to
+ * and finally properties for libcanberra specific usage. */
 #define CA_PROP_MEDIA_NAME                         "media.name"
 #define CA_PROP_MEDIA_TITLE                        "media.title"
 #define CA_PROP_MEDIA_ARTIST                       "media.artist"
@@ -92,6 +95,10 @@
 #define CA_PROP_EVENT_MOUSE_X                      "event.mouse.x"
 #define CA_PROP_EVENT_MOUSE_Y                      "event.mouse.y"
 #define CA_PROP_EVENT_MOUSE_BUTTON                 "event.mouse.button"
+#define CA_PROP_WINDOW_NAME                        "window.name"
+#define CA_PROP_WINDOW_ID                          "window.id"
+#define CA_PROP_WINDOW_ICON                        "window.icon"
+#define CA_PROP_WINDOW_ICON_NAME                   "window.icon_name"
 #define CA_PROP_WINDOW_X11_DISPLAY                 "window.x11.display"
 #define CA_PROP_WINDOW_X11_XID                     "window.x11.xid"
 #define CA_PROP_APPLICATION_NAME                   "application.name"
@@ -112,22 +119,28 @@
 /* Context object */
 typedef struct ca_context ca_context;
 
-/** Playback completion event callback */
-typedef void ca_finish_callback_t(ca_context *c, uint32_t id, void *userdata);
+/** Playback completion event callback. This callback will be called
+ * from a background thread. */
+typedef void ca_finish_callback_t(ca_context *c, uint32_t id, int error_code, void *userdata);
 
 /** Error codes */
 enum {
     CA_SUCCESS = 0,
-    CA_ERROR_NOT_SUPPORTED = -1,
+    CA_ERROR_NOTSUPPORTED = -1,
     CA_ERROR_INVALID = -2,
     CA_ERROR_STATE = -3,
     CA_ERROR_OOM = -4,
-    CA_ERROR_NO_DRIVER = -5,
+    CA_ERROR_NODRIVER = -5,
     CA_ERROR_SYSTEM = -6,
     CA_ERROR_CORRUPT = -7,
     CA_ERROR_TOOBIG = -8,
-    CA_ERROR_NOT_FOUND = -9,
-    _CA_ERROR_MAX = -10
+    CA_ERROR_NOTFOUND = -9,
+    CA_ERROR_DESTROYED = -10,
+    CA_ERROR_CANCELED = -11,
+    CA_ERROR_NOTAVAILABLE = -12,
+    CA_ERROR_ACCESS = -13,
+    CA_ERROR_IO = -14,
+    _CA_ERROR_MAX = -15
 };
 
 typedef struct ca_proplist ca_proplist;
@@ -141,6 +154,10 @@ int ca_proplist_set(ca_proplist *p, const char *key, const void *data, size_t nb
 /** Create an (unconnected) context object */
 int ca_context_create(ca_context **c);
 
+int ca_context_set_driver(ca_context **c, const char *driver);
+
+int ca_context_change_device(ca_context **c, const char *device);
+
 /** Connect the context. This call is implicitly called if necessary. It
  * is recommended to initialize the application.* properties before
  * issuing this call */
@@ -165,7 +182,8 @@ int ca_context_change_props_full(ca_context *c, ca_proplist *p);
  * at once. It is recommended to pass 0 for the id if the event sound
  * shall never be canceled. If the requested sound is not cached in
  * the server yet this call might result in the sample being uploaded
- * temporarily or permanently. */
+ * temporarily or permanently. This function will only start playback
+ * in the background. It will not wait until playback completed. */
 int ca_context_play(ca_context *c, uint32_t id, ...) CA_GCC_SENTINEL;
 
 /** Play one event sound, and call the specified callback function
@@ -174,7 +192,8 @@ int ca_context_play(ca_context *c, uint32_t id, ...) CA_GCC_SENTINEL;
 int ca_context_play_full(ca_context *c, uint32_t id, ca_proplist *p, ca_finish_callback_t cb, void *userdata);
 
 /** Upload the specified sample into the server and attach the
- * specified properties to it */
+ * specified properties to it. This function will only return after
+ * the sample upload was finished. */
 int ca_context_cache(ca_context *c, ...) CA_GCC_SENTINEL;
 
 /** Upload the specified sample into the server and attach the
diff --git a/common.c b/common.c
index bc535d8..bf11450 100644
--- a/common.c
+++ b/common.c
@@ -68,11 +68,63 @@ int ca_context_destroy(ca_context *c) {
         ca_assert_se(ca_proplist_destroy(c->props) == CA_SUCCESS);
 
     ca_mutex_free(c->mutex);
+    ca_free(c->driver);
+    ca_free(c->device);
     ca_free(c);
 
     return ret;
 }
 
+int ca_context_set_driver(ca_context *c, char *driver) {
+    char *n;
+    int ret;
+
+    ca_return_val_if_fail(c, CA_ERROR_INVALID);
+    ca_mutex_lock(c->mutex);
+    ca_return_val_if_fail_unlock(!c->opened, CA_ERROR_STATE, c->mutex);
+
+    if (!(n = ca_strdup(driver))) {
+        ret = CA_ERROR_OOM;
+        goto fail;
+    }
+
+    ca_free(c->driver);
+    c->driver = n;
+
+    ret = CA_SUCCESS;
+
+fail:
+    ca_mutex_unlock(c->mutex);
+
+    return ret;
+}
+
+int ca_context_change_device(ca_context *c, char *device) {
+    char *n;
+    int ret;
+
+    ca_return_val_if_fail(c, CA_ERROR_INVALID);
+    ca_mutex_lock(c->mutex);
+
+    if (!(n = ca_strdup(device))) {
+        ret = CA_ERROR_OOM;
+        goto fail;
+    }
+
+    ret = c->opened ? driver_change_device(c, n) : CA_SUCCESS;
+
+    if (ret == CA_SUCCESS) {
+        ca_free(c->device);
+        c->device = n;
+    } else
+        ca_free(n);
+
+fail:
+    ca_mutex_unlock(c->mutex);
+
+    return ret;
+}
+
 static int context_open_unlocked(ca_context *c) {
     int ret;
 
@@ -309,3 +361,20 @@ const char *ca_strerror(int code) {
 
     return error_table[-code];
 }
+
+/* Not exported */
+int ca_parse_cache_control(ca_cache_control_t *control, const char *c) {
+    ca_return_val_if_fail(control, CA_ERROR_INVALID);
+    ca_return_val_if_fail(c, CA_ERROR_INVALID);
+
+    if (streq(control, "never"))
+        *control = CA_CACHE_CONTROL_NEVER;
+    else if (streq(control, "permanent"))
+        *control = CA_CACHE_CONTROL_PERMANENT;
+    else if (streq(control, "volatile"))
+        *control = CA_CACHE_CONTROL_VOLATILE;
+    else
+        return CA_ERROR_INVALID;
+
+    return CA_SUCCESS;
+}
diff --git a/common.h b/common.h
index 02e10c2..13139b3 100644
--- a/common.h
+++ b/common.h
@@ -33,7 +33,18 @@ struct ca_context {
 
     ca_proplist *props;
 
+    char *driver;
+    char *device;
+
     void *private;
 };
 
+typedef enum ca_cache_control {
+    CA_CACHE_CONTROL_NEVER,
+    CA_CACHE_CONTROL_PERMANENT,
+    CA_CACHE_CONTROL_VOLATILE
+} ca_cache_control_t;
+
+int ca_parse_cache_control(ca_cache_control_t *control, const char *c);
+
 #endif
diff --git a/driver.h b/driver.h
index 03d4461..732a5d4 100644
--- a/driver.h
+++ b/driver.h
@@ -28,7 +28,9 @@
 int driver_open(ca_context *c);
 int driver_destroy(ca_context *c);
 
+int driver_change_device(ca_context *c, char *device);
 int driver_change_props(ca_context *c, ca_proplist *changed, ca_proplist *merged);
+
 int driver_play(ca_context *c, uint32_t id, ca_proplist *p, ca_finish_callback_t cb, void *userdata);
 int driver_cancel(ca_context *c, uint32_t id);
 int driver_cache(ca_context *c, ca_proplist *p);
diff --git a/pulse.c b/pulse.c
index f5e3996..f74a91d 100644
--- a/pulse.c
+++ b/pulse.c
@@ -20,6 +20,11 @@
   <http://www.gnu.org/licenses/>.
 ***/
 
+/* The locking order needs to be strictly followed! First take the
+ * mainloop mutex, only then take outstanding_mutex if you need both!
+ * Not the other way round, beacause we might then enter a
+ * deadlock!  */
+
 #include <pulse/thread-mainloop.h>
 #include <pulse/context.h>
 #include <pulse/scache.h>
@@ -28,23 +33,202 @@
 #include "common."
 #include "driver.h"
 
+enum outstanding_type {
+    OUTSTANDING_SAMPLE,
+    OUTSTANDING_STREAM,
+    OUTSTANDING_UPLOAD
+};
+
+struct outstanding {
+    PA_LLIST_FIELDS(struct outstanding);
+    enum outstanding_type type;
+    ca_context *context;
+    uint32_t id;
+    uint32_t sink_input;
+    ca_stream *stream;
+    ca_finish_callback_t callback;
+    void *userdata;
+    ca_sound_file *file;
+    int error;
+    ca_bool_t clean_up;
+};
+
 struct private {
     pa_threaded_mainloop *mainloop;
     pa_context *context;
+    ca_theme_data *theme;
+    ca_bool_t subscribed;
+
+    ca_mutex *outstanding_mutex;
+    PA_LLIST_HEAD(struct outstanding, outstanding);
 };
 
 #define PRIVATE(c) ((struct private *) ((c)->private)
 
-static pa_proplist *convert_proplist(ca_proplist *c) {
+static void outstanding_free(struct outstanding *o) {
+    ca_assert(o);
+
+    if (o->file)
+        ca_sound_file_free(o->file);
+
+    if (o->stream) {
+        pa_stream_disconnect(o->stream);
+        pa_stream_unref(o->stream);
+    }
+
+    ca_free(o);
+}
+
+static int convert_proplist(pa_proplist **_l, pa_proplist *c) {
+    pa_proplist *l;
+    ca_prop *i;
+
+    ca_return_val_if_fail(_l, CA_ERROR_INVALID);
+    ca_return_val_if_fail(c, CA_ERROR_INVALID);
+
+    if (!(l = pa_proplist_new()))
+        return CA_ERROR_OOM;
+
+    ca_mutex_lock(c->mutex);
+
+    for (i = c->first_item; i; i = i->next_item)
+        if (pa_proplist_put(l, i->key, CA_PROP_DATA(i), i->nbytes) < 0) {
+            ca_mutex_unlock(c->mutex);
+            pa_proplist_free(l);
+            return PA_ERROR_INVALID;
+        }
+
+    ca_mutex_unlock(c->mutex);
+
+    *_l = l;
+
+    return PA_SUCCESS;
+}
+
+static pa_proplist *strip_canberra_data(pa_proplist *l) {
+    const char *key;
+    void *state = NULL;
+    ca_assert(l);
+
+    while ((key = pa_proplist_iterate(l, &state)))
+        if (strncmp(key, "canberra.", 12) == 0)
+            pa_proplist_remove(l, key);
+
+    return l;
+}
+
+static int translate_error(int error) {
+    static const int table[PA_ERR_MAX] = {
+        [PA_OK]                       = CA_SUCCESS,
+        [PA_ERR_ACCESS]               = CA_ERROR_ACCESS,
+        [PA_ERR_COMMAND]              = CA_ERROR_IO,
+        [PA_ERR_INVALID]              = CA_ERROR_INVALID,
+        [PA_ERR_EXIST]                = CA_ERROR_IO,
+        [PA_ERR_NOENTITY]             = CA_ERROR_NOTFOUND,
+        [PA_ERR_CONNECTIONREFUSED]    = CA_ERROR_NOTAVAILABLE,
+        [PA_ERR_PROTOCOL]             = CA_ERROR_IO,
+        [PA_ERR_TIMEOUT]              = CA_ERROR_IO,
+        [PA_ERR_AUTHKEY]              = CA_ERROR_ACCESS,
+        [PA_ERR_INTERNAL]             = CA_ERROR_IO,
+        [PA_ERR_CONNECTIONTERMINATED] = CA_ERROR_IO,
+        [PA_ERR_KILLED]               = CA_ERROR_DESTROYED,
+        [PA_ERR_INVALIDSERVER]        = CA_ERROR_INVALID,
+        [PA_ERR_MODINITFAILED]        = CA_ERROR_NODRIVER,
+        [PA_ERR_BADSTATE]             = CA_ERROR_STATE,
+        [PA_ERR_NODATA]               = CA_ERROR_IO,
+        [PA_ERR_VERSION]              = CA_ERROR_NOTSUPPORTED,
+        [PA_ERR_TOOLARGE]             = CA_ERROR_TOOBIG
+    };
+
+    ca_assert(error >= 0);
+
+    if (error >= PA_ERR_MAX)
+        return CA_ERROR_IO;
+
+    return table[error];
+}
+
+static void context_state_cb(pa_context *pc, void *userdata) {
+    ca_context *c = userdata;
+    pa_context_state_t state;
+
+    ca_assert(pc);
+    ca_assert(c);
+
+    state = pa_context_get_state(pc);
+
+    if (state == PA_CONTEXT_FAILED || state == PA_CONTEXT_TERMINATED) {
+        struct outstanding *out;
+        int ret;
+
+        ret = translate_error(pa_context_errno(pc));
+
+        ca_mutex_lock(c->outstanding_mutex);
+
+        while ((out = c->outstanding)) {
+
+            PA_LLIST_REMOVE(struct outstanding, c->outstanding, out);
+            ca_mutex_unlock(c->outstanding_mutex);
+
+            if (out->callback)
+                out->callback(c, out->id, ret, out->userdata);
+            outstanding_free(c->outstanding);
+
+            ca_mutex_lock(c->outstanding_mutex);
+        }
+
+        ca_mutex_unlock(c->outstanding_mutex);
+    }
+
+    pa_threaded_mainloop_signal(c->mainloop, FALSE);
+}
+
+static void context_subscribe_cb(pa_context *pc, pa_subscription_event_type_t t, uint32_t idx, void *userdata) {
+    struct outstanding *out, *n;
+    PA_LLIST_HEAD(struct outstanding, l);
+    ca_context *c = userdata;
+
+    ca_assert(pc);
     ca_assert(c);
+
+    if (t != PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_REMOVE)
+        return;
+
+    PA_LLIST_HEAD_INIT(struct outstanding, l);
+
+    ca_mutex_lock(c->outstanding_mutex);
+
+    for (out = c->outstanding; out; out = n) {
+        n = out->next;
+
+        if (out->type != OUTSTANDING_SAMPLE || out->sink_input != idx)
+            continue;
+
+        PA_LLIST_REMOVE(struct outstanding, c->outstanding, out);
+        PA_LLIST_PREPEND(struct outstanding, l, out);
+    }
+
+    ca_mutex_unlock(c->outstanding_mutex);
+
+    while (l) {
+        out = l;
+
+        PA_LLIST_REMOVE(struct outstanding, l, out);
+
+        if (out->callback)
+            out->callback(c, out->id, CA_SUCCESS, out->userdata);
+
+        outstanding_free(out);
+    }
 }
 
 int driver_open(ca_context *c) {
     pa_proplist *l;
     struct private *p;
-    ca_prop *i;
 
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
+    ca_return_val_if_fail(!c->driver || streq(c->driver, "pulse"), PA_ERROR_NO_DRIVER);
+    ca_return_val_if_fail(!PRIVATE(c), PA_ERROR_STATE);
 
     if (!(p = PRIVATE(c) = ca_new0(struct private, 1)))
         return PA_ERROR_OOM;
@@ -54,14 +238,10 @@ int driver_open(ca_context *c) {
         return PA_ERROR_OOM;
     }
 
-    l = pa_proplist_new();
-
-    for (i = c->first_item; i; i = i->next_item)
-        if (pa_proplist_put(l, i->key, CA_PROP_DATA(i), i->nbytes) < 0) {
-            driver_destroy(c);
-            pa_proplist_free(l);
-            return PA_ERROR_INVALID;
-        }
+    if ((ret = convert_proplist(&l, c->proplist))) {
+        driver_destroy(c);
+        return ret;
+    }
 
     if (!(p->context = pa_context_new_with_proplist(pa_threaded_mainloop_get_api(p->mainloop), l))) {
         pa_proplist_free(l);
@@ -72,6 +252,7 @@ int driver_open(ca_context *c) {
     pa_proplist_free(l);
 
     pa_context_set_state_callback(p->context, context_state_cb, c);
+    pa_context_set_subscribe_callback(p->context, context_subscribe_cb, c);
 
     if (pa_context_connect(p->context, NULL, 0, NULL) < 0) {
         int ret = translate_error(pa_context_errno(p->context));
@@ -96,12 +277,16 @@ int driver_open(ca_context *c) {
         return ret;
     }
 
-    return PA_SUCCESS;
+    pa_threaded_mainloop_unlock(p->mainloop);
+
+    return CA_SUCCESS;
 }
 
 int driver_destroy(ca_context *c) {
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
-    ca_assert_se(p = PRIVATE(c));
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
+
+    p = PRIVATE(c);
 
     if (p->mainloop)
         pa_threaded_mainloop_stop(p->mainloop);
@@ -114,29 +299,52 @@ int driver_destroy(ca_context *c) {
     if (p->mainloop)
         pa_threaded_mainloop_free(p->mainloop);
 
+    if (p->theme)
+        ca_theme_data_free(p->theme);
+
+    while (p->outstanding) {
+        struct outstanding *out = p->outstanding;
+        PA_LLIST_REMOVE(struct outstanding, p->outstanding, out);
+
+        if (out->callback)
+            out->callback(c, out->id, CA_ERROR_DESTROYED, out->userdata);
+
+        outstanding_free(out);
+    }
+
     ca_free(p);
 }
 
-int driver_prop_put(ca_context *c, const char *key, const void* data, size_t nbytes) {
+int driver_change_device(ca_context *c, char *device) {
+    ca_return_val_if_fail(c, PA_ERROR_INVALID);
+
+    /* We're happy with any device change. We might however add code
+     * here eventually to move all currently played back event sounds
+     * to the new device. */
+
+    return CA_SUCCESS;
+}
+
+int driver_change_props(ca_context *c, ca_proplist *changed, ca_proplist *merged) {
     struct private *p;
     pa_operation *o;
     pa_proplist *l;
     int ret = CA_SUCCESS;
 
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
-    ca_return_val_if_fail(key, PA_ERROR_INVALID);
+    ca_return_val_if_fail(changed, PA_ERROR_INVALID);
+    ca_return_val_if_fail(merged, PA_ERROR_INVALID);
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
 
     p = PRIVATE(c);
 
     ca_return_val_if_fail(p->mainloop, PA_ERROR_STATE);
     ca_return_val_if_fail(p->context, PA_ERROR_STATE);
 
-    l = pa_proplist_new();
+    if ((ret = convert_proplist(&l, changed)))
+        return ret;
 
-    if (pa_proplist_put(l, key, data, nbytes) < 0) {
-        pa_proplist_free(l);
-        return PA_ERROR_INVALID;
-    }
+    strip_canberra_data(l);
 
     pa_threaded_mainloop_lock(p->mainloop);
 
@@ -150,240 +358,546 @@ int driver_prop_put(ca_context *c, const char *key, const void* data, size_t nby
 
     pa_threaded_mainloop_unlock(p->mainloop);
 
+    pa_proplist_free(l);
+
     return ret;
 }
 
-int driver_prop_unset(ca_context *c, const char *key) {
+static int subscribe(ca_context *c) {
     struct private *p;
     pa_operation *o;
-    const char *a[2];
-    int ret = CA_SUCCESS;
 
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
-    ca_return_val_if_fail(key, PA_ERROR_INVALID);
-
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
     p = PRIVATE(c);
 
     ca_return_val_if_fail(p->mainloop, PA_ERROR_STATE);
     ca_return_val_if_fail(p->context, PA_ERROR_STATE);
-
-    a[0] = key;
-    a[1] = NULL;
+    ca_return_val_if_fail(!p->subscribed, PA_SUCCESS);
 
     pa_threaded_mainloop_lock(p->mainloop);
 
     /* We start these asynchronously and don't care about the return
      * value */
 
-    if (!(o = pa_context_proplist_remove(p->context, a, NULL, NULL)))
+    if (!(o = pa_context_subscribe(p->context, PA_SUBSCRIPTION_MASK_SINK_INPUT, NULL, NULL)))
         ret = translate_error(pa_context_errno(p->context));
     else
         pa_operation_unref(o);
 
     pa_threaded_mainloop_unlock(p->mainloop);
 
+    p->subscribed = TRUE;
+
     return ret;
 }
 
-static pa_proplist* proplist_unroll(va_list ap) {
-    pa_proplist *l;
+static void play_sample_cb(pa_context *c, uint32_t idx, void *userdata) {
+    struct outstanding *out = userdata;
 
-    l = pa_proplist_new();
+    ca_assert(c);
+    ca_assert(out);
 
-    for (;;) {
-        const char *key, *value;
+    if (idx != PA_INVALID_INDEX) {
+        out->error = CA_SUCCESS;
+        out->sink_input = idx;
+    } else
+        out->error = translate_error(pa_context_errno(c));
+}
 
-        if (!(key = va_arg(ap, const char*)))
-            break;
+static void stream_state_cb(pa_stream *s, void *userdata) {
+    struct private *p;
+    struct outstanding *out = userdata;
 
-        if (!(value = va_arg(ap, const char *))) {
-            pa_proplist_free(l);
-            return NULL;
+    ca_assert(s);
+    ca_assert(out);
+
+    p = PRIVATE(out->context);
+
+    if (out->clean_up) {
+        pa_stream_state_t state;
+
+        state = pa_stream_get_state(s);
+
+        if (state == PA_STREAM_FAILED || state == PA_STREAM_TERMINATED) {
+            int err;
+
+            ca_mutex_lock(p->context->outstanding_mutex);
+            PA_LLIST_REMOVE(struct outstanding, c->outstanding, out);
+            ca_mutex_unlock(p->context->outstanding_mutex);
+
+            err = state == PA_STREAM_FAILED ? translate_error(pa_context_errno(pa_stream_get_context(s)))
+
+            if (out->callback)
+                out->callback(c, out->id, ret, out->userdata);
+
+            outstanding_free(c->outstanding);
         }
+    }
 
-        if (pa_proplist_puts(l, key, value) < 0) {
-            pa_proplist_free(l);
-            return NULL;
+    pa_threaded_mainloop_signal(c->mainloop, TRUE);
+}
+
+static void stream_drain_cb(pa_stream *s, int success, void *userdata) {
+    struct private *p;
+    struct outstanding *out = userdata;
+
+    ca_assert(s);
+    ca_assert(out);
+
+    p = PRIVATE(out->context);
+
+    ca_assert(out->type = OUTSTANDING_STREAM);
+    ca_assert(out->clean_up);
+
+    ca_mutex_lock(p->context->outstanding_mutex);
+    PA_LLIST_REMOVE(struct outstanding, c->outstanding, out);
+    ca_mutex_unlock(p->context->outstanding_mutex);
+
+    if (out->callback) {
+        int err;
+
+        err = success ? CA_SUCCESS : translate_error(pa_context_errno(p->context));
+        out->callback(out->context, out->id, err, out->userdata);
+    }
+
+    outstanding_free(out);
+}
+
+static void stream_write_cb(pa_stream *s, size_t bytes, void *userdata) {
+    struct outstanding *out = userdata;
+    struct private *p;
+    void *data;
+    int ret;
+
+    ca_assert(s);
+    ca_assert(bytes > 0);
+    ca_assert(out);
+
+    p = PRIVATE(out->context);
+
+    if (!(data = ca_malloc(bytes))) {
+        ret = CA_ERROR_OOM
+        goto finish;
+    }
+
+    if ((ret = ca_sound_file_read_arbitrary(out->file, data, &bytes)) < 0)
+        goto finish;
+
+    if (bytes > 0) {
+
+        if ((ret = pa_stream_write(p, data, bytes)) < 0) {
+            ret = translate_error(ret);
+            goto finish;
+        }
+
+    } else {
+        /* We reached EOF */
+
+        if (out->type == OUTSTANDING_UPLOAD) {
+
+            if (pa_stream_finish_upload(s) < 0) {
+                ret = translate_error(pa_context_errno(p->context));
+                goto finish;
+            }
+
+            /* Let's just signal driver_cache() which has been waiting for us */
+            pa_threaded_mainloop_signal(c->mainloop, TRUE);
+
+        } else {
+            ca_assert(out->type = OUTSTANDING_STREAM);
+
+            if (!(o = pa_stream_drain(p->context, stream_drain_cb, out))) {
+                ret = translate_error(pa_context_errno(p->context));
+                goto fail;
+            }
+
+            pa_operation_unref(o);
         }
     }
 
-    return l;
+    ca_free(data);
+
+    return;
+
+finish:
+
+    ca_free(data);
+
+    if (out->clean_up) {
+        ca_mutex_lock(p->context->outstanding_mutex);
+        PA_LLIST_REMOVE(struct outstanding, c->outstanding, out);
+        ca_mutex_unlock(p->context->outstanding_mutex);
+
+        if (out->callback)
+            out->callback(out->context, out->id, ret, out->userdata);
+
+        outstanding_free(out);
+    } else {
+        pa_stream_disconnect(p);
+        pa_threaded_mainloop_signal(c->mainloop, TRUE);
+        out->error = ret;
+    }
 }
 
-int driver_play(ca_context *c, uint32_t id, va_list ap) {
+static const pa_sample_format_t sample_type_table[] = {
+    [CA_SAMPLE_S16NE] = PA_SAMPLE_S16NE,
+    [CA_SAMPLE_S16RE] = PA_SAMPLE_S16RE,
+    [CA_SAMPLE_U8] = PA_SAMPLE_U8
+};
+
+int driver_play(ca_context *c, uint32_t id, ca_proplist *proplist, ca_finish_callback_t cb, void *userdata) {
     struct private *p;
-    pa_proplist *l;
-    const char *name;
-    ca_bool_t played = FALSE;
+    pa_proplist *l = NULL;
+    const char *n, *vol, *ct;
+    char *name = NULL;
+    int err = 0;
+    pa_volume_t v = PA_VOLUME_NORM;
+    pa_sample_spec ss;
+    ca_cache_control_t cache_control = CA_CACHE_CONTROL_NEVER;
+    struct outstanding *out = NULL;
+    int try = 3;
 
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
+    ca_return_val_if_fail(proplist, PA_ERROR_INVALID);
+    ca_return_val_if_fail(!userdata || cb, PA_ERROR_INVALID);
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
 
     p = PRIVATE(c);
 
     ca_return_val_if_fail(p->mainloop, PA_ERROR_STATE);
     ca_return_val_if_fail(p->context, PA_ERROR_STATE);
 
-    if (!(l = proplist_unroll(ap)))
-        return PA_ERROR_INVALID;
+    if (!(out = pa_xnew0(struct outstanding, 1))) {
+        ret = PA_ERROR_OOM;
+        goto finish;
+    }
 
-    if (!(name = pa_proplist_gets(l, CA_PROP_EVENT_ID)))
-        if (!(name = ca_context_gets(l, CA_PROP_EVENT_ID))) {
-            pa_proplist_free(l);
-            return PA_ERROR_INVALID;
+    out->type = OUTSTANDING_SAMPLE;
+    out->context = c;
+    out->sink_input = PA_INVALID_INDEX;
+    out->id = id;
+    out->callback = cb;
+    out->userdata = userdata;
+
+    if ((ret = convert_proplist(&l, proplist)))
+        goto finish;
+
+    if (!(n = pa_proplist_gets(l, CA_PROP_EVENT_ID))) {
+        ret = PA_ERROR_INVALID;
+        goto finish;
+    }
+
+    if (!(name = ca_strdup(n))) {
+        ret = PA_ERROR_OOM;
+        goto finish;
+    }
+
+    if ((vol = pa_proplist_gets(l, CA_PROP_CANBERRA_VOLUME))) {
+        char *e = NULL;
+        double dvol;
+
+        errno = 0;
+        dvol = strtod(vol, &e);
+        if (errno != 0 || !e || *e) {
+            ret = PA_ERROR_INVALID;
+            goto finish;
         }
 
-    pa_threaded_mainloop_lock(p->mainloop);
+        v = pa_sw_volume_from_dB(dvol);
+    }
+
+    if ((ct = pa_proplist_gets(l, CA_PROP_CANBERRA_CACHE_CONTROL)))
+        if ((ret = ca_parse_cache_control(&cache_control, ct)) < 0) {
+            ret = PA_ERROR_INVALID;
+            goto finish;
+        }
+
+    strip_canberra_data(l);
 
-    if ((o = pa_context_play_sample_with_proplist(p->context, name, NULL, PA_VOLUME_NORM, l, id, play_success_cb, c))) {
+    if (cb)
+        if ((ret = subscribe(c)) < 0)
+            goto finish;
+
+    for (;;) {
+        pa_threaded_mainloop_lock(p->mainloop);
+
+        /* Let's try to play the sample */
+        if (!(o = pa_context_play_sample_with_proplist(p->context, name, NULL, v, l, id, play_sample_cb, out))) {
+            ret = translate_error(pa_context_errno(p->context));
+            pa_threaded_mainloop_unlock(p->mainloop);
+            goto finish;
+        }
 
         while (pa_operation_get_state(o) != OPERATION_DONE)
             pa_threaded_mainloop_wait(m);
 
         pa_operation_unref(o);
-    }
 
-    pa_threaded_mainloop_unlock(p->mainloop);
+        pa_threaded_mainloop_unlock(p->mainloop);
 
-    if (played)
-        return CA_SUCCESS;
+        /* Did we manage to play the sample or did some other error occur? */
+        if (out->error != CA_ERROR_NOT_FOUND)
+            goto finish;
 
-    va_copy(aq, ap);
-    ret = file_open(&f, c, aq);
-    va_end(aq);
+        /* Hmm, we need to play it directly */
+        if (cache_control == CA_CACHE_CONTROL_NEVER)
+            break;
 
-    if (ret < 0) {
-        pa_proplist_free(l);
-        return ret;
+        /* Don't loop forever */
+        if (--try <= 0)
+            break;
+
+        /* Let's upload the sample and retry playing */
+        if ((ret = driver_cache(c, proplist)) < 0)
+            goto fail;
     }
 
-    pa_threaded_mainloop_lock(p->mainloop);
+    out->type = OUTSTANDING_STREAM;
 
-    s = pa_stream_new_with_proplist(p->context, name, &ss, NULL, l);
-    pa_proplist_free(l);
+    /* Let's stream the sample directly */
+    if ((ret = ca_lookup_sound(&out->file, &p->theme, proplist)) < 0)
+        goto fail;
+
+    ss.channels = sample_type_table[ca_sound_file_get_sample_type(f)];
+    ss.channels = ca_sound_file_get_nchannels(f);
+    ss.rate = ca_sound_file_get_rate(f);
 
-    if (!s) {
+    pa_threaded_mainloop_lock(p->mainloop);
+
+    if (!(out->stream = pa_stream_new_with_proplist(p->context, name, &ss, NULL, l))) {
         ret = translate_error(pa_context_errno(p->context));
-        file_close(f);
         pa_threaded_mainloop_unlock(p->mainloop);
-        return ret;
+        goto fail;
     }
 
-    pa_stream_set_state_callback(p->stream, stream_state_cb, c);
-    pa_stream_set_write_callback(p->stream, stream_request_cb, c);
+    pa_stream_set_userdata(p->stream, out);
+    pa_stream_set_state_callback(p->stream, stream_state_cb, out);
+    pa_stream_set_write_callback(p->stream, stream_request_cb, out);
 
     if (pa_stream_connect_playback(s, NULL, NULL, 0, NULL, NULL) < 0) {
         ret = translate_error(pa_context_errno(p->context));
-        file_close(f);
-        pa_stream_disconnect(s);
-        pa_stream_unref(s);
         pa_threaded_mainloop_unlock(p->mainloop);
-        return ret;
+        goto fail;
     }
 
-    while (!done) {
+    for (;;) {
+        pa_stream_state state = pa_stream_get_state(s);
 
-        if (pa_stream_get_state(s) != PA_STREAM_READY ||
-            pa_context_get_state(c) != PA_CONTEXT_READY) {
+        /* Stream sucessfully created */
+        if (state == PA_STREAM_READY)
+            break;
 
+        /* Check for failure */
+        if (state == PA_STREAM_FAILED) {
             ret = translate_error(pa_context_errno(p->context));
-            file_close(f);
-            pa_stream_disconnect(s);
-            pa_stream_unref(s);
             pa_threaded_mainloop_unlock(p->mainloop);
-            return ret;
+            goto fail;
+        }
+
+        if (state == PA_STREAM_TERMINATED) {
+            ret = out->error;
+            pa_threaded_mainloop_unlock(p->mainloop);
+            goto fail;
         }
 
         pa_threaded_mainloop_wait(p->mainloop);
     }
 
-    pa_stream_disconnect(s);
-    pa_stream_unref(s);
+    if ((out->sink_input = pa_stream_get_index(s)) == PA_INVALID_INDEX) {
+        ret = translate_error(pa_context_errno(p->context));
+        pa_threaded_mainloop_unlock(p->mainloop);
+        goto fail;
+    }
 
-    return CA_SUCCESS;
+    pa_threaded_mainloop_unlock(p->mainloop);
+
+    ret = CA_SUCCESS;
+
+finish:
+
+    /* We keep the outstanding struct around if we need clean up later to */
+    if (ret == CA_SUCCESS && (out->type == OUTSTANDING_STREAM || cb)) {
+        out->clean_up = TRUE;
+
+        pa_mutex_lock(p->outstanding_mutex);
+        PA_LLIST_PREPEND(struct outstanding, p->outstanding, out);
+        pa_mutex_unlock(p->outstanding_mutex);
+    } else
+        outstanding_free(out);
+
+    if (l)
+        pa_proplist_free(l);
+
+    ca_free(name);
 
+    return ret;
 }
 
 int driver_cancel(ca_context *c, uint32_t id) {
+    struct private *p;
+    pa_operation *o;
+    int ret = CA_SUCCESS;
+    struct outstanding *out, *n;
+
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
+
+    p = PRIVATE(c);
+
+    ca_return_val_if_fail(p->mainloop, PA_ERROR_STATE);
+    ca_return_val_if_fail(p->context, PA_ERROR_STATE);
+
+    pa_threaded_mainloop_lock(p->mainloop);
+
+    ca_mutex_lock(p->outstanding_mutex);
+
+    /* We start these asynchronously and don't care about the return
+     * value */
+
+    for (out = p->outstanding; out; out = n) {
+        int ret2;
+        n = out->next;
+
+        if (out->type == OUTSTANDING_UPLOAD ||
+            out->id != id ||
+            out->sink_input == PA_INVALID_INDEX)
+            continue;
+
+        if (!(o = pa_context_kill_sink_input(p->context, out->sink_input, NULL, NULL)))
+            ret2 = translate_error(pa_context_errno(p->context));
+        else
+            pa_operation_unref(o);
+
+        /* We make sure here to kill all streams identified by the id
+         * here. However, we will return only the first error we
+         * encounter */
+
+        if (ret2 && ret == CA_SUCCESS)
+            ret = ret2;
+
+        if (out->callback)
+            out->callback(c, out->id, CA_ERROR_CANCELED, out->userdata);
+
+        CA_LLIST_REMOVE(struct outstanding, p->outstanding, out);
+        outstanding_free(out);
+    }
+
+    ca_mutex_unlock(p->outstanding_mutex);
 
+    pa_threaded_mainloop_unlock(p->mainloop);
+
+    return ret;
 }
 
-int driver_cache(ca_context *c, va_list ap) {
+int driver_cache(ca_context *c, ca_proplist *proplist) {
     struct private *p;
-    pa_proplist *l;
-    ca_file *f;
-    int ret;
-    va_list aq;
-    pa_stream *s;
-    const char *name;
+    pa_proplist *l = NULL;
+    const char *n, *ct;
+    char *name = NULL;
+    pa_sample_spec ss;
+    ca_cache_control_t cache_control = CA_CACHE_CONTROL_NEVER;
+    struct outstanding out;
 
     ca_return_val_if_fail(c, PA_ERROR_INVALID);
+    ca_return_val_if_fail(proplist, PA_ERROR_INVALID);
+    ca_return_val_if_fail(c->private, PA_ERROR_STATE);
 
     p = PRIVATE(c);
 
     ca_return_val_if_fail(p->mainloop, PA_ERROR_STATE);
     ca_return_val_if_fail(p->context, PA_ERROR_STATE);
 
-    if (!(l = proplist_unroll(ap)))
-        return PA_ERROR_INVALID;
+    if (!(out = ca_new(struct outstanding, 1))) {
+        ret = CA_ERROR_OOM;
+        goto finish;
+    }
 
-    if (!(name = pa_proplist_gets(l, CA_PROP_EVENT_ID)))
-        if (!(name = ca_context_gets(l, CA_PROP_EVENT_ID)))  {
-            pa_proplist_free(l);
-            return PA_ERROR_INVALID;
+    out->type = OUTSTANDING_UPLOAD;
+    out->context = c;
+    out->sink_input = PA_INVALID_INDEX;
+
+    if ((ret = convert_proplist(&l, proplist)))
+        goto finish;
+
+    if (!(n = pa_proplist_gets(l, CA_PROP_EVENT_ID))) {
+        ret = PA_ERROR_INVALID;
+        goto finish;
+    }
+
+    if (!(name = ca_strdup(n))) {
+        ret = PA_ERROR_OOM;
+        goto finish;
+    }
+
+    if ((ct = pa_proplist_gets(l, CA_PROP_CANBERRA_CACHE_CONTROL)))
+        if ((ret = ca_parse_cache_control(&cache_control, ct)) < 0) {
+            ret = PA_ERROR_INVALID;
+            goto finish;
         }
 
-    va_copy(aq, ap);
-    ret = file_open(&f, c, aq);
-    va_end(aq);
+    strip_canberra_data(l);
 
-    if (ret < 0) {
-        pa_proplist_free(l);
-        return ret;
+    if (ct == CA_CACHE_CONTROL_NEVER) {
+        ret = PA_ERROR_INVALID;
+        goto finish;
     }
 
-    pa_threaded_mainloop_lock(p->mainloop);
+    /* Let's stream the sample directly */
+    if ((ret = ca_lookup_sound(&out->file, &p->theme, proplist)) < 0)
+        goto fail;
 
-    s = pa_stream_new_with_proplist(p->context, name, &ss, NULL, l);
-    pa_proplist_free(l);
+    ss.channels = sample_type_table[ca_sound_file_get_sample_type(out->file)];
+    ss.channels = ca_sound_file_get_nchannels(out->file);
+    ss.rate = ca_sound_file_get_rate(out->file);
+
+    pa_threaded_mainloop_lock(p->mainloop);
 
-    if (!s) {
+    if (!(out->stream = pa_stream_new_with_proplist(p->context, name, &ss, NULL, l))) {
         ret = translate_error(pa_context_errno(p->context));
-        file_close(f);
         pa_threaded_mainloop_unlock(p->mainloop);
-        return ret;
+        goto fail;
     }
 
-    pa_stream_set_state_callback(p->stream, stream_state_cb, c);
-    pa_stream_set_write_callback(p->stream, stream_request_cb, c);
+    pa_stream_set_userdata(out->stream, out);
+    pa_stream_set_state_callback(out->stream, stream_state_cb, out);
+    pa_stream_set_write_callback(out->stream, stream_request_cb, out);
 
-    if (pa_stream_connect_upload(s, f->nbytes) < 0) {
+    if (pa_stream_connect_upload(s, ca_sound_file_get_size(out->file)) < 0) {
         ret = translate_error(pa_context_errno(p->context));
-        file_close(f);
-        pa_stream_disconnect(s);
-        pa_stream_unref(s);
         pa_threaded_mainloop_unlock(p->mainloop);
-        return ret;
+        goto fail;
     }
 
-    while (!done) {
+    for (;;) {
+        pa_stream_state state = pa_stream_get_state(s);
 
-        if (pa_stream_get_state(s) != PA_STREAM_READY ||
-            pa_context_get_state(c) != PA_CONTEXT_READY) {
+        /* Stream sucessfully created and uploaded */
+        if (state == PA_STREAM_TERMINATED)
+            break;
 
+        /* Check for failure */
+        if (state == PA_STREAM_FAILED) {
             ret = translate_error(pa_context_errno(p->context));
-            file_close(f);
-            pa_stream_disconnect(s);
-            pa_stream_unref(s);
             pa_threaded_mainloop_unlock(p->mainloop);
-            return ret;
+            goto fail;
         }
 
         pa_threaded_mainloop_wait(p->mainloop);
     }
 
-    pa_stream_disconnect(s);
-    pa_stream_unref(s);
+    pa_threaded_mainloop_unlock(p->mainloop);
+
+    ret = CA_SUCCESS;
 
-    return CA_SUCCESS;
+finish:
+
+    outstanding_free(out);
+
+    if (l)
+        pa_proplist_free(l);
+
+    ca_free(name);
+
+    return ret;
 }
diff --git a/read-sound-file.c b/read-sound-file.c
index 13066b1..8670445 100644
--- a/read-sound-file.c
+++ b/read-sound-file.c
@@ -135,3 +135,41 @@ int ca_sound_file_read_uint8(ca_sound_file *fn, uint8_t *d, unsigned *n) {
     if (f->wav)
         return ca_wav_read_u8(f->wav, d, n);
 }
+
+int ca_sound_file_read_arbitrary(ca_sound_file *f, void *d, size_t *n) {
+    int ret;
+
+    ca_return_val_if_fail(f, CA_ERROR_INVALID);
+    ca_return_val_if_fail(d, CA_ERROR_INVALID);
+    ca_return_val_if_fail(n, CA_ERROR_INVALID);
+    ca_return_val_if_fail(*n > 0, CA_ERROR_INVALID);
+    ca_return_val_if_fail(f->wav && !f->vorbis, CA_ERROR_STATE);
+
+    switch (f->type) {
+        case CA_SAMPLE_S16NE:
+        case CA_SAMPLE_S16RE: {
+            unsigned k;
+
+            k = *n / sizeof(int16_t);
+            if ((ret = ca_sound_file_read_int16(f, d, &k)) == CA_SUCCESS)
+                *n = k * sizeof(int16_t);
+
+            break;
+        }
+
+        case CA_SAMPLE_S16RE: {
+            unsigned k;
+
+            k = *n;
+            if ((ret = ca_sound_file_read_uint8(f, d, &k)) == CA_SUCCESS)
+                *n = k;
+
+            break;
+        }
+
+        default:
+            ca_assert_not_reached();
+    }
+
+    return ret;
+}
diff --git a/read-sound-file.h b/read-sound-file.h
index 8dcc1e0..5ab7e66 100644
--- a/read-sound-file.h
+++ b/read-sound-file.h
@@ -38,7 +38,11 @@ unsigned ca_sound_file_get_nchannels(ca_sound_file *f);
 unsigned ca_sound_file_get_rate(ca_sound_file *f);
 ca_sample_type_t ca_sound_file_get_sample_type(ca_sound_file *f);
 
+size_t ca_sound_file_get_size(ca_sound_file *f);
+
 int ca_sound_file_read_int16(ca_sound_file *f, int16_t *d, unsigned *n);
 int ca_sound_file_read_uint8(ca_sound_file *f, uint8_t *d, unsigned *n);
 
+int ca_sound_file_read_arbitrary(ca_sound_file *f, void *d, size_t *n);
+
 #endif
diff --git a/test-gtk.c b/test-gtk.c
new file mode 100644
index 0000000..71d8aac
--- /dev/null
+++ b/test-gtk.c
@@ -0,0 +1,18 @@
+
+
+static void trigger(GtkWidget *w) {
+
+    ca_context_play(
+            gca_default_context(),
+            CA_PROP_EVENT_ID, "clicked",
+            GCA_PROPS_FOR_WIDGET(w),
+            GCA_PROPS_FOR_MOUSE_EVENT(e),
+            GCA_PROPS_FOR_APP(),
+            NULL);
+
+}
+
+int main(int argc, char *argv[]) {
+
+    ca_
+}