path: root/src/plugin.c

#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,
        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;

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 pa_usec_t latency;
static int do_trigger;

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 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(c)) {
        case PA_STREAM_CREATING:
            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:
            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, &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 && 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;
    
    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);
                    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->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;
                                       
            }
        }
    }
    
    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;
}
    
static void polyp_get_volume(int *l, int *r) {
    struct request r;
    int v;

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

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