void pulse_get_default_sink_volume(struct pulseaudio_t *pulse) {
    pa_operation *op;

    op = pa_context_get_sink_info_by_name(pulse->cxt, pulse->default_sink, pulse_get_default_sink_volume_cb, pulse);
    pulse_async_wait(pulse, op);
    pa_operation_unref(op);
}

static bool set_mic_mute_cb(LSHandle *handle, LSMessage *message, void *user_data)
{
	struct audio_service *service = user_data;
	const char *payload;
	jvalue_ref parsed_obj = NULL;
	struct luna_service_req_data *req;
	pa_operation *op;

	if (!service->context_initialized) {
		luna_service_message_reply_custom_error(handle, message, "Not yet initialized");
		return true;
	}

	payload = LSMessageGetPayload(message);
	parsed_obj = luna_service_message_parse_and_validate(payload);
	if (jis_null(parsed_obj)) {
		luna_service_message_reply_error_bad_json(handle, message);
		goto cleanup;
	}

	service->mic_mute = luna_service_message_get_boolean(parsed_obj, "micMute", service->mic_mute);

	req = luna_service_req_data_new(handle, message);
	req->user_data = service;

	op = pa_context_get_source_info_list(service->context, mm_sourceinfo_cb, req);
	pa_operation_unref(op);

cleanup:
	if (!jis_null(parsed_obj))
		j_release(&parsed_obj);

	return true;
}

/**
 * Synchronously wait until an operation completed.
 */
static int drvHostPulseAudioWaitFor(pa_operation *pOP, RTMSINTERVAL cMsTimeout)
{
    AssertPtrReturn(pOP, VERR_INVALID_POINTER);

    int rc = VINF_SUCCESS;
    if (pOP)
    {
        uint64_t u64StartMs = RTTimeMilliTS();
        while (pa_operation_get_state(pOP) == PA_OPERATION_RUNNING)
        {
            if (!g_fAbortMainLoop)
                pa_threaded_mainloop_wait(g_pMainLoop);
            g_fAbortMainLoop = false;

            uint64_t u64ElapsedMs = RTTimeMilliTS() - u64StartMs;
            if (u64ElapsedMs >= cMsTimeout)
            {
                rc = VERR_TIMEOUT;
                break;
            }
        }

        pa_operation_unref(pOP);
    }

    return rc;
}

static void play_feedback_sample(struct play_feedback_data *pfd)
{
	pa_operation *op;
	pa_proplist *proplist;
	char *sink = pfd->sink;

	if (!pfd->play) {
		luna_service_message_reply_success(pfd->handle, pfd->message);
		return;
	}

	if (!sink)
		sink = pfd->service->default_sink_name;

	if (!sink) {
		luna_service_message_reply_custom_error(pfd->handle, pfd->message, "No sink found to play sample on");
		return;
	}

	/* make sure we're running as event to enable ducking */
	proplist = pa_proplist_new();
	pa_proplist_setf(proplist, PA_PROP_MEDIA_ROLE, "event");

	op = pa_context_play_sample_with_proplist(pfd->service->context, pfd->name, sink, PA_VOLUME_NORM, proplist, NULL, NULL);
	if (op)
		pa_operation_unref(op);

	luna_service_message_reply_success(pfd->handle, pfd->message);
}

/*****************************************************************************
 * Close: close the audio device
 *****************************************************************************/
static void Close ( vlc_object_t *p_this )
{
    aout_instance_t *p_aout = (aout_instance_t *)p_this;
    struct aout_sys_t * p_sys = p_aout->output.p_sys;

    msg_Dbg(p_aout, "Pulse Close");

    if(p_sys->stream){
        pa_operation *o;
        pa_threaded_mainloop_lock(p_sys->mainloop);
        pa_stream_set_write_callback(p_sys->stream, NULL, NULL);

        if((o = pa_stream_drain(p_sys->stream, success_cb, p_aout))){
            while (pa_operation_get_state(o) != PA_OPERATION_DONE) {
                CHECK_DEAD_GOTO(fail);
                pa_threaded_mainloop_wait(p_sys->mainloop);
            }

        fail:

            pa_operation_unref(o);
        }

        pa_threaded_mainloop_unlock(p_sys->mainloop);
    }
    uninit(p_aout);
}

/* Someone requested that the latency is shown */
static void sigusr1_signal_callback(pa_mainloop_api*m, pa_signal_event *e, int sig, void *userdata) {

    if (!stream)
        return;

    pa_operation_unref(pa_stream_update_timing_info(stream, stream_update_timing_callback, NULL));
}

static PyObject* PulseAudio_resume(output_PulseAudio *self, PyObject *args)
{
    /*ensure output stream is still running*/
    /*FIXME*/

    /*uncork output stream, if corked*/
    pa_threaded_mainloop_lock(self->mainloop);

    if (pa_stream_is_corked(self->stream)) {
        pa_operation *op = pa_stream_cork(
            self->stream,
            0,
            (pa_stream_success_cb_t)success_callback,
            self->mainloop);

        while (pa_operation_get_state(op) == PA_OPERATION_RUNNING) {
            pa_threaded_mainloop_wait(self->mainloop);
        }

        pa_operation_unref(op);
    }

    pa_threaded_mainloop_unlock(self->mainloop);

    Py_INCREF(Py_None);
    return Py_None;
}

static void ext_device_restore_test_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
    pa_operation *o = userdata;
    uint32_t version = PA_INVALID_INDEX;

    pa_assert(pd);
    pa_assert(o);
    pa_assert(PA_REFCNT_VALUE(o) >= 1);

    if (!o->context)
        goto finish;

    if (command != PA_COMMAND_REPLY) {
        if (pa_context_handle_error(o->context, command, t, false) < 0)
            goto finish;

    } else if (pa_tagstruct_getu32(t, &version) < 0 ||
               !pa_tagstruct_eof(t)) {

        pa_context_fail(o->context, PA_ERR_PROTOCOL);
        goto finish;
    }

    if (o->callback) {
        pa_ext_device_restore_test_cb_t cb = (pa_ext_device_restore_test_cb_t) o->callback;
        cb(o->context, version, o->userdata);
    }

finish:
    pa_operation_done(o);
    pa_operation_unref(o);
}

void PulseAudioPlayer::Play(int64_t start,int64_t count)
{
	//printf("Starting PulseAudio playback\n");
	if (!open) OpenStream();

	if (is_playing) {
		// If we're already playing, do a quick "reset"
		is_playing = false;

		pa_threaded_mainloop_lock(mainloop);
		pa_operation *op = pa_stream_flush(stream, (pa_stream_success_cb_t)pa_stream_success, this);
		pa_threaded_mainloop_unlock(mainloop);
		stream_success.Wait();
		pa_operation_unref(op);
		if (!stream_success_val) {
			paerror = pa_context_errno(context);
			printf("PulseAudio player: Error flushing stream: %s (%d)\n", pa_strerror(paerror), paerror);
		}
	}

	start_frame = start;
	cur_frame = start;
	end_frame = start + count;
	//printf("start=%lu end=%lu\n", start_frame, end_frame);

	is_playing = true;

	play_start_time = 0;
	pa_threaded_mainloop_lock(mainloop);
	paerror = pa_stream_get_time(stream, (pa_usec_t*) &play_start_time);
	pa_threaded_mainloop_unlock(mainloop);
	if (paerror) {
		printf("PulseAudio player: Error getting stream time: %s (%d)\n", pa_strerror(paerror), paerror);
	}

	PulseAudioPlayer::pa_stream_write(stream, pa_stream_writable_size(stream), this);

	pa_threaded_mainloop_lock(mainloop);
	pa_operation *op = pa_stream_trigger(stream, (pa_stream_success_cb_t)pa_stream_success, this);
	pa_threaded_mainloop_unlock(mainloop);
	stream_success.Wait();
	pa_operation_unref(op);
	if (!stream_success_val) {
		paerror = pa_context_errno(context);
		printf("PulseAudio player: Error triggering stream: %s (%d)\n", pa_strerror(paerror), paerror);
	}
}

static int Open (vlc_object_t *obj)
{
    services_discovery_t *sd = (services_discovery_t *)obj;
    pa_operation *op;
    pa_context *ctx;

    services_discovery_sys_t *sys = malloc (sizeof (*sys));
    if (unlikely(sys == NULL))
        return VLC_ENOMEM;

    ctx = vlc_pa_connect (obj, &sys->mainloop);
    if (ctx == NULL)
    {
        free (sys);
        return VLC_EGENERIC;
    }

    sd->p_sys = sys;
    sd->description = _("Audio capture");
    sys->context = ctx;
    sys->root = NULL;

    /* Subscribe for source events */
    const pa_subscription_mask_t mask = PA_SUBSCRIPTION_MASK_SOURCE;
    pa_threaded_mainloop_lock (sys->mainloop);
    pa_context_set_subscribe_callback (ctx, ContextCallback, sd);
    op = pa_context_subscribe (ctx, mask, NULL, NULL);
    if (likely(op != NULL))
        pa_operation_unref (op);

    /* Enumerate existing sources */
    op = pa_context_get_source_info_list (ctx, SourceCallback, sd);
    if (likely(op != NULL))
    {
        //while (pa_operation_get_state (op) == PA_OPERATION_RUNNING)
        //    pa_threaded_mainloop_wait (sys->mainloop);
        pa_operation_unref (op);
    }
    pa_threaded_mainloop_unlock (sys->mainloop);
    return VLC_SUCCESS;
/*
error:
    pa_threaded_mainloop_unlock (sys->mainloop);
    vlc_pa_disconnect (obj, ctx, sys->mainloop);
    free (sys);
    return VLC_EGENERIC;*/
}

void SinkInput::setVolume(pa_cvolume v)
{
    pa_operation *o;
    this->d->svolume = v;
    o = pa_context_set_sink_input_volume(d->context->cObject(),
			d->index, &d->svolume, SinkInput::volume_cb, this);
    pa_operation_unref(o);
}

/** Wait until the specified operation completes */
static void wait_for_operation(struct pa_operation *o) {
    assert(o && context && mainloop);

    while (pa_operation_get_state(o) == PA_OPERATION_RUNNING)
        pa_mainloop_iterate(mainloop, 1, NULL);

    pa_operation_unref(o);
}

void processContextState( pa_context* context, void* userData )
{
	StateData* stateData = static_cast<StateData*>( userData );
	pa_context_state_t state = ::pa_context_get_state( context );
	switch( state ) {
		case PA_CONTEXT_READY: {
			switch( stateData->request ) {
				case StateData::GET_SINKS:
					pa_operation_unref( pa_context_get_sink_info_list( context, processSinkInfo, userData ) );
				break;

				case StateData::GET_SOURCES:
					pa_operation_unref( pa_context_get_source_info_list( context, processSourceInfo, userData ) );
				break;

				case StateData::GET_DEFAULT_SINK_NAME:
					pa_operation_unref( pa_context_get_server_info( context, processServerInfo, userData ) );
				break;

				case StateData::GET_DEFAULT_SOURCE_NAME:
					pa_operation_unref( pa_context_get_server_info( context, processServerInfo, userData ) );
				break;

				case StateData::DONE:
					stateData->mainLoopApi->quit( stateData->mainLoopApi, 0 );
				break;

				case StateData::INVALID:
					stateData->mainLoopApi->quit( stateData->mainLoopApi, -1 );
				break;
			}
		}
		break;

		case PA_CONTEXT_TERMINATED:
			stateData->mainLoopApi->quit( stateData->mainLoopApi, 0 );
		break;

		case PA_CONTEXT_FAILED:
			stateData->mainLoopApi->quit( stateData->mainLoopApi, -1 );
		break;
		
		default:
		break;
	}
}

static void stream_latency_cb(pa_stream *p, void *userdata) {
#ifndef INTERPOLATE
    pa_operation *o;

    o = pa_stream_update_timing_info(p, NULL, NULL);
    pa_operation_unref(o);
#endif
}

void Source::setVolume(pa_cvolume v)
{
    pa_operation *o;
    this->d->svolume = v;
    o = pa_context_set_source_volume_by_index(d->context->cObject(),
			(Device::d)->index, &d->svolume, Source::volume_cb, this);
    pa_operation_unref(o);
}

void PulseOutput::drain() {
  if (stream) {
    pa_operation * operation = pa_stream_drain(stream,nullptr,nullptr);
    while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
      context->wait_plugin_events();
    pa_operation_unref(operation);
    }
  }

void CPulseAESound::Stop()
{
  if (m_op)
  {
    pa_operation_cancel(m_op);
    pa_operation_unref(m_op);
    m_op = NULL;
  }
}

void CPulseAESound::Play()
{
  pa_threaded_mainloop_lock(m_mainLoop);
  /* we only keep the most recent operation as it is the only one needed for IsPlaying to function */
  if (m_op)
    pa_operation_unref(m_op);
  m_op = pa_context_play_sample(m_context, m_pulseName.c_str(), NULL, PA_VOLUME_INVALID, NULL, NULL);
  pa_threaded_mainloop_unlock(m_mainLoop);
}

static void
tsmf_pulse_wait_for_operation(TSMFPulseAudioDevice * pulse, pa_operation * operation)
{
	while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
	{
		pa_threaded_mainloop_wait(pulse->mainloop);
	}
	pa_operation_unref(operation);
}

static void get_sink_info_callback(pa_context *c, const pa_sink_info *i, int is_last, void *userdata) {

    if ( is_last )
    {
        pa_operation_unref(pa_context_get_server_info(c, get_server_info_callback, userdata));
        return;
    }
    printf("Sink: %s - %s\n",i->description,i->name);
}