#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);
}