/* $Id$ */

#include <stdio.h>
#include <assert.h>
#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>

#include <plugin.h>
#include <polyp/mainloop.h>
#include <polyp/polyplib.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,
        MESSAGE_UNPAUSE,
        MESSAGE_LATENCY,
        MESSAGE_WRITABLE,
        MESSAGE_TRIGGER,
        MESSAGE_GETVOLUME,
        MESSAGE_SETVOLUME
    } message;
    void *data;
    struct pa_sample_spec ss;
    size_t length;
    int success, done;
    uint32_t value;
    pa_usec_t latency;
    pa_volume_t volume;
};

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 pa_volume_t volume = PA_VOLUME_NORM;
static size_t written = 0;
static int do_trigger = 0, triggered = 0;
static struct pa_sample_spec sample_spec;
static int locked_by_thread = 0;

static void *memdup(void *p, size_t l) {
    void *r = malloc(l);
    memcpy(r, p, l);
    return r;
}

static void lock_request(void) {
    assert(thread_running && pthread_equal(pthread_self(), thread_id));
    
    if (!(locked_by_thread++))
        pthread_mutex_lock(&request_mutex);
}

static void unlock_request(void) {
    assert(thread_running && pthread_equal(pthread_self(), thread_id) && locked_by_thread > 0);

    if (!(--locked_by_thread))
        pthread_mutex_unlock(&request_mutex);
}

static void finish_request(int success) {
    if (!success)
        failed = 1;

    lock_request();

    if (current_request) {
        if (!success)
            fprintf(stderr, "failure: %i\n", success);
        current_request->done = 1;
        current_request->success = success;
        pthread_cond_broadcast(&request_cond);
    }
    
    unlock_request();
}

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);
    
    switch(pa_stream_get_state(s)) {
        case PA_STREAM_CREATING:
            break;
        case PA_STREAM_READY:
            assert(current_request && current_request->message == 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:
            finish_request(0);
    }
}

static void context_state_callback(struct pa_context *c, void *userdata) {
    assert(c && c == context);

    fprintf(stderr, "state\n");

    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);
            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, "xmms", &current_request->ss);
            assert(stream);

            pa_stream_set_state_callback(stream, stream_state_callback, NULL);
            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 && c == context);
    mainloop_api->quit(mainloop_api, 0);
    finish_request(1);
}

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, const struct pa_latency_info* i, void *userdata) {
    assert(s == stream && s);
    assert(current_request && current_request->message == MESSAGE_LATENCY);
    if (i) {
        current_request->latency = i->buffer_usec + i->sink_usec;
        current_request->value = i->playing;
    }
    finish_request(!!i);
}

static void request_func(struct pa_mainloop_api*api, struct pa_io_event *io, int fd, enum pa_io_event_flags f, void *userdata) {
    char x;

    /*fprintf(stderr, "REQUEST %p\n", current_request);*/

    assert(api && io && f == PA_IO_EVENT_INPUT);

    read(fd, &x, 1);
    
    lock_request();

    if (current_request) {
        if (failed && current_request->message != MESSAGE_CLOSE) {
            finish_request(0);
        } else {

            /*fprintf(stderr, "req: %i\n", current_request->message);*/
            
            switch (current_request->message) {
                case MESSAGE_OPEN:
                    assert(!context && !stream);

                    fprintf(stderr, "opening\n");
                    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;

                    fprintf(stderr, "closing\n");
                    
                    assert(context);
                    if ((o = pa_context_drain(context, context_drain_callback, NULL)))
                        pa_operation_unref(o);
                    else {
                        api->quit(mainloop_api, 0);
                        finish_request(1);
                    }
                    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);
                    current_request->data = NULL;
                    finish_request(1);
                    break;

                case MESSAGE_FLUSH:
                    assert(context && stream);
                    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->message == MESSAGE_UNPAUSE, stream_success_callback, NULL));
                    break;

                case MESSAGE_LATENCY:
                    if (!context || !stream) {
                        current_request->latency = 0;
                        finish_request(1);
                    } else
                        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);

                    if (current_request->volume == volume)
                        finish_request(1);
                    else
                        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;
                                       
            }
        }
    }
    
    unlock_request();

    /*fprintf(stderr, "REQ_DONE\n");*/
}

static void* thread_func(void *t) {
    struct pa_mainloop *m;
    struct pa_io_event *io;

    assert(pipe_fds[0] >= 0 && !mainloop_api);

    fprintf(stderr, "THREAD START\n");
    
    failed = locked_by_thread = 0;
    thread_running = 1;
    
    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);

    if (context) {
        pa_context_unref(context);
        context = NULL;
    }

    if (stream) {
        pa_stream_unref(stream);
        stream = NULL;
    }

    mainloop_api->io_free(io);
    pa_mainloop_free(m);

    mainloop_api = NULL;
    thread_running = 0;

    pthread_cond_broadcast(&request_cond);

    fprintf(stderr, "THREAD STOP\n");
    
    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);

    thread_running = 1;
}

static void stop_thread(void) {
    fprintf(stderr, "joining\n");
    pthread_join(thread_id, NULL);

    thread_running = 0;

    close(pipe_fds[0]);
    close(pipe_fds[1]);
    pipe_fds[0] = pipe_fds[1] = -1;
    fprintf(stderr, "join done\n");
}
    
static void execute_request(struct request *r) {
    char x = 'x';
    assert(r);

    r->success = r->done = 0;
    
    pthread_mutex_lock(&request_mutex);

    while (current_request && thread_running)
        pthread_cond_wait(&request_cond, &request_mutex);
        
    if (!thread_running) {
        r->success = 0;
        r->done = 1;
    } else {
        current_request = r;
        
        assert(pipe_fds[1] >= 0);
        write(pipe_fds[1], &x, sizeof(x));

        while (!r->done && thread_running) {
            pthread_cond_wait(&request_cond, &request_mutex);
            /*fprintf(stderr, "COND done\n"); */
        }

        if (!thread_running) {
            r->success = 0;
            r->done = 1;
        }

        current_request = NULL;

        /* Notify other waiting threads, that the request was completed */
        pthread_cond_broadcast(&request_cond);
    }
    
    pthread_mutex_unlock(&request_mutex);
}

static void polyp_get_volume(int *l, int *r) {
    struct request req;
    int v;

    /*fprintf(stderr, "get_volume\n");*/
    
    req.message = MESSAGE_GETVOLUME;
    execute_request(&req);

    if (!req.success) {
        *l = *r = 100;
        return;
    }

    v = (req.volume*100)/PA_VOLUME_NORM;
    
    *r = *l = v > 100 ? 100 : v;
}

void polyp_set_volume(int l, int r) {
    struct request req;

    /*fprintf(stderr, "set_volume\n");*/

    req.message = MESSAGE_SETVOLUME;
    req.volume = ((l+r)*PA_VOLUME_NORM)/200;
    execute_request(&req);
}

static void polyp_pause(short b) {
    struct request r;

    fprintf(stderr, "*** PAUSE ***\n");

    fprintf(stderr, "pause: %s\n", b ? "yes" : "no");

    r.message = b ? MESSAGE_PAUSE : MESSAGE_UNPAUSE;
    execute_request(&r);
}

static int polyp_free(void) {
    int ret;
    struct request r;

    r.message = MESSAGE_WRITABLE;
    execute_request(&r);

    if (!r.success)
        return 0;

    ret = r.value;
    fprintf(stderr, "free: %u", ret);
    
    if (do_trigger && !triggered) {
        fprintf(stderr, "trigger\n");
        r.message = MESSAGE_TRIGGER;
        execute_request(&r);
        triggered = 1;
    }
    
    do_trigger = !!ret;
    return ret;
}

static int polyp_playing(void) {
    struct request r;

    r.message = MESSAGE_LATENCY;
    execute_request(&r);

    if (!r.success)
        return 0;

    fprintf(stderr, "playing : %s\n", r.value ? "yes" : "no");
    
    return r.value != 0;
}

static int polyp_get_written_time(void) {
    return (int)  ((((double) 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;
    execute_request(&r);

    if (!r.success)
        return 0;

    t = polyp_get_written_time();
    ms = r.latency/1000;
    
    return ms > t ? 0 : t-ms+100;
}

static void polyp_flush(int time) {
    struct request r;

    fprintf(stderr, "*** FLUSH ***\n");
    
    /*fprintf(stderr, "flush: %i\n", time);*/

    r.message = MESSAGE_FLUSH;
    execute_request(&r);

    written = (size_t) (((double)time*sample_spec.rate/1000)*pa_frame_size(&sample_spec));
    fprintf(stderr, "flush: %u -- %u\n", written, time);
}

static void polyp_write(void* ptr, int length) {
    struct request r;

    r.message = MESSAGE_WRITE;
    r.data = memdup(ptr, length);
    r.length = length;
    
    execute_request(&r);

    written += length;
    do_trigger = 0;
}

static void polyp_close(void) {
    struct request r;
    fprintf(stderr, "close\n");

    assert(thread_running);

    fprintf(stderr, "msg close\n");
    r.message = MESSAGE_CLOSE;
    execute_request(&r);
    fprintf(stderr, "msg done\n");

    stop_thread();
}

static int polyp_open(AFormat fmt, int rate, int nch) {
    struct request r;
    fprintf(stderr, "open\n");

    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 == FMT_S16_BE)
        r.ss.format = PA_SAMPLE_S16BE;
    else if (fmt == FMT_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;

    sample_spec = r.ss;
    
    start_thread();

    fprintf(stderr, "go for req\n");
    r.message = MESSAGE_OPEN;
    execute_request(&r);
    fprintf(stderr, "done req\n");

    if (!r.success) {
        polyp_close();
        fprintf(stderr, "open() failed.\n");
        return 0;
    }

    written = do_trigger = triggered = failed = 0;

    fprintf(stderr, "READY\n");
    
    return 1;
}

static void polyp_init(void) {
    fprintf(stderr, "init\n");
}

static OutputPlugin polyp_plugin = {
    NULL,
    NULL,
    "Polypaudio Output Plugin", /* Description */
    polyp_init,                        
    NULL, /* polyp_about, */
    NULL, /* polyp_configure, */
    polyp_get_volume,
    polyp_set_volume,
    polyp_open,                         
    polyp_write,                        
    polyp_close,                        
    polyp_flush,                        
    polyp_pause,                        
    polyp_free,                         
    polyp_playing,                      
    polyp_get_output_time,              
    polyp_get_written_time,             
};

OutputPlugin *get_oplugin_info(void) {
    return &polyp_plugin;
}