/* $Id$ */

/***
  This file is part of xmms-polyp.
 
  xmms-polyp is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; either version 2 of the License, or
  (at your option) any later version.
 
  xmms-polyp is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  General Public License for more details.
 
  You should have received a copy of the GNU General Public License
  along with xmms-polyp; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  USA.
***/

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

#include <gtk/gtk.h>
#include <plugin.h>
#include <xmmsctrl.h>
#include <util.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;

/* This function is from xmms' core */
gint ctrlsocket_get_session_id(void);

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

#ifndef HOST_NAME_MAX
#define HOST_NAME_MAX 255
#endif

static const char* get_song_name(void) {
    static char t[256];
    gint session, pos;
    char *str;

    session = ctrlsocket_get_session_id();
    pos = xmms_remote_get_playlist_pos(session);
    if (!(str = xmms_remote_get_playlist_title(session, pos)))
        return "xmms";

    snprintf(t, sizeof(t), "XMMS [%s]", str);
    
    return t;
}

static const char* get_host(void) {
    static char t[HOST_NAME_MAX+1];
    gethostname(t, sizeof(t));
    t[HOST_NAME_MAX] = 0;
    return t;
}

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) {
        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);

    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, get_song_name(), &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: {
                    char t[64];
                    assert(!context && !stream);

                    snprintf(t, sizeof(t), "XMMS (PID %lu on %s)", (unsigned long) getpid(), get_host());
                    context = pa_context_new(api, t);
                    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);
                        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_PAUSE, 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);
    
    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);
    
    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) {
    pthread_join(thread_id, NULL);
    thread_running = 0;
    close(pipe_fds[0]);
    close(pipe_fds[1]);
    pipe_fds[0] = pipe_fds[1] = -1;
}
    
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);

        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;

    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;

    req.message = MESSAGE_SETVOLUME;
    req.volume = ((l>r?l:r)*PA_VOLUME_NORM)/100;
    execute_request(&req);
}

static void polyp_pause(short b) {
    struct request r;
    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;
    
    if (do_trigger && !triggered) {
        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;

    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;

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

    written = (size_t) (((double)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 = memdup(ptr, length);
    r.length = length;
    
    execute_request(&r);

    written += length;
    do_trigger = 0;
}

static void polyp_close(void) {
    struct request r;
    assert(thread_running);

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

    stop_thread();
}

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 == 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();

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

    if (!r.success) {
        polyp_close();
        return 0;
    }

    written = do_trigger = triggered = failed = 0;

    return 1;
}

static void polyp_init(void) {
}

static void polyp_about(void) {
    static GtkWidget *dialog;
    
    if (dialog != NULL)
        return;
    
    dialog = xmms_show_message(
        "About XMMS Polypaudio Output Plugin",
        "XMMS Polypaudio Output Plugin\n\n "
          "This program is free software; you can redistribute it and/or modify\n"
          "it under the terms of the GNU General Public License as published by\n"
          "the Free Software Foundation; either version 2 of the License, or\n"
          "(at your option) any later version.\n"
          "\n"
          "This program is distributed in the hope that it will be useful,\n"
          "but WITHOUT ANY WARRANTY; without even the implied warranty of\n"
          "MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\n"
          "GNU General Public License for more details.\n"
          "\n"
          "You should have received a copy of the GNU General Public License\n"
          "along with this program; if not, write to the Free Software\n"
          "Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,\n"
          "USA.", "OK", FALSE, NULL, NULL);
    gtk_signal_connect(GTK_OBJECT(dialog), "destroy",
                       GTK_SIGNAL_FUNC(gtk_widget_destroyed),
                       &dialog);
}


static OutputPlugin polyp_plugin = {
    NULL,
    NULL,
    "Polypaudio Output Plugin",
    polyp_init,                        
    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;
}