path: root/module-4a-client.c

/***
  This file is part of PulseAudio.

  Copyright 2018 Collabora Ltd.
    Author: George Kiagiadakis <george.kiagiadakis@collabora.com>

  PulseAudio is free software; you can redistribute it and/or modify
  it under the terms of the GNU Lesser General Public License as published
  by the Free Software Foundation; either version 2.1 of the License,
  or (at your option) any later version.

  PulseAudio 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 Lesser General Public License
  along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
***/

#include <pulsecore/pulsecore-config.h>
#include <pulsecore/core-util.h>
#include <pulsecore/llist.h>
#include <pulsecore/module.h>
#include <pulsecore/mutex.h>
#include <pulsecore/modargs.h>
#include <pulsecore/dynarray.h>
#include <pulsecore/semaphore.h>

#include <stdbool.h>

#include "m4a_afb_comm.h"

#define DEFAULT_URI "unix:/run/user/0/apis/ws/ahl-4a"

PA_MODULE_AUTHOR("George Kiagiadakis");
PA_MODULE_DESCRIPTION("Makes PulseAudio work as a client of the AGL Advanced Audio Architecture");
PA_MODULE_LOAD_ONCE(true);
PA_MODULE_VERSION(PACKAGE_VERSION);
PA_MODULE_USAGE(
          "uri=<ahl-4a unix socket uri>"
          "default_role=<role>");

static const char* const valid_modargs[] = {
    "uri",
    "default_role",
    NULL
};

typedef struct _m4a_stream {
    pa_module *self;

    union {
        pa_sink_input *sink_input;
        pa_source_output *source_output;
    };

    char *role;
    char *device_uri;

    bool entity_is_sink;
    bool entity_loaded;
    uint32_t entity_id;
    uint32_t entity_module_id;

    pa_semaphore *semaphore;

    PA_LLIST_FIELDS(struct _m4a_stream);
} m4a_stream;

typedef struct {
    /* configuration data */
    pa_modargs *ma;

    /* operational data */
    pa_mutex *lock;
    pa_dynarray *roles;  /* element-type: char* */
    PA_LLIST_HEAD(m4a_stream, streams);
    pa_hook_slot
        *sink_input_put_slot,
        *sink_input_unlink_post_slot;
    m4a_afb_comm *comm;
} m4a_data;

static bool role_exists(m4a_data *d, const char *role) {
    char *r;
    int idx;

    PA_DYNARRAY_FOREACH(r, d->roles, idx) {
        if (!strcmp(role, r))
            return true;
    }
    return false;
}

static char *decide_role(m4a_data *d, pa_proplist *p) {
    const char *role;

    role = pa_proplist_gets(p, PA_PROP_MEDIA_ROLE);
    if (role && role_exists(d, role)) {
        return pa_xstrdup(role);
    }

    /* FIXME we might want to do some mapping here between the standard
     * pulseaudio roles and whatever roles might be available in 4a */

    /* try the default role specified in the module arguments, or fall back
     * to "multimedia", which we can just hope that it exists */
    role = pa_modargs_get_value(d->ma, "default_role", "multimedia");
    if (role_exists(d, role)) {
        return pa_xstrdup(role);
    }

    return NULL;
}

static pa_sink *load_sink_module(pa_core *core, const char *module, char *params) {
    pa_module *m = NULL;
    pa_sink *target = NULL;
    uint32_t idx;

    if (!(m = pa_module_load(core, module, params))) {
        pa_log("Failed to load module %s %s", module, params);
        return NULL;
    }
    pa_xfree(params);
    PA_IDXSET_FOREACH(target, core->sinks, idx) {
        if (target->module == m)
            break;
    }
    if (target->module != m) {
        pa_log("%s didn't create any sinks", module);
        pa_module_unload(m, false);
        return NULL;
    }
    return target;
}

/* invoked in the afb communication thread */
static void got_roles_cb(enum m4a_afb_reply r,
                         json_object *response,
                         pa_module *self) {
    m4a_data *d = self->userdata;
    int length, i;
    json_object *jr;
    char *role;

    pa_mutex_lock(d->lock);

    if (r == M4A_AFB_REPLY_OK && json_object_get_type(response) == json_type_array) {
        length = json_object_array_length(response);
        for (i = 0; i < length; i++) {
            jr = json_object_array_get_idx(response, i);
            if (json_object_get_type(jr) == json_type_string) {
                role = pa_xstrdup(json_object_get_string(jr));
                pa_log_debug("Found 4a role: %s", role);
                pa_dynarray_append(d->roles, role);
            }
        }
    }

    pa_mutex_unlock(d->lock);
}

static m4a_stream *m4a_stream_new(pa_module *self, bool sink) {
    m4a_data *d = self->userdata;
    m4a_stream *stream;

    stream = pa_xnew0(m4a_stream, 1);
    stream->self = self;
    stream->entity_is_sink = sink;
    stream->semaphore = pa_semaphore_new(0);
    pa_mutex_lock(d->lock);
    PA_LLIST_PREPEND(m4a_stream, d->streams, stream);
    pa_mutex_unlock(d->lock);
    return stream;
}

static void m4a_stream_free(m4a_stream *stream) {
    m4a_data *d = stream->self->userdata;

    if (stream->role)
        pa_xfree(stream->role);
    if (stream->device_uri)
        pa_xfree(stream->device_uri);
    if (stream->entity_loaded)
        pa_module_unload_request_by_index(stream->self->core,
                                          stream->entity_module_id, false);
    pa_semaphore_free(stream->semaphore);
    pa_mutex_lock(d->lock);
    PA_LLIST_REMOVE(m4a_stream, d->streams, stream);
    pa_mutex_unlock(d->lock);
    pa_xfree(stream);
}

/* invoked in the afb communication thread */
static void device_open_cb(enum m4a_afb_reply r,
                           json_object *response,
                           m4a_stream *stream) {
    json_object *jdu;
    const char *device_uri = NULL;

    if (r == M4A_AFB_REPLY_OK && json_object_get_type(response) == json_type_object) {
        json_object_object_get_ex(response, "device_uri", &jdu);
        if (json_object_object_get_ex(response, "device_uri", &jdu) &&
            json_object_get_type(jdu) == json_type_string) {
            device_uri = json_object_get_string(jdu);
        }
    }

    pa_log_debug("4A replied: %s, device_uri: %s",
                 (r == M4A_AFB_REPLY_OK) ? "OK" : "ERROR",
                 device_uri ? device_uri : "(null)");

    stream->device_uri = pa_xstrdup(device_uri);
    pa_semaphore_post(stream->semaphore);
}

/* invoked in the afb communication thread */
static void device_close_cb(enum m4a_afb_reply r,
                            json_object *response,
                            m4a_stream *stream) {
    pa_log_debug("4A replied: %s",
                 (r == M4A_AFB_REPLY_OK) ? "OK" : "ERROR");
    m4a_stream_free(stream);
}

static pa_hook_result_t sink_input_put_cb(pa_core *core,
                                          pa_sink_input *i,
                                          pa_module *self) {
    m4a_data *d = self->userdata;
    json_object *jparams;
    char *role;

    pa_mutex_lock(d->lock);
    role = decide_role(d, i->proplist);
    pa_mutex_unlock(d->lock);

    pa_log_info("New sink_input with role: %s", role ? role : "(null)");

    if (role) {
        m4a_stream *stream = m4a_stream_new(self, true);
        stream->role = role;
        stream->sink_input = i;

        pa_log_debug("Calling 4A to open the device");

        jparams = json_object_new_object();
        json_object_object_add(jparams, "action", json_object_new_string("open"));

        m4a_afb_call_async(d->comm, role, jparams,
                           (m4a_afb_done_cb_t) device_open_cb, stream);

        pa_log_debug("waiting for 4A to reply");
        pa_semaphore_wait(stream->semaphore);

        if (stream->device_uri) {
            pa_sink *s;

            if ((s = load_sink_module(self->core, "module-alsa-sink",
                        pa_sprintf_malloc("device=%s", stream->device_uri)))) {
                stream->entity_id = s->index;
                stream->entity_module_id = s->module->index;
                stream->entity_loaded = true;

                pa_log_info("moving sink_input to alsa sink");
                pa_sink_input_move_to(i, s, false);
            } else {
                pa_log_info("failed to load alsa sink, closing 4A device");

                jparams = json_object_new_object();
                json_object_object_add(jparams, "action",
                                       json_object_new_string("close"));

                m4a_afb_call_async(d->comm, role, jparams,
                           (m4a_afb_done_cb_t) device_close_cb, stream);
                return PA_HOOK_CANCEL;
            }
        } else {
            pa_log_info("sink_input is not authorized to connect to 4A");
            //TODO maybe queue and play this stream when 4A allows it

            m4a_stream_free(stream);
            return PA_HOOK_CANCEL;
        }
    }

    return PA_HOOK_OK;
}

static pa_hook_result_t sink_input_unlink_post_cb(pa_core *core,
                                                  pa_sink_input *i,
                                                  pa_module *self) {
    m4a_data *d = self->userdata;
    m4a_stream *stream = NULL;
    json_object *jparams;

    PA_LLIST_FOREACH(stream, d->streams) {
        if (stream->sink_input == i)
            break;
    }

    if (stream && stream->sink_input == i) {
        pa_log_debug("unloading module-alsa-sink (%s)", stream->role);

        pa_module_unload_by_index(stream->self->core,
                                  stream->entity_module_id, false);
        stream->entity_loaded = false;

        jparams = json_object_new_object();
        json_object_object_add(jparams, "action", json_object_new_string("close"));

        m4a_afb_call_async(d->comm, stream->role, jparams,
                        (m4a_afb_done_cb_t) device_close_cb, stream);
    }

    return PA_HOOK_OK;
}

void pa__done(pa_module *self) {
    m4a_data *d;

    pa_assert(self);

    d = self->userdata;

    while (d->streams)
        m4a_stream_free(d->streams);

    pa_dynarray_free(d->roles);
    pa_mutex_free(d->lock);

    if (d->ma)
        pa_modargs_free(d->ma);

    if (d->sink_input_put_slot)
        pa_hook_slot_free(d->sink_input_put_slot);
    if (d->sink_input_unlink_post_slot)
        pa_hook_slot_free(d->sink_input_unlink_post_slot);

    if (d->comm)
        m4a_afb_comm_free(d->comm);
}

int pa__init(pa_module *self) {
    m4a_data *d;
    const char *uri;

    pa_assert(self);

    self->userdata = d = pa_xnew0(m4a_data, 1);
    d->roles = pa_dynarray_new(pa_xfree);
    d->lock = pa_mutex_new(false, false);

    if (!(d->ma = pa_modargs_new(self->argument, valid_modargs))) {
        pa_log("Failed to parse module arguments");
        goto fail;
    }

    uri = pa_modargs_get_value(d->ma, "uri", DEFAULT_URI);
    if (!(d->comm = m4a_afb_comm_new(uri))) {
        goto fail;
    }

    if (!m4a_afb_call_async(d->comm, "get_roles", NULL,
                            (m4a_afb_done_cb_t) got_roles_cb, self)) {
        goto fail;
    }

    d->sink_input_put_slot = pa_hook_connect(&self->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT],
            PA_HOOK_NORMAL, (pa_hook_cb_t) sink_input_put_cb, self);
    d->sink_input_unlink_post_slot = pa_hook_connect(&self->core->hooks[PA_CORE_HOOK_SINK_INPUT_UNLINK_POST],
            PA_HOOK_NORMAL, (pa_hook_cb_t) sink_input_unlink_post_cb, self);
    return 0;

fail:
    pa__done(self);
    return -1;
}