add some more unfinished stuff

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

1 files changed, 203 insertions, 0 deletions

diff --git a/src/thread.c b/src/thread.c
new file mode 100644
index 0000000..0384f49
--- /dev/null
+++ b/src/thread.c
@@ -0,0 +1,203 @@
+#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;
+    void *data;
+    struct pa_sample_spec ss;
+    size_t length;
+    int success, done;
+};
+
+static struct request* current_request = NULL;
+
+static int pipe_fds[2] = { -1, -1};
+
+static pthread_t thread_id;
+static int thread_running = 0;
+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 void finish_request(int success) {
+    failed = 1;
+
+    pthread_mutex_lock(&request_mutex);
+
+    if (current_request) {
+        current_request->done = 1;
+        current_request->success = success;
+        pthread_cond_signal(&request_cond);
+    }
+
+    pthread_mutex_unlock(&request_mutex);
+}
+
+static void stream_state_callback(struct pa_stream *s, void *userdata) {
+    assert(stream == s);
+    
+    switch(pa_stream_get_state(c)) {
+        case PA_STREAM_CREATING:
+            break;
+        case PA_STREAM_READY:
+            assert(current_request && current_request->type == MESSAGE_OPEN);
+            finish_request(1);
+            break;
+        default:
+            finish_request(0);
+    }
+}
+
+static void context_state_callback(struct pa_context *c, void *userdata) {
+    assert(c == context);
+
+    switch (pa_context_get_state(c)) {
+        case PA_CONTEXT_CONNECTING:
+        case PA_CONTEXT_AUTHORIZING:
+        case PA_CONTEXT_SETTING_NAME:
+            break;
+
+        case PA_CONTEXT_READY :
+            assert(!stream && current_request);
+            stream = pa_stream_new(c, &current_request->ss);
+            assert(stream);
+
+            pa_stream_set_state(stream, stream_state_callback);
+            pa_stream_connect_playback(stream, NULL, NULL);
+            break;
+        
+        default:
+            finish_request(0);
+    }
+}
+
+static void context_drain_callback(struct pa_context *c, void *userdata) {
+    assert(c == context);
+    mainloop_api->quit(mainloop_api, 0);
+}
+
+static void request_func(struct pa_mainloop*api, struct pa_io_event *io, enum pa_io_event_flags f, void *userdata) {
+    char x;
+    
+    assert(api && io && f == PA_IO_EVENT_INPUT);
+
+    read(pipe_fds[0], &x, 1);
+    
+    pthread_mutex_lock(&request_mutex);
+
+    if (current_request) {
+        if (failed) {
+            fail();
+        } else {
+            switch (current_request->type) {
+                case MESSAGE_OPEN:
+                    assert(!context && !stream);
+                    context = pa_context_new(api, "xmms");
+                    assert(context);
+                    pa_context_set_state_callback(context, context_state_callback, NULL);
+                    pa_context_connect(context, NULL);
+                    break;
+
+                case MESSAGE_CLOSE: {
+                    struct pa_operation *o;
+                    assert(context);
+                    if ((o = pa_context_drain(context, context_drain_callback, NULL)))
+                        pa_operation_unref(o);
+                    else
+                        api->quit(mainloop_api, 0);
+                    break;
+                }
+                    
+                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);
+                    break;
+
+                case MESSAGE_FLUSH:
+                    assert(context && stream);
+                    pa_stream_flush
+            }
+        }
+    }
+    
+    pthread_mutex_unlock(&request_mutex);
+}
+
+static void* thread_func(void *t) {
+    struct pa_mainloop *m;
+    struct pa_io_event *io;
+
+    assert(pipe_fds[0] >= 0 && !mainloop_api);
+
+    failed = 0;
+    
+    m = pa_mainloop_new();
+    assert(m);
+    mainloop_api = pa_mainloop_get_api(m);
+    assert(mainloop_api);
+
+    io = mainloop_api->io_new(mainloop_api, pipe_fds[0], PA_IO_EVENT_INPUT, &request_func, NULL);
+    assert(io);
+    
+    pa_mainloop_run(m, NULL);
+
+    api->io_free(io);
+    pa_mainloop_free(m);
+
+    mainloop_api = NULL;
+    
+    return NULL;
+}
+
+static void start_thread(void) {
+    int r;
+    assert(!thread_running);
+
+    r = pipe(pipe_fds);
+    assert(r >= 0 && pipe_fds[0] >= 0 && pipe_fds[1] >= 0);
+
+    current_request = NULL;
+
+    r = pthread_create(&thread_id, NULL, thread_func, NULL);
+    assert(!r);
+}
+
+static void stop_thread(void) {
+    struct request req;
+    assert(thread_running);
+
+    pthread_join(thread_id, NULL);
+
+    thread_running = 0;
+    assert(!current_request);
+
+    close(pipe_fds[0]);
+    close(pipe_fds[1]);
+    pipe_fds[0] = pipe_fds[1] = -1;
+}
+    
+
+void request_execute(struct request *r) {
+    char x = 'x';
+    assert(r);
+    
+    r->success = r->done = 0;
+    
+    pthread_mutex_lock(&request_mutex);
+    assert(!current_request);
+    current_request = r;
+
+    assert(pipe_fds[1] >= 0);
+    write(pipe_fds[1], &x, sizeof(x));
+    
+    while (!r->done)
+        pthread_cond_wait(&request_cond, &request_mutex);
+
+    current_request = NULL;
+    
+    pthread_mutex_unlock(&request_mutex);
+}
+