/* Called from IO thread context */
static void sink_input_state_change_cb(pa_sink_input *i, pa_sink_input_state_t state) {
    struct userdata *u;

    pa_sink_input_assert_ref(i);
    pa_assert_se(u = i->userdata);

    /* If we are added for the first time, ask for a rewinding so that
     * we are heard right-away. */
    if (PA_SINK_INPUT_IS_LINKED(state) &&
        i->thread_info.state == PA_SINK_INPUT_INIT) {
        pa_log_debug("Requesting rewind due to state change.");
        pa_sink_input_request_rewind(i, 0, false, true, true);
    }
}

/* Called from main context */
void pa_source_output_unlink(pa_source_output*o) {
    pa_bool_t linked;
    pa_assert(o);
    pa_assert_ctl_context();

    /* See pa_sink_unlink() for a couple of comments how this function
     * works */

    pa_source_output_ref(o);

    linked = PA_SOURCE_OUTPUT_IS_LINKED(o->state);

    if (linked)
        pa_hook_fire(&o->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK], o);

    if (o->direct_on_input)
        pa_idxset_remove_by_data(o->direct_on_input->direct_outputs, o, NULL);

    pa_idxset_remove_by_data(o->core->source_outputs, o, NULL);

    if (o->source)
        if (pa_idxset_remove_by_data(o->source->outputs, o, NULL))
            pa_source_output_unref(o);

    if (o->client)
        pa_idxset_remove_by_data(o->client->source_outputs, o, NULL);

    update_n_corked(o, PA_SOURCE_OUTPUT_UNLINKED);
    o->state = PA_SOURCE_OUTPUT_UNLINKED;

    if (linked && o->source)
        if (o->source->asyncmsgq)
            pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_REMOVE_OUTPUT, o, 0, NULL) == 0);

    reset_callbacks(o);

    if (linked) {
        pa_subscription_post(o->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_REMOVE, o->index);
        pa_hook_fire(&o->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK_POST], o);
    }

    if (o->source) {
        pa_source_update_status(o->source);
        o->source = NULL;
    }

    pa_core_maybe_vacuum(o->core);

    pa_source_output_unref(o);
}

/* Called from main context */
static int raw_source_set_state(pa_source *s, pa_source_state_t state) {
    struct userdata *u;
    int ret;
    ENTER();

    pa_source_assert_ref(s);
    pa_assert_se(u = s->userdata);

    ret = voice_source_set_state(s, u->voip_source, state);

    pa_log_debug("(%p): called with %d", (void *)s, state);

    return ret;
}

static pa_simple_protocol* simple_protocol_new(pa_core *c) {
    pa_simple_protocol *p;

    pa_assert(c);

    p = pa_xnew(pa_simple_protocol, 1);
    PA_REFCNT_INIT(p);
    p->core = c;
    p->connections = pa_idxset_new(NULL, NULL);

    pa_assert_se(pa_shared_set(c, "simple-protocol", p) >= 0);

    return p;
}

static DBusHandlerResult profile_handler(DBusConnection *c, DBusMessage *m, void *userdata) {
    pa_bluetooth_backend *b = userdata;
    DBusMessage *r = NULL;
    const char *path, *interface, *member;

    pa_assert(b);

    path = dbus_message_get_path(m);
    interface = dbus_message_get_interface(m);
    member = dbus_message_get_member(m);

    pa_log_debug("dbus: path=%s, interface=%s, member=%s", path, interface, member);

    if (!pa_streq(path, HSP_AG_PROFILE))
        return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;

    if (dbus_message_is_method_call(m, "org.freedesktop.DBus.Introspectable", "Introspect")) {
        const char *xml = PROFILE_INTROSPECT_XML;

        pa_assert_se(r = dbus_message_new_method_return(m));
        pa_assert_se(dbus_message_append_args(r, DBUS_TYPE_STRING, &xml, DBUS_TYPE_INVALID));

    } else if (dbus_message_is_method_call(m, BLUEZ_PROFILE_INTERFACE, "Release")) {
    } else if (dbus_message_is_method_call(m, BLUEZ_PROFILE_INTERFACE, "RequestDisconnection")) {
        r = profile_request_disconnection(c, m, userdata);
    } else if (dbus_message_is_method_call(m, BLUEZ_PROFILE_INTERFACE, "NewConnection"))
        r = profile_new_connection(c, m, userdata);
    else
        return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;

    if (r) {
        pa_assert_se(dbus_connection_send(pa_dbus_connection_get(b->connection), r, NULL));
        dbus_message_unref(r);
    }

    return DBUS_HANDLER_RESULT_HANDLED;
}

pa_mutex* pa_mutex_new(bool recursive, bool inherit_priority) {
    pa_mutex *m;
    pthread_mutexattr_t attr;
    int r;

    pa_assert_se(pthread_mutexattr_init(&attr) == 0);

    if (recursive)
        pa_assert_se(pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE) == 0);

#ifdef HAVE_PTHREAD_PRIO_INHERIT
    if (inherit_priority) {
        r = pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);
        pa_assert(r == 0 || r == ENOTSUP);
    }
#endif

    m = pa_xnew(pa_mutex, 1);

#ifndef HAVE_PTHREAD_PRIO_INHERIT
    pa_assert_se(pthread_mutex_init(&m->mutex, &attr) == 0);

#else
    if ((r = pthread_mutex_init(&m->mutex, &attr))) {

        /* If this failed, then this was probably due to non-available
         * priority inheritance. In which case we fall back to normal
         * mutexes. */
        pa_assert(r == ENOTSUP && inherit_priority);

        pa_assert_se(pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_NONE) == 0);
        pa_assert_se(pthread_mutex_init(&m->mutex, &attr) == 0);
    }
#endif

    return m;
}

int pa_shm_attach(pa_shm *m, unsigned id, bool writable) {
    char fn[32];
    int fd = -1;
    int prot;
    struct stat st;

    pa_assert(m);

    segment_name(fn, sizeof(fn), m->id = id);

    if ((fd = shm_open(fn, writable ? O_RDWR : O_RDONLY, 0)) < 0) {
        if (errno != EACCES && errno != ENOENT)
            pa_log("shm_open() failed: %s", pa_cstrerror(errno));
        goto fail;
    }

    if (fstat(fd, &st) < 0) {
        pa_log("fstat() failed: %s", pa_cstrerror(errno));
        goto fail;
    }

    if (st.st_size <= 0 ||
        st.st_size > (off_t) (MAX_SHM_SIZE+SHM_MARKER_SIZE) ||
        PA_ALIGN((size_t) st.st_size) != (size_t) st.st_size) {
        pa_log("Invalid shared memory segment size");
        goto fail;
    }

    m->size = (size_t) st.st_size;

    prot = writable ? PROT_READ | PROT_WRITE : PROT_READ;
    if ((m->ptr = mmap(NULL, PA_PAGE_ALIGN(m->size), prot, MAP_SHARED, fd, (off_t) 0)) == MAP_FAILED) {
        pa_log("mmap() failed: %s", pa_cstrerror(errno));
        goto fail;
    }

    m->do_unlink = false;
    m->shared = true;

    pa_assert_se(pa_close(fd) == 0);

    return 0;

fail:
    if (fd >= 0)
        pa_close(fd);

    return -1;
}

void pa_classify_add_source(struct userdata *u, const char *type, const char *prop,
                            enum pa_classify_method method, const char *arg,
                            pa_hashmap *ports, uint32_t flags)
{
    struct pa_classify *classify;

    pa_assert(u);
    pa_assert_se((classify = u->classify));
    pa_assert(classify->sources);
    pa_assert(type);
    pa_assert(prop);
    pa_assert(arg);

    devices_add(&classify->sources, type, prop, method, arg, ports, flags);
}

/* Called from I/O thread context */
static void sink_update_requested_latency_cb(pa_sink *s) {
    struct userdata *u;

    pa_sink_assert_ref(s);
    pa_assert_se(u = s->userdata);

    if (!PA_SINK_IS_LINKED(u->sink->thread_info.state) ||
        !PA_SINK_INPUT_IS_LINKED(u->sink_input->thread_info.state))
        return;

    /* Just hand this one over to the master sink */
    pa_sink_input_set_requested_latency_within_thread(
            u->sink_input,
            pa_sink_get_requested_latency_within_thread(s));
}

/* Called from I/O thread context */
static void source_update_requested_latency_cb(pa_source *s) {
    struct userdata *u;

    pa_source_assert_ref(s);
    pa_assert_se(u = s->userdata);

    if (!PA_SOURCE_IS_LINKED(u->source->thread_info.state) ||
        !PA_SOURCE_OUTPUT_IS_LINKED(u->source_output->thread_info.state))
        return;

    /* Just hand this one over to the master source */
    pa_source_output_set_requested_latency_within_thread(
            u->source_output,
            pa_source_get_requested_latency_within_thread(s));
}

int pa_classify_card(struct userdata *u, struct pa_card *card,
                     uint32_t flag_mask, uint32_t flag_value,
                     char *buf, int size)
{
    struct pa_classify *classify;
    struct pa_classify_card_def *defs;
    const char *name;
    char **profs;
    int    len;

    pa_assert(u);
    pa_assert_se((classify = u->classify));
    pa_assert(classify->cards);
    pa_assert_se((defs = classify->cards->defs));

    name  = pa_card_ext_get_name(card);
    profs = pa_card_ext_get_profiles(card);

    len = cards_classify(defs, name,profs, flag_mask,flag_value, buf,size);

    pa_xfree(profs);

    return len;
}

/* Called from I/O thread context */
static int source_output_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
    struct userdata *u;
    pa_assert_se(u = PA_SOURCE_OUTPUT(o)->userdata);

    switch (code) {
        case PA_SOURCE_OUTPUT_MESSAGE_GET_LATENCY:
            *((pa_usec_t*) data) = pa_bytes_to_usec(pa_memblockq_get_length(u->memblockq), &u->source_output->sample_spec);

            /* Fall through, the default handler will add in the extra
             * latency added by the resampler */
            break;
    }

    return pa_source_output_process_msg(o, code, data, offset, chunk);
}

/* Called from output thread context */
static void sink_input_attach_cb(pa_sink_input *i) {
    struct userdata *u;

    pa_sink_input_assert_ref(i);
    pa_sink_input_assert_io_context(i);
    pa_assert_se(u = i->userdata);

    u->rtpoll_item_read = pa_rtpoll_item_new_asyncmsgq_read(
            i->sink->thread_info.rtpoll,
            PA_RTPOLL_LATE,
            u->asyncmsgq);

    pa_memblockq_set_prebuf(u->memblockq, pa_sink_input_get_max_request(i)*2);
    pa_memblockq_set_maxrewind(u->memblockq, pa_sink_input_get_max_rewind(i));
}

static void stream_read_cb(pa_stream *p, size_t nbytes, void *userdata) {
    /* We don't care about the data, just drop it */

    for (;;) {
        const void *data;

        pa_assert_se((nbytes = pa_stream_readable_size(p)) != (size_t) -1);

        if (nbytes <= 0)
            break;

        fail_unless(pa_stream_peek(p, &data, &nbytes) == 0);
        fail_unless(pa_stream_drop(p) == 0);
    }
}

/* Called from I/O thread context */
static void sink_request_rewind_cb(pa_sink *s) {
    struct userdata *u;

    pa_sink_assert_ref(s);
    pa_assert_se(u = s->userdata);

    if (!PA_SINK_IS_LINKED(u->sink->thread_info.state) ||
        !PA_SINK_INPUT_IS_LINKED(u->sink_input->thread_info.state))
        return;

    /* Just hand this one over to the master sink */
    pa_sink_input_request_rewind(u->sink_input,
                                 s->thread_info.rewind_nbytes +
                                 pa_memblockq_get_length(u->memblockq), TRUE, FALSE, FALSE);
}

/* Called from main context */
static void cmtspeech_sink_input_kill_cb(pa_sink_input *i) {
    struct userdata *u;

    pa_log_debug("Kill called");

    pa_sink_input_assert_ref(i);
    pa_assert_se(u = i->userdata);
    pa_assert(u->sink_input == i);

    pa_log_warn("Kill called for cmtspeech sink input");
    cmtspeech_trigger_unload(u);

    pa_sink_input_unref(u->sink_input);
    u->sink_input = NULL;
}

/* Called from main context */
pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o) {
    pa_source_output_assert_ref(o);
    pa_assert_ctl_context();

    if (PA_SOURCE_OUTPUT_IS_LINKED(o->state) && o->source) {
        pa_usec_t usec = 0;
        pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL) == 0);
        return usec;
    }

    /* If this source output is not realized yet or is being moved, we
     * have to touch the thread info data directly */

    return o->thread_info.requested_source_latency;
}

pa_socket_server* pa_socket_server_new_unix(pa_mainloop_api *m, const char *filename) {
    int fd = -1;
    struct sockaddr_un sa;
    pa_socket_server *s;

    pa_assert(m);
    pa_assert(filename);

    if ((fd = pa_socket_cloexec(PF_UNIX, SOCK_STREAM, 0)) < 0) {
        pa_log("socket(): %s", pa_cstrerror(errno));
        goto fail;
    }

    memset(&sa, 0, sizeof(sa));
    sa.sun_family = AF_UNIX;
    pa_strlcpy(sa.sun_path, filename, sizeof(sa.sun_path));

    pa_make_socket_low_delay(fd);

    if (bind(fd, (struct sockaddr*) &sa, (socklen_t) SUN_LEN(&sa)) < 0) {
        pa_log("bind(): %s", pa_cstrerror(errno));
        goto fail;
    }

    /* Allow access from all clients. Sockets like this one should
     * always be put inside a directory with proper access rights,
     * because not all OS check the access rights on the socket
     * inodes. */
    chmod(filename, 0777);

    if (listen(fd, 5) < 0) {
        pa_log("listen(): %s", pa_cstrerror(errno));
        goto fail;
    }

    pa_assert_se(s = pa_socket_server_new(m, fd));

    s->filename = pa_xstrdup(filename);
    s->type = SOCKET_SERVER_UNIX;

    return s;

fail:
    if (fd >= 0)
        pa_close(fd);

    return NULL;
}

/* DBusTimeoutToggledFunction callback for pa mainloop */
static void toggle_timeout(DBusTimeout *timeout, void *data) {
    struct timeout_data *d = data;
    pa_time_event *ev;
    struct timeval tv;

    pa_assert(d);
    pa_assert(d->connection);
    pa_assert(timeout);

    pa_assert_se(ev = dbus_timeout_get_data(timeout));

    if (dbus_timeout_get_enabled(timeout))
        d->connection->mainloop->time_restart(ev, pa_timeval_rtstore(&tv, pa_rtclock_now() + dbus_timeout_get_interval(timeout) * PA_USEC_PER_MSEC, d->connection->use_rtclock));
    else
        d->connection->mainloop->time_restart(ev, pa_timeval_rtstore(&tv, PA_USEC_INVALID, d->connection->use_rtclock));
}

int pa_fdsem_before_poll(pa_fdsem *f) {
    pa_assert(f);

    flush(f);

    if (pa_atomic_cmpxchg(&f->data->signalled, 1, 0))
        return -1;

    pa_atomic_inc(&f->data->waiting);

    if (pa_atomic_cmpxchg(&f->data->signalled, 1, 0)) {
        pa_assert_se(pa_atomic_dec(&f->data->waiting) >= 1);
        return -1;
    }
    return 0;
}