michael@0: /*
michael@0:  * Copyright © 2011 Mozilla Foundation
michael@0:  *
michael@0:  * This program is made available under an ISC-style license.  See the
michael@0:  * accompanying file LICENSE for details.
michael@0:  */
michael@0: #undef NDEBUG
michael@0: #include <assert.h>
michael@0: #include <dlfcn.h>
michael@0: #include <stdlib.h>
michael@0: #include <pulse/pulseaudio.h>
michael@0: #include <string.h>
michael@0: #include "cubeb/cubeb.h"
michael@0: #include "cubeb-internal.h"
michael@0: 
michael@0: #ifdef DISABLE_LIBPULSE_DLOPEN
michael@0: #define WRAP(x) x
michael@0: #else
michael@0: #define WRAP(x) cubeb_##x
michael@0: #define MAKE_TYPEDEF(x) static typeof(x) * cubeb_##x
michael@0: MAKE_TYPEDEF(pa_channel_map_init_auto);
michael@0: MAKE_TYPEDEF(pa_context_connect);
michael@0: MAKE_TYPEDEF(pa_context_disconnect);
michael@0: MAKE_TYPEDEF(pa_context_drain);
michael@0: MAKE_TYPEDEF(pa_context_get_state);
michael@0: MAKE_TYPEDEF(pa_context_new);
michael@0: MAKE_TYPEDEF(pa_context_rttime_new);
michael@0: MAKE_TYPEDEF(pa_context_set_state_callback);
michael@0: MAKE_TYPEDEF(pa_context_unref);
michael@0: MAKE_TYPEDEF(pa_context_get_sink_info_by_name);
michael@0: MAKE_TYPEDEF(pa_context_get_server_info);
michael@0: MAKE_TYPEDEF(pa_frame_size);
michael@0: MAKE_TYPEDEF(pa_operation_get_state);
michael@0: MAKE_TYPEDEF(pa_operation_unref);
michael@0: MAKE_TYPEDEF(pa_rtclock_now);
michael@0: MAKE_TYPEDEF(pa_stream_begin_write);
michael@0: MAKE_TYPEDEF(pa_stream_cancel_write);
michael@0: MAKE_TYPEDEF(pa_stream_connect_playback);
michael@0: MAKE_TYPEDEF(pa_stream_cork);
michael@0: MAKE_TYPEDEF(pa_stream_disconnect);
michael@0: MAKE_TYPEDEF(pa_stream_get_latency);
michael@0: MAKE_TYPEDEF(pa_stream_get_state);
michael@0: MAKE_TYPEDEF(pa_stream_get_time);
michael@0: MAKE_TYPEDEF(pa_stream_new);
michael@0: MAKE_TYPEDEF(pa_stream_set_state_callback);
michael@0: MAKE_TYPEDEF(pa_stream_set_write_callback);
michael@0: MAKE_TYPEDEF(pa_stream_unref);
michael@0: MAKE_TYPEDEF(pa_stream_update_timing_info);
michael@0: MAKE_TYPEDEF(pa_stream_write);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_free);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_get_api);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_lock);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_in_thread);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_new);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_signal);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_start);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_stop);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_unlock);
michael@0: MAKE_TYPEDEF(pa_threaded_mainloop_wait);
michael@0: MAKE_TYPEDEF(pa_usec_to_bytes);
michael@0: #undef MAKE_TYPEDEF
michael@0: #endif
michael@0: 
michael@0: static struct cubeb_ops const pulse_ops;
michael@0: 
michael@0: struct cubeb {
michael@0:   struct cubeb_ops const * ops;
michael@0:   void * libpulse;
michael@0:   pa_threaded_mainloop * mainloop;
michael@0:   pa_context * context;
michael@0:   pa_sink_info * default_sink_info;
michael@0:   char * context_name;
michael@0:   int error;
michael@0: };
michael@0: 
michael@0: struct cubeb_stream {
michael@0:   cubeb * context;
michael@0:   pa_stream * stream;
michael@0:   cubeb_data_callback data_callback;
michael@0:   cubeb_state_callback state_callback;
michael@0:   void * user_ptr;
michael@0:   pa_time_event * drain_timer;
michael@0:   pa_sample_spec sample_spec;
michael@0:   int shutdown;
michael@0: };
michael@0: 
michael@0: enum cork_state {
michael@0:   UNCORK = 0,
michael@0:   CORK = 1 << 0,
michael@0:   NOTIFY = 1 << 1
michael@0: };
michael@0: 
michael@0: static void
michael@0: sink_info_callback(pa_context * context, const pa_sink_info * info, int eol, void * u)
michael@0: {
michael@0:   cubeb * ctx = u;
michael@0:   if (!eol) {
michael@0:     ctx->default_sink_info = malloc(sizeof(pa_sink_info));
michael@0:     memcpy(ctx->default_sink_info, info, sizeof(pa_sink_info));
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_signal)(ctx->mainloop, 0);
michael@0: }
michael@0: 
michael@0: static void
michael@0: server_info_callback(pa_context * context, const pa_server_info * info, void * u)
michael@0: {
michael@0:   WRAP(pa_context_get_sink_info_by_name)(context, info->default_sink_name, sink_info_callback, u);
michael@0: }
michael@0: 
michael@0: static void
michael@0: context_state_callback(pa_context * c, void * u)
michael@0: {
michael@0:   cubeb * ctx = u;
michael@0:   if (!PA_CONTEXT_IS_GOOD(WRAP(pa_context_get_state)(c))) {
michael@0:     ctx->error = 1;
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_signal)(ctx->mainloop, 0);
michael@0: }
michael@0: 
michael@0: static void
michael@0: context_notify_callback(pa_context * c, void * u)
michael@0: {
michael@0:   cubeb * ctx = u;
michael@0:   WRAP(pa_threaded_mainloop_signal)(ctx->mainloop, 0);
michael@0: }
michael@0: 
michael@0: static void
michael@0: stream_success_callback(pa_stream * s, int success, void * u)
michael@0: {
michael@0:   cubeb_stream * stm = u;
michael@0:   WRAP(pa_threaded_mainloop_signal)(stm->context->mainloop, 0);
michael@0: }
michael@0: 
michael@0: static void
michael@0: stream_drain_callback(pa_mainloop_api * a, pa_time_event * e, struct timeval const * tv, void * u)
michael@0: {
michael@0:   cubeb_stream * stm = u;
michael@0:   /* there's no pa_rttime_free, so use this instead. */
michael@0:   a->time_free(stm->drain_timer);
michael@0:   stm->drain_timer = NULL;
michael@0:   stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_DRAINED);
michael@0: }
michael@0: 
michael@0: static void
michael@0: stream_state_callback(pa_stream * s, void * u)
michael@0: {
michael@0:   cubeb_stream * stm = u;
michael@0:   if (!PA_STREAM_IS_GOOD(WRAP(pa_stream_get_state)(s))) {
michael@0:     stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_ERROR);
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_signal)(stm->context->mainloop, 0);
michael@0: }
michael@0: 
michael@0: static void
michael@0: stream_request_callback(pa_stream * s, size_t nbytes, void * u)
michael@0: {
michael@0:   cubeb_stream * stm;
michael@0:   void * buffer;
michael@0:   size_t size;
michael@0:   int r;
michael@0:   long got;
michael@0:   size_t towrite;
michael@0:   size_t frame_size;
michael@0: 
michael@0:   stm = u;
michael@0: 
michael@0:   if (stm->shutdown)
michael@0:     return;
michael@0: 
michael@0:   frame_size = WRAP(pa_frame_size)(&stm->sample_spec);
michael@0: 
michael@0:   assert(nbytes % frame_size == 0);
michael@0: 
michael@0:   towrite = nbytes;
michael@0: 
michael@0:   while (towrite) {
michael@0:     size = towrite;
michael@0:     r = WRAP(pa_stream_begin_write)(s, &buffer, &size);
michael@0:     assert(r == 0);
michael@0:     assert(size > 0);
michael@0:     assert(size % frame_size == 0);
michael@0: 
michael@0:     got = stm->data_callback(stm, stm->user_ptr, buffer, size / frame_size);
michael@0:     if (got < 0) {
michael@0:       WRAP(pa_stream_cancel_write)(s);
michael@0:       stm->shutdown = 1;
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     r = WRAP(pa_stream_write)(s, buffer, got * frame_size, NULL, 0, PA_SEEK_RELATIVE);
michael@0:     assert(r == 0);
michael@0: 
michael@0:     if ((size_t) got < size / frame_size) {
michael@0:       pa_usec_t latency = 0;
michael@0:       r = WRAP(pa_stream_get_latency)(s, &latency, NULL);
michael@0:       if (r == -PA_ERR_NODATA) {
michael@0:         /* this needs a better guess. */
michael@0:         latency = 100 * PA_USEC_PER_MSEC;
michael@0:       }
michael@0:       assert(r == 0 || r == -PA_ERR_NODATA);
michael@0:       /* pa_stream_drain is useless, see PA bug# 866. this is a workaround. */
michael@0:       /* arbitrary safety margin: double the current latency. */
michael@0:       stm->drain_timer = WRAP(pa_context_rttime_new)(stm->context->context, WRAP(pa_rtclock_now)() + 2 * latency, stream_drain_callback, stm);
michael@0:       stm->shutdown = 1;
michael@0:       return;
michael@0:     }
michael@0: 
michael@0:     towrite -= size;
michael@0:   }
michael@0: 
michael@0:   assert(towrite == 0);
michael@0: }
michael@0: 
michael@0: static int
michael@0: wait_until_context_ready(cubeb * ctx)
michael@0: {
michael@0:   for (;;) {
michael@0:     pa_context_state_t state = WRAP(pa_context_get_state)(ctx->context);
michael@0:     if (!PA_CONTEXT_IS_GOOD(state))
michael@0:       return -1;
michael@0:     if (state == PA_CONTEXT_READY)
michael@0:       break;
michael@0:     WRAP(pa_threaded_mainloop_wait)(ctx->mainloop);
michael@0:   }
michael@0:   return 0;
michael@0: }
michael@0: 
michael@0: static int
michael@0: wait_until_stream_ready(cubeb_stream * stm)
michael@0: {
michael@0:   for (;;) {
michael@0:     pa_stream_state_t state = WRAP(pa_stream_get_state)(stm->stream);
michael@0:     if (!PA_STREAM_IS_GOOD(state))
michael@0:       return -1;
michael@0:     if (state == PA_STREAM_READY)
michael@0:       break;
michael@0:     WRAP(pa_threaded_mainloop_wait)(stm->context->mainloop);
michael@0:   }
michael@0:   return 0;
michael@0: }
michael@0: 
michael@0: static int
michael@0: operation_wait(cubeb * ctx, pa_stream * stream, pa_operation * o)
michael@0: {
michael@0:   while (WRAP(pa_operation_get_state)(o) == PA_OPERATION_RUNNING) {
michael@0:     WRAP(pa_threaded_mainloop_wait)(ctx->mainloop);
michael@0:     if (!PA_CONTEXT_IS_GOOD(WRAP(pa_context_get_state)(ctx->context)))
michael@0:       return -1;
michael@0:     if (stream && !PA_STREAM_IS_GOOD(WRAP(pa_stream_get_state)(stream)))
michael@0:       return -1;
michael@0:   }
michael@0:   return 0;
michael@0: }
michael@0: 
michael@0: static void
michael@0: stream_cork(cubeb_stream * stm, enum cork_state state)
michael@0: {
michael@0:   pa_operation * o;
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(stm->context->mainloop);
michael@0:   o = WRAP(pa_stream_cork)(stm->stream, state & CORK, stream_success_callback, stm);
michael@0:   if (o) {
michael@0:     operation_wait(stm->context, stm->stream, o);
michael@0:     WRAP(pa_operation_unref)(o);
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_unlock)(stm->context->mainloop);
michael@0: 
michael@0:   if (state & NOTIFY) {
michael@0:     stm->state_callback(stm, stm->user_ptr,
michael@0:                         state & CORK ? CUBEB_STATE_STOPPED : CUBEB_STATE_STARTED);
michael@0:   }
michael@0: }
michael@0: 
michael@0: static void pulse_context_destroy(cubeb * ctx);
michael@0: static void pulse_destroy(cubeb * ctx);
michael@0: 
michael@0: static int
michael@0: pulse_context_init(cubeb * ctx)
michael@0: {
michael@0:   if (ctx->context) {
michael@0:     assert(ctx->error == 1);
michael@0:     pulse_context_destroy(ctx);
michael@0:   }
michael@0: 
michael@0:   ctx->context = WRAP(pa_context_new)(WRAP(pa_threaded_mainloop_get_api)(ctx->mainloop),
michael@0:                                 ctx->context_name);
michael@0:   WRAP(pa_context_set_state_callback)(ctx->context, context_state_callback, ctx);
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(ctx->mainloop);
michael@0:   WRAP(pa_context_connect)(ctx->context, NULL, 0, NULL);
michael@0: 
michael@0:   if (wait_until_context_ready(ctx) != 0) {
michael@0:     WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0:     pulse_context_destroy(ctx);
michael@0:     ctx->context = NULL;
michael@0:     return -1;
michael@0:   }
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0: 
michael@0:   ctx->error = 0;
michael@0: 
michael@0:   return 0;
michael@0: }
michael@0: 
michael@0: /*static*/ int
michael@0: pulse_init(cubeb ** context, char const * context_name)
michael@0: {
michael@0:   void * libpulse = NULL;
michael@0:   cubeb * ctx;
michael@0: 
michael@0:   *context = NULL;
michael@0: 
michael@0: #ifndef DISABLE_LIBPULSE_DLOPEN
michael@0:   libpulse = dlopen("libpulse.so.0", RTLD_LAZY);
michael@0:   if (!libpulse) {
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0: #define LOAD(x) do { \
michael@0:     cubeb_##x = dlsym(libpulse, #x); \
michael@0:     if (!cubeb_##x) { \
michael@0:       dlclose(libpulse); \
michael@0:       return CUBEB_ERROR; \
michael@0:     } \
michael@0:   } while(0)
michael@0:   LOAD(pa_channel_map_init_auto);
michael@0:   LOAD(pa_context_connect);
michael@0:   LOAD(pa_context_disconnect);
michael@0:   LOAD(pa_context_drain);
michael@0:   LOAD(pa_context_get_state);
michael@0:   LOAD(pa_context_new);
michael@0:   LOAD(pa_context_rttime_new);
michael@0:   LOAD(pa_context_set_state_callback);
michael@0:   LOAD(pa_context_get_sink_info_by_name);
michael@0:   LOAD(pa_context_get_server_info);
michael@0:   LOAD(pa_context_unref);
michael@0:   LOAD(pa_frame_size);
michael@0:   LOAD(pa_operation_get_state);
michael@0:   LOAD(pa_operation_unref);
michael@0:   LOAD(pa_rtclock_now);
michael@0:   LOAD(pa_stream_begin_write);
michael@0:   LOAD(pa_stream_cancel_write);
michael@0:   LOAD(pa_stream_connect_playback);
michael@0:   LOAD(pa_stream_cork);
michael@0:   LOAD(pa_stream_disconnect);
michael@0:   LOAD(pa_stream_get_latency);
michael@0:   LOAD(pa_stream_get_state);
michael@0:   LOAD(pa_stream_get_time);
michael@0:   LOAD(pa_stream_new);
michael@0:   LOAD(pa_stream_set_state_callback);
michael@0:   LOAD(pa_stream_set_write_callback);
michael@0:   LOAD(pa_stream_unref);
michael@0:   LOAD(pa_stream_update_timing_info);
michael@0:   LOAD(pa_stream_write);
michael@0:   LOAD(pa_threaded_mainloop_free);
michael@0:   LOAD(pa_threaded_mainloop_get_api);
michael@0:   LOAD(pa_threaded_mainloop_lock);
michael@0:   LOAD(pa_threaded_mainloop_in_thread);
michael@0:   LOAD(pa_threaded_mainloop_new);
michael@0:   LOAD(pa_threaded_mainloop_signal);
michael@0:   LOAD(pa_threaded_mainloop_start);
michael@0:   LOAD(pa_threaded_mainloop_stop);
michael@0:   LOAD(pa_threaded_mainloop_unlock);
michael@0:   LOAD(pa_threaded_mainloop_wait);
michael@0:   LOAD(pa_usec_to_bytes);
michael@0: #undef LOAD
michael@0: #endif
michael@0: 
michael@0:   ctx = calloc(1, sizeof(*ctx));
michael@0:   assert(ctx);
michael@0: 
michael@0:   ctx->ops = &pulse_ops;
michael@0:   ctx->libpulse = libpulse;
michael@0: 
michael@0:   ctx->mainloop = WRAP(pa_threaded_mainloop_new)();
michael@0:   ctx->default_sink_info = NULL;
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_start)(ctx->mainloop);
michael@0: 
michael@0:   ctx->context_name = context_name ? strdup(context_name) : NULL;
michael@0:   if (pulse_context_init(ctx) != 0) {
michael@0:     pulse_destroy(ctx);
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(ctx->mainloop);
michael@0:   WRAP(pa_context_get_server_info)(ctx->context, server_info_callback, ctx);
michael@0:   WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0: 
michael@0:   *context = ctx;
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static char const *
michael@0: pulse_get_backend_id(cubeb * ctx)
michael@0: {
michael@0:   return "pulse";
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_get_max_channel_count(cubeb * ctx, uint32_t * max_channels)
michael@0: {
michael@0:   assert(ctx && max_channels);
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(ctx->mainloop);
michael@0:   while (!ctx->default_sink_info) {
michael@0:     WRAP(pa_threaded_mainloop_wait)(ctx->mainloop);
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0: 
michael@0:   *max_channels = ctx->default_sink_info->channel_map.channels;
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_get_preferred_sample_rate(cubeb * ctx, uint32_t * rate)
michael@0: {
michael@0:   assert(ctx && rate);
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(ctx->mainloop);
michael@0:   while (!ctx->default_sink_info) {
michael@0:     WRAP(pa_threaded_mainloop_wait)(ctx->mainloop);
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0: 
michael@0:   *rate = ctx->default_sink_info->sample_spec.rate;
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_get_min_latency(cubeb * ctx, cubeb_stream_params params, uint32_t * latency_ms)
michael@0: {
michael@0:   // According to PulseAudio developers, this is a safe minimum.
michael@0:   *latency_ms = 40;
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static void
michael@0: pulse_context_destroy(cubeb * ctx)
michael@0: {
michael@0:   pa_operation * o;
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(ctx->mainloop);
michael@0:   o = WRAP(pa_context_drain)(ctx->context, context_notify_callback, ctx);
michael@0:   if (o) {
michael@0:     operation_wait(ctx, NULL, o);
michael@0:     WRAP(pa_operation_unref)(o);
michael@0:   }
michael@0:   WRAP(pa_context_set_state_callback)(ctx->context, NULL, NULL);
michael@0:   WRAP(pa_context_disconnect)(ctx->context);
michael@0:   WRAP(pa_context_unref)(ctx->context);
michael@0:   WRAP(pa_threaded_mainloop_unlock)(ctx->mainloop);
michael@0: }
michael@0: 
michael@0: static void
michael@0: pulse_destroy(cubeb * ctx)
michael@0: {
michael@0:   pa_operation * o;
michael@0: 
michael@0:   if (ctx->context_name) {
michael@0:     free(ctx->context_name);
michael@0:   }
michael@0:   if (ctx->context) {
michael@0:     pulse_context_destroy(ctx);
michael@0:   }
michael@0: 
michael@0:   if (ctx->mainloop) {
michael@0:     WRAP(pa_threaded_mainloop_stop)(ctx->mainloop);
michael@0:     WRAP(pa_threaded_mainloop_free)(ctx->mainloop);
michael@0:   }
michael@0: 
michael@0:   if (ctx->libpulse) {
michael@0:     dlclose(ctx->libpulse);
michael@0:   }
michael@0:   if (ctx->default_sink_info) {
michael@0:     free(ctx->default_sink_info);
michael@0:   }
michael@0:   free(ctx);
michael@0: }
michael@0: 
michael@0: static void pulse_stream_destroy(cubeb_stream * stm);
michael@0: 
michael@0: static int
michael@0: pulse_stream_init(cubeb * context, cubeb_stream ** stream, char const * stream_name,
michael@0:                   cubeb_stream_params stream_params, unsigned int latency,
michael@0:                   cubeb_data_callback data_callback, cubeb_state_callback state_callback,
michael@0:                   void * user_ptr)
michael@0: {
michael@0:   pa_sample_spec ss;
michael@0:   cubeb_stream * stm;
michael@0:   pa_operation * o;
michael@0:   pa_buffer_attr battr;
michael@0:   int r;
michael@0: 
michael@0:   assert(context);
michael@0: 
michael@0:   *stream = NULL;
michael@0: 
michael@0:   switch (stream_params.format) {
michael@0:   case CUBEB_SAMPLE_S16LE:
michael@0:     ss.format = PA_SAMPLE_S16LE;
michael@0:     break;
michael@0:   case CUBEB_SAMPLE_S16BE:
michael@0:     ss.format = PA_SAMPLE_S16BE;
michael@0:     break;
michael@0:   case CUBEB_SAMPLE_FLOAT32LE:
michael@0:     ss.format = PA_SAMPLE_FLOAT32LE;
michael@0:     break;
michael@0:   case CUBEB_SAMPLE_FLOAT32BE:
michael@0:     ss.format = PA_SAMPLE_FLOAT32BE;
michael@0:     break;
michael@0:   default:
michael@0:     return CUBEB_ERROR_INVALID_FORMAT;
michael@0:   }
michael@0: 
michael@0:   // If the connection failed for some reason, try to reconnect
michael@0:   if (context->error == 1 && pulse_context_init(context) != 0) {
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   ss.rate = stream_params.rate;
michael@0:   ss.channels = stream_params.channels;
michael@0: 
michael@0:   stm = calloc(1, sizeof(*stm));
michael@0:   assert(stm);
michael@0: 
michael@0:   stm->context = context;
michael@0: 
michael@0:   stm->data_callback = data_callback;
michael@0:   stm->state_callback = state_callback;
michael@0:   stm->user_ptr = user_ptr;
michael@0: 
michael@0:   stm->sample_spec = ss;
michael@0: 
michael@0:   battr.maxlength = -1;
michael@0:   battr.tlength = WRAP(pa_usec_to_bytes)(latency * PA_USEC_PER_MSEC, &stm->sample_spec);
michael@0:   battr.prebuf = -1;
michael@0:   battr.minreq = battr.tlength / 4;
michael@0:   battr.fragsize = -1;
michael@0: 
michael@0:   WRAP(pa_threaded_mainloop_lock)(stm->context->mainloop);
michael@0:   stm->stream = WRAP(pa_stream_new)(stm->context->context, stream_name, &ss, NULL);
michael@0:   if (!stm->stream) {
michael@0:     pulse_stream_destroy(stm);
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0:   WRAP(pa_stream_set_state_callback)(stm->stream, stream_state_callback, stm);
michael@0:   WRAP(pa_stream_set_write_callback)(stm->stream, stream_request_callback, stm);
michael@0:   WRAP(pa_stream_connect_playback)(stm->stream, NULL, &battr,
michael@0:                              PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_INTERPOLATE_TIMING |
michael@0:                              PA_STREAM_START_CORKED,
michael@0:                              NULL, NULL);
michael@0: 
michael@0:   r = wait_until_stream_ready(stm);
michael@0:   if (r == 0) {
michael@0:     /* force a timing update now, otherwise timing info does not become valid
michael@0:        until some point after initialization has completed. */
michael@0:     o = WRAP(pa_stream_update_timing_info)(stm->stream, stream_success_callback, stm);
michael@0:     if (o) {
michael@0:       r = operation_wait(stm->context, stm->stream, o);
michael@0:       WRAP(pa_operation_unref)(o);
michael@0:     }
michael@0:   }
michael@0:   WRAP(pa_threaded_mainloop_unlock)(stm->context->mainloop);
michael@0: 
michael@0:   if (r != 0) {
michael@0:     pulse_stream_destroy(stm);
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   *stream = stm;
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static void
michael@0: pulse_stream_destroy(cubeb_stream * stm)
michael@0: {
michael@0:   if (stm->stream) {
michael@0:     stream_cork(stm, CORK);
michael@0: 
michael@0:     WRAP(pa_threaded_mainloop_lock)(stm->context->mainloop);
michael@0: 
michael@0:     if (stm->drain_timer) {
michael@0:       /* there's no pa_rttime_free, so use this instead. */
michael@0:       WRAP(pa_threaded_mainloop_get_api)(stm->context->mainloop)->time_free(stm->drain_timer);
michael@0:     }
michael@0: 
michael@0:     WRAP(pa_stream_set_state_callback)(stm->stream, NULL, NULL);
michael@0:     WRAP(pa_stream_disconnect)(stm->stream);
michael@0:     WRAP(pa_stream_unref)(stm->stream);
michael@0:     WRAP(pa_threaded_mainloop_unlock)(stm->context->mainloop);
michael@0:   }
michael@0: 
michael@0:   free(stm);
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_stream_start(cubeb_stream * stm)
michael@0: {
michael@0:   stream_cork(stm, UNCORK | NOTIFY);
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_stream_stop(cubeb_stream * stm)
michael@0: {
michael@0:   stream_cork(stm, CORK | NOTIFY);
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static int
michael@0: pulse_stream_get_position(cubeb_stream * stm, uint64_t * position)
michael@0: {
michael@0:   int r, in_thread;
michael@0:   pa_usec_t r_usec;
michael@0:   uint64_t bytes;
michael@0: 
michael@0:   in_thread = WRAP(pa_threaded_mainloop_in_thread)(stm->context->mainloop);
michael@0: 
michael@0:   if (!in_thread) {
michael@0:     WRAP(pa_threaded_mainloop_lock)(stm->context->mainloop);
michael@0:   }
michael@0:   r = WRAP(pa_stream_get_time)(stm->stream, &r_usec);
michael@0:   if (!in_thread) {
michael@0:     WRAP(pa_threaded_mainloop_unlock)(stm->context->mainloop);
michael@0:   }
michael@0: 
michael@0:   if (r != 0) {
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   bytes = WRAP(pa_usec_to_bytes)(r_usec, &stm->sample_spec);
michael@0:   *position = bytes / WRAP(pa_frame_size)(&stm->sample_spec);
michael@0: 
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: int
michael@0: pulse_stream_get_latency(cubeb_stream * stm, uint32_t * latency)
michael@0: {
michael@0:   pa_usec_t r_usec;
michael@0:   int negative, r;
michael@0: 
michael@0:   if (!stm) {
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   r = WRAP(pa_stream_get_latency)(stm->stream, &r_usec, &negative);
michael@0:   assert(!negative);
michael@0:   if (r) {
michael@0:     return CUBEB_ERROR;
michael@0:   }
michael@0: 
michael@0:   *latency = r_usec * stm->sample_spec.rate / PA_USEC_PER_SEC;
michael@0:   return CUBEB_OK;
michael@0: }
michael@0: 
michael@0: static struct cubeb_ops const pulse_ops = {
michael@0:   .init = pulse_init,
michael@0:   .get_backend_id = pulse_get_backend_id,
michael@0:   .get_max_channel_count = pulse_get_max_channel_count,
michael@0:   .get_min_latency = pulse_get_min_latency,
michael@0:   .get_preferred_sample_rate = pulse_get_preferred_sample_rate,
michael@0:   .destroy = pulse_destroy,
michael@0:   .stream_init = pulse_stream_init,
michael@0:   .stream_destroy = pulse_stream_destroy,
michael@0:   .stream_start = pulse_stream_start,
michael@0:   .stream_stop = pulse_stream_stop,
michael@0:   .stream_get_position = pulse_stream_get_position,
michael@0:   .stream_get_latency = pulse_stream_get_latency
michael@0: };