implemented completely, testing required

git-svn-id: file:///home/lennart/svn/public/xmms-pulse/trunk@5 ef929aba-56e2-0310-84e0-b7573d389508

1 files changed, 263 insertions, 5 deletions

diff --git a/src/thread.c b/src/thread.c
index 0384f49..4f9b515 100644
--- a/src/thread.c
+++ b/src/thread.c
@@ -1,14 +1,28 @@
+#include <pthread.h>
+
 #include <polyp/mainloop.h>
 
 static pthread_cond_t request_cond = PTHREAD_COND_INITIALIZER;
 static pthread_mutex_t request_mutex = PTHREAD_MUTEX_INITIALIZER;
 
 struct request {
-    enum { MESSAGE_OPEN, MESSAGE_CLOSE, MESSAGE_WRITE, MESSAGE_FLUSH, MESSAGE_PAUSE } type;
+    enum {
+        MESSAGE_OPEN,
+        MESSAGE_CLOSE,
+        MESSAGE_WRITE,
+        MESSAGE_FLUSH,
+        MESSAGE_PAUSE,
+        MESSAGE_UNPAUSE,
+        MESSAGE_LATENCY,
+        MESSAGE_WRITABLE,
+        MESSAGE_TRIGGER,
+    } type;
     void *data;
     struct pa_sample_spec ss;
     size_t length;
     int success, done;
+    uint32_t value;
+    pa_volume_t volume;
 };
 
 static struct request* current_request = NULL;
@@ -21,6 +35,10 @@ static struct pa_context *context = NULL;
 static struct pa_stream *stream = NULL;
 static struct pa_mainloop_api *mainloop_api = NULL;
 static int failed = 0;
+static pa_volume_t volume = PA_VOLUME_NORM;
+static size_t written = 0;
+static pa_usec_t latency;
+static int do_trigger;
 
 static void finish_request(int success) {
     failed = 1;
@@ -36,6 +54,24 @@ static void finish_request(int success) {
     pthread_mutex_unlock(&request_mutex);
 }
 
+static void info_callback(struct pa_context *c, const struct pa_sink_input_info *i, int is_last, void *userdata) {
+    assert(c && c == context);
+
+    if (!i)
+        return;
+
+    volume = i->volume;
+}
+
+static void subscribe_callback(struct pa_context *c, enum pa_subscription_event_type t, uint32_t index, void *userdata) {
+    assert(c && c == context);
+
+    if (!stream || index != pa_stream_get_index(stream) || t != (PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_CHANGE))
+        return;
+
+    pa_operation_unref(pa_context_get_sink_input_info(c, index, info_callback, NULL);
+}
+
 static void stream_state_callback(struct pa_stream *s, void *userdata) {
     assert(stream == s);
     
@@ -44,6 +80,7 @@ static void stream_state_callback(struct pa_stream *s, void *userdata) {
             break;
         case PA_STREAM_READY:
             assert(current_request && current_request->type == MESSAGE_OPEN);
+            pa_operation_unref(pa_context_get_sink_input_info(context, pa_stream_get_index(s), info_callback, NULL);
             finish_request(1);
             break;
         default:
@@ -52,7 +89,7 @@ static void stream_state_callback(struct pa_stream *s, void *userdata) {
 }
 
 static void context_state_callback(struct pa_context *c, void *userdata) {
-    assert(c == context);
+    assert(c && c == context);
 
     switch (pa_context_get_state(c)) {
         case PA_CONTEXT_CONNECTING:
@@ -62,6 +99,9 @@ static void context_state_callback(struct pa_context *c, void *userdata) {
 
         case PA_CONTEXT_READY :
             assert(!stream && current_request);
+            pa_context_set_subscribe_callback(context, subscribe_callback, NULL);
+            pa_operation_unref(pa_context_subscribe(context, PA_SUBSCRIPTION_MASK_SINK_INPUT, NULL, NULL);
+            
             stream = pa_stream_new(c, &current_request->ss);
             assert(stream);
 
@@ -75,10 +115,27 @@ static void context_state_callback(struct pa_context *c, void *userdata) {
 }
 
 static void context_drain_callback(struct pa_context *c, void *userdata) {
-    assert(c == context);
+    assert(c && c == context);
     mainloop_api->quit(mainloop_api, 0);
 }
 
+static void stream_success_callback(struct pa_stream *s, int success, void *userdata) {
+    assert(s == stream && s);
+    finish_request(!!success);
+}
+
+static void context_success_callback(struct pa_context *c, int success, void *userdata) {
+    assert(c == context && c);
+    finish_request(!!success);
+}
+
+static void latency_callback(struct pa_stream *s, pa_usec_t latency, void *userdata) {
+    assert(s == stream && s);
+    assert(current_request && current_request>type == MESSAGE_LATENCY);
+    current_request->value = latency;
+    finish_request(latency != (pa_usec_t) -1);
+}
+
 static void request_func(struct pa_mainloop*api, struct pa_io_event *io, enum pa_io_event_flags f, void *userdata) {
     char x;
     
@@ -114,11 +171,48 @@ static void request_func(struct pa_mainloop*api, struct pa_io_event *io, enum pa
                 case MESSAGE_WRITE:
                     assert(context && stream && current_request->data && current_request->length > 0);
                     pa_stream_write(stream, current_request->data, current_request->length, free, 0);
+                    current_request->data = NULL;
+                    finish_request(1);
                     break;
 
                 case MESSAGE_FLUSH:
                     assert(context && stream);
-                    pa_stream_flush
+                    pa_operation_unref(pa_stream_flush(stream, stream_success_callback, NULL));
+                    break;
+
+                case MESSAGE_PAUSE:
+                case MESSAGE_UNPAUSE:
+                    assert(context && stream);
+                    pa_operation_unref(pa_stream_cork(stream, current_request->type == MESSAGE_UNPAUSE, stream_success_callback, NULL));
+                    break;
+
+                case MESSAGE_LATENCY:
+                    assert(context && stream);
+                    pa_operation_unref(pa_stream_get_latency(stream, latency_callback, NULL));
+                    break;
+
+                case MESSAGE_WRITABLE:
+                    assert(context && stream);
+                    current_request->value = pa_stream_writable_size(stream);
+                    finish_request(1);
+                    break;
+
+                case MESSAGE_GETVOLUME:
+                    assert(context && stream);
+                    current_request->volume = volume;
+                    finish_request(1);
+                    break;
+
+                case MESSAGE_SETVOLUME:
+                    assert(context && stream);
+                    pa_operation_unref(pa_context_set_sink_input_volume(context, pa_stream_get_index(stream), current_request->volume, context_success_callback, NULL));
+                    break;
+
+                case MESSAGE_TRIGGER:
+                    assert(context && stream);
+                    pa_operation_unref(pa_stream_trigger(stream, stream_success_callback, NULL));
+                    break;
+                                       
             }
         }
     }
@@ -179,8 +273,27 @@ static void stop_thread(void) {
     pipe_fds[0] = pipe_fds[1] = -1;
 }
     
+static void polyp_get_volume(int *l, int *r) {
+    struct request r;
+    int v;
 
-void request_execute(struct request *r) {
+    r.message = MESSAGE_GET_VOLUME;
+    request_execute(&r);
+
+    v = (r.volume*100)/PA_VOLUME_NORM;
+    
+    *r = *l = v > 100 ? 100 : v;
+}
+
+void polyp_set_volume(int l, int r) {
+    struct request r;
+
+    r.message = MESSAGE_SET_VOLUME;
+    r.volume = ((l+r)*PA_VOLUME_NORM)/200;
+    request_execute(&r);
+}
+
+static void request_execute(struct request *r) {
     char x = 'x';
     assert(r);
     
@@ -201,3 +314,148 @@ void request_execute(struct request *r) {
     pthread_mutex_unlock(&request_mutex);
 }
 
+static void polyp_pause(short b) {
+    struct request r;
+    
+    r.message = b ? MESSAGE_PAUSE : MESSAGE_UNPAUSE;
+    request_execute(&r);
+}
+
+static int polyp_free(void) {
+    int ret;
+    struct request r;
+    r.message = MESSAGE_WRITABLE;
+    request_execute(&r);
+
+    ret = r.value;
+    
+    if (do_trigger) {
+        r.message = MESSAGE_TRIGGER;
+        request_execute(&r);
+    }
+    
+    do_trigger = 1;
+    return ret;
+}
+
+static int polyp_playing(void) {
+    struct request r;
+    r.message = MESSAGE_LATENCY;
+    request_execute(&r);
+
+    return r.value != 0;
+}
+
+static int polyp_get_written_time(void) {
+    return ((written/pa_frame_size(&sample_spec))*1000)/sample_spec.rate;
+}
+
+static int polyp_get_output_time(void) {
+    int t, ms;
+    struct request r;
+    r.message = MESSAGE_LATENCY;
+    request_execute(&r);
+
+    t = polyp_get_output_time();
+    ms = r.value/1000;
+
+    if (ms > t)
+        return 0;
+    else
+        return t-ms;
+}
+
+static void polyp_flush(int time) {
+    struct request r;
+    r.message = MESSAGE_FLUSH;
+    request_execute(&r);
+
+    written = (time*sample_spec.rate/1000)*pa_frame_size(&sample_spec);
+}
+
+static void polyp_write(void* ptr, int length) {
+    struct request r;
+    r.message = MESSAGE_WRITE;
+    r.data = ptr;
+    r.length = length;
+    
+    request_execute(&r);
+
+    written += length;
+    do_trigger = 0;
+}
+
+static int polyp_open(AFormat fmt, int rate, int nch) {
+    struct request r;
+
+    if (fmt == FMT_U8)
+        r.ss.format = PA_SAMPLE_U8;
+    else if (fmt == FMT_S16_LE)
+        r.ss.format = PA_SAMPLE_S16LE;
+    else if (fmt == FM_S16_BE)
+        r.ss.format = PA_SAMPLE_S16BE;
+    else if (fmt == FM_S16_NE)
+        r.ss.format = PA_SAMPLE_S16NE;
+    else
+        return 0;
+
+    r.ss.rate = rate;
+    r.ss.channels = nch;
+
+    if (!pa_sample_spec_valid(&r.ss))
+        return 0;
+
+    start_thread();
+
+    r.message = MESSAGE_OPEN;
+    request_execute(&r);
+
+    if (!r->success) {
+        stop_thread();
+        return 0;
+    }
+
+    written = do_trigger = 0;
+    
+    return 1;
+}
+
+static void polyp_close(void) {
+    struct request r;
+
+    assert(thread_running);
+
+    r.message = MESSAGE_CLOSE;
+    request_execute(&r);
+
+    stop_thread();
+}
+
+
+static void polyp_init(void) {
+}
+
+static OutputPlugin polyp_plugin = {
+    NULL,
+    NULL,
+    "Polypaudio Output Plugin", /* Description */
+    polyp_init,                            /* done */
+    NULL, /* polyp_about, */
+    NULL, /* polyp_configure, */
+    polyp_get_volume,
+    polyp_set_volume,
+    
+    polyp_open,                            /* done */
+    polyp_write,                           /* done */
+    polyp_close,                           /* done */
+    polyp_flush,                           /* done */
+    polyp_pause,                           /* done */
+    polyp_free,                            /* done */
+    polyp_playing,                         /* done */
+    polyp_get_output_time,                 /* done */
+    polyp_get_written_time,                /* done */
+};
+
+OutputPlugin *get_oplugin_info(void) {
+    return &polyp_plugin;
+}