rdpsnd_pulse.c

 
#include <freerdp/config.h>
#include <freerdp/utils/helpers.h>
 
#include <errno.h>
 
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
 
#include <winpr/crt.h>
#include <winpr/cast.h>
#include <winpr/assert.h>
#include <winpr/stream.h>
#include <winpr/cmdline.h>
#include <winpr/sysinfo.h>
 
#include <pulse/pulseaudio.h>
 
#include <freerdp/types.h>
#include <freerdp/codec/dsp.h>
 
#include "rdpsnd_main.h"
 
/* target latency when /sound:latency is not given */
#define PULSE_DEFAULT_LATENCY_MS 100
 
typedef struct
{
  rdpsndDevicePlugin device;
 
  char* device_name;
  char* client_name;
  char* stream_name;
  pa_threaded_mainloop* mainloop;
  pa_context* context;
  pa_sample_spec sample_spec;
  pa_stream* stream;
  UINT32 latency;
  UINT32 volume;
  time_t reconnect_delay_seconds;
  time_t reconnect_time;
  UINT64 play_end; /* tick count when the queued audio has been played */
} rdpsndPulsePlugin;
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_check_pulse(rdpsndPulsePlugin* pulse, BOOL haveStream)
{
  BOOL rc = TRUE;
  WINPR_ASSERT(pulse);
 
  if (!pulse->context)
  {
    WLog_WARN(TAG, "pulse->context=nullptr");
    rc = FALSE;
  }
 
  if (haveStream)
  {
    if (!pulse->stream)
    {
      WLog_WARN(TAG, "pulse->stream=%p", WINPR_CXX_COMPAT_CAST(const void*, pulse->stream));
      rc = FALSE;
    }
  }
 
  if (!pulse->mainloop)
  {
    WLog_WARN(TAG, "pulse->mainloop=%p", WINPR_CXX_COMPAT_CAST(const void*, pulse->mainloop));
    rc = FALSE;
  }
 
  return rc;
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_format_supported(WINPR_ATTR_UNUSED rdpsndDevicePlugin* device,
                                          const AUDIO_FORMAT* format);
 
static void rdpsnd_pulse_get_sink_info(WINPR_ATTR_UNUSED pa_context* c, const pa_sink_info* i,
                                       WINPR_ATTR_UNUSED int eol, void* userdata)
{
  UINT16 dwVolumeLeft = ((50 * 0xFFFF) / 100);  /* 50% */
  UINT16 dwVolumeRight = ((50 * 0xFFFF) / 100); /* 50% */
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)userdata;
 
  WINPR_ASSERT(c);
  if (!rdpsnd_check_pulse(pulse, FALSE) || !i)
    return;
 
  for (uint8_t x = 0; x < i->volume.channels; x++)
  {
    pa_volume_t volume = i->volume.values[x];
 
    if (volume >= PA_VOLUME_NORM)
      volume = PA_VOLUME_NORM - 1;
 
    switch (x)
    {
      case 0:
        dwVolumeLeft = (UINT16)volume;
        break;
 
      case 1:
        dwVolumeRight = (UINT16)volume;
        break;
 
      default:
        break;
    }
  }
 
  pulse->volume = ((UINT32)dwVolumeLeft << 16U) | dwVolumeRight;
}
 
static void rdpsnd_pulse_context_state_callback(pa_context* context, void* userdata)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)userdata;
 
  WINPR_ASSERT(context);
  WINPR_ASSERT(pulse);
 
  pa_context_state_t state = pa_context_get_state(context);
 
  switch (state)
  {
    case PA_CONTEXT_READY:
      pa_threaded_mainloop_signal(pulse->mainloop, 0);
      break;
 
    case PA_CONTEXT_FAILED:
      // Destroy context now, create new one for next connection attempt
      pa_context_unref(pulse->context);
      pulse->context = nullptr;
      if (pulse->reconnect_delay_seconds >= 0)
        pulse->reconnect_time = time(nullptr) + pulse->reconnect_delay_seconds;
      pa_threaded_mainloop_signal(pulse->mainloop, 0);
      break;
 
    case PA_CONTEXT_TERMINATED:
      pa_threaded_mainloop_signal(pulse->mainloop, 0);
      break;
 
    default:
      break;
  }
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_connect(rdpsndDevicePlugin* device)
{
  BOOL rc = 0;
  pa_operation* o = nullptr;
  pa_context_state_t state = PA_CONTEXT_FAILED;
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!rdpsnd_check_pulse(pulse, FALSE))
    return FALSE;
 
  pa_threaded_mainloop_lock(pulse->mainloop);
 
  if (pa_context_connect(pulse->context, nullptr, PA_CONTEXT_NOFLAGS, nullptr) < 0)
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    return FALSE;
  }
 
  /* the context state callback releases pulse->context on PA_CONTEXT_FAILED,
   * so it must be re-checked after every wait */
  while (pulse->context)
  {
    state = pa_context_get_state(pulse->context);
 
    if (state == PA_CONTEXT_READY)
      break;
 
    if (!PA_CONTEXT_IS_GOOD(state))
    {
      break;
    }
 
    pa_threaded_mainloop_wait(pulse->mainloop);
  }
 
  if (pulse->context)
  {
    o = pa_context_get_sink_info_by_index(pulse->context, 0, rdpsnd_pulse_get_sink_info, pulse);
 
    if (o)
      pa_operation_unref(o);
  }
 
  if (pulse->context && (state == PA_CONTEXT_READY))
  {
    rc = TRUE;
  }
  else
  {
    if (pulse->context)
      pa_context_disconnect(pulse->context);
    rc = FALSE;
  }
 
  pa_threaded_mainloop_unlock(pulse->mainloop);
  return rc;
}
 
static void rdpsnd_pulse_stream_success_callback(WINPR_ATTR_UNUSED pa_stream* stream,
                                                 WINPR_ATTR_UNUSED int success, void* userdata)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)userdata;
 
  if (!rdpsnd_check_pulse(pulse, TRUE))
    return;
 
  pa_threaded_mainloop_signal(pulse->mainloop, 0);
}
 
static void rdpsnd_pulse_wait_for_operation(rdpsndPulsePlugin* pulse, pa_operation* operation)
{
  if (!rdpsnd_check_pulse(pulse, TRUE))
    return;
 
  if (!operation)
    return;
 
  while (pa_operation_get_state(operation) == PA_OPERATION_RUNNING)
  {
    pa_threaded_mainloop_wait(pulse->mainloop);
  }
 
  pa_operation_unref(operation);
}
 
static void rdpsnd_pulse_stream_state_callback(pa_stream* stream, void* userdata)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)userdata;
 
  WINPR_ASSERT(stream);
  if (!rdpsnd_check_pulse(pulse, TRUE))
    return;
 
  pa_stream_state_t state = pa_stream_get_state(stream);
 
  switch (state)
  {
    case PA_STREAM_READY:
      pa_threaded_mainloop_signal(pulse->mainloop, 0);
      break;
 
    case PA_STREAM_FAILED:
    case PA_STREAM_TERMINATED:
      /* The stream is dead (e.g. its device vanished). Drop our
       * reference; pa_stream dispatches state callbacks under its own
       * ref/unref guard, so unref here is safe. */
      pa_stream_unref(pulse->stream);
      pulse->stream = nullptr;
      pa_threaded_mainloop_signal(pulse->mainloop, 0);
      break;
 
    default:
      break;
  }
}
 
static void rdpsnd_pulse_stream_request_callback(WINPR_ATTR_UNUSED pa_stream* stream,
                                                 WINPR_ATTR_UNUSED size_t length, void* userdata)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)userdata;
 
  WINPR_ASSERT(stream);
  if (!rdpsnd_check_pulse(pulse, TRUE))
    return;
 
  pa_threaded_mainloop_signal(pulse->mainloop, 0);
}
 
static void rdpsnd_pulse_close(rdpsndDevicePlugin* device)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  WINPR_ASSERT(pulse);
 
  if (!rdpsnd_check_pulse(pulse, FALSE))
    return;
 
  pa_threaded_mainloop_lock(pulse->mainloop);
  if (pulse->stream)
  {
    rdpsnd_pulse_wait_for_operation(
        pulse, pa_stream_drain(pulse->stream, rdpsnd_pulse_stream_success_callback, pulse));
    /* The stream may have died while draining; the state callback then
     * released it and cleared pulse->stream. */
    if (pulse->stream)
    {
      pa_stream_disconnect(pulse->stream);
      pa_stream_unref(pulse->stream);
      pulse->stream = nullptr;
    }
  }
  pa_threaded_mainloop_unlock(pulse->mainloop);
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_set_format_spec(rdpsndPulsePlugin* pulse, const AUDIO_FORMAT* format)
{
  WINPR_ASSERT(format);
 
  if (!rdpsnd_check_pulse(pulse, FALSE))
    return FALSE;
 
  if (!rdpsnd_pulse_format_supported(&pulse->device, format))
    return FALSE;
 
  pa_sample_format_t sformat = PA_SAMPLE_INVALID;
  switch (format->wFormatTag)
  {
    case WAVE_FORMAT_PCM:
      switch (format->wBitsPerSample)
      {
        case 8:
          sformat = PA_SAMPLE_U8;
          break;
 
        case 16:
          sformat = PA_SAMPLE_S16LE;
          break;
 
        default:
          return FALSE;
      }
 
      break;
 
    default:
      return FALSE;
  }
 
  const pa_sample_spec sample_spec = { .format = sformat,
                                     .rate = format->nSamplesPerSec,
                                     .channels =
                                         WINPR_ASSERTING_INT_CAST(uint8_t, format->nChannels) };
  pulse->sample_spec = sample_spec;
  return TRUE;
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_context_connect(rdpsndDevicePlugin* device)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
  WINPR_ASSERT(pulse);
 
  pulse->context =
      pa_context_new(pa_threaded_mainloop_get_api(pulse->mainloop), pulse->client_name);
 
  if (!pulse->context)
    return FALSE;
 
  pa_context_set_state_callback(pulse->context, rdpsnd_pulse_context_state_callback, pulse);
 
  return rdpsnd_pulse_connect(&pulse->device);
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_open_stream(rdpsndDevicePlugin* device)
{
  pa_stream_state_t state = PA_STREAM_FAILED;
  int flags = PA_STREAM_NOFLAGS;
  pa_buffer_attr buffer_attr = WINPR_C_ARRAY_INIT;
  char ss[PA_SAMPLE_SPEC_SNPRINT_MAX] = WINPR_C_ARRAY_INIT;
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
  WINPR_ASSERT(pulse);
 
  pulse->play_end = 0; /* a new stream starts empty */
  if (pa_sample_spec_valid(&pulse->sample_spec) == 0)
  {
    pa_sample_spec_snprint(ss, sizeof(ss), &pulse->sample_spec);
    return FALSE;
  }
 
  pa_threaded_mainloop_lock(pulse->mainloop);
  if (!pulse->context)
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    if (pulse->reconnect_delay_seconds >= 0 && time(nullptr) - pulse->reconnect_time >= 0)
    {
      if (!rdpsnd_pulse_context_connect(device))
        return FALSE;
    }
    pa_threaded_mainloop_lock(pulse->mainloop);
  }
 
  if (!rdpsnd_check_pulse(pulse, FALSE))
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    return FALSE;
  }
 
  pulse->stream = pa_stream_new(pulse->context, pulse->stream_name, &pulse->sample_spec, nullptr);
 
  if (!pulse->stream)
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    return FALSE;
  }
 
  /* register essential callbacks */
  pa_stream_set_state_callback(pulse->stream, rdpsnd_pulse_stream_state_callback, pulse);
  pa_stream_set_write_callback(pulse->stream, rdpsnd_pulse_stream_request_callback, pulse);
  flags = PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_ADJUST_LATENCY;
 
  /* Always ask for a target latency: without one the server picks its
   * own buffering, and after a pause PulseAudio drops the start of the
   * next sound (as much as its sink latency) while PipeWire runs dry
   * shortly after the start */
  const UINT32 ms = (pulse->latency > 0) ? pulse->latency : PULSE_DEFAULT_LATENCY_MS;
  const size_t val = pa_usec_to_bytes(1000ULL * ms, &pulse->sample_spec);
  buffer_attr.maxlength = UINT32_MAX;
  buffer_attr.tlength = (val > UINT32_MAX) ? UINT32_MAX : (UINT32)val;
  buffer_attr.prebuf = UINT32_MAX;
  buffer_attr.minreq = UINT32_MAX;
  buffer_attr.fragsize = UINT32_MAX;
 
  // NOLINTNEXTLINE(clang-analyzer-optin.core.EnumCastOutOfRange)
  pa_stream_flags_t eflags = (pa_stream_flags_t)flags;
  if (pa_stream_connect_playback(pulse->stream, pulse->device_name, &buffer_attr, eflags, nullptr,
                                 nullptr) < 0)
  {
    WLog_ERR(TAG, "error connecting playback stream");
    pa_stream_unref(pulse->stream);
    pulse->stream = nullptr;
    pa_threaded_mainloop_unlock(pulse->mainloop);
    return FALSE;
  }
 
  while (pulse->stream)
  {
    state = pa_stream_get_state(pulse->stream);
 
    if (state == PA_STREAM_READY)
      break;
 
    if (!PA_STREAM_IS_GOOD(state))
    {
      break;
    }
 
    pa_threaded_mainloop_wait(pulse->mainloop);
  }
 
  pa_threaded_mainloop_unlock(pulse->mainloop);
 
  if (state == PA_STREAM_READY)
    return TRUE;
 
  rdpsnd_pulse_close(device);
  return FALSE;
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_open(rdpsndDevicePlugin* device, const AUDIO_FORMAT* format,
                              UINT32 latency)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  WINPR_ASSERT(format);
 
  if (!rdpsnd_check_pulse(pulse, FALSE))
    return TRUE;
 
  if (!rdpsnd_pulse_set_format_spec(pulse, format))
    return FALSE;
 
  pulse->latency = latency;
 
  return rdpsnd_pulse_open_stream(device);
}
 
static void rdpsnd_pulse_free(rdpsndDevicePlugin* device)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!pulse)
    return;
 
  rdpsnd_pulse_close(device);
 
  if (pulse->mainloop)
    pa_threaded_mainloop_stop(pulse->mainloop);
 
  if (pulse->context)
  {
    pa_context_disconnect(pulse->context);
    pa_context_unref(pulse->context);
    pulse->context = nullptr;
  }
 
  if (pulse->mainloop)
  {
    pa_threaded_mainloop_free(pulse->mainloop);
    pulse->mainloop = nullptr;
  }
 
  free(pulse->device_name);
  free(pulse->client_name);
  free(pulse->stream_name);
  free(pulse);
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_default_format(rdpsndDevicePlugin* device, const AUDIO_FORMAT* desired,
                                        AUDIO_FORMAT* defaultFormat)
{
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!pulse || !defaultFormat)
    return FALSE;
 
  *defaultFormat = *desired;
  defaultFormat->data = nullptr;
  defaultFormat->cbSize = 0;
  defaultFormat->wFormatTag = WAVE_FORMAT_PCM;
  if ((defaultFormat->nChannels < 1) || (defaultFormat->nChannels > PA_CHANNELS_MAX))
    defaultFormat->nChannels = 2;
  if ((defaultFormat->nSamplesPerSec < 1) || (defaultFormat->nSamplesPerSec > PA_RATE_MAX))
    defaultFormat->nSamplesPerSec = 44100;
  if ((defaultFormat->wBitsPerSample != 8) && (defaultFormat->wBitsPerSample != 16))
    defaultFormat->wBitsPerSample = 16;
 
  defaultFormat->nBlockAlign = defaultFormat->nChannels * defaultFormat->wBitsPerSample / 8;
  defaultFormat->nAvgBytesPerSec = defaultFormat->nBlockAlign * defaultFormat->nSamplesPerSec;
  return TRUE;
}
 
BOOL rdpsnd_pulse_format_supported(WINPR_ATTR_UNUSED rdpsndDevicePlugin* device,
                                   const AUDIO_FORMAT* format)
{
  WINPR_ASSERT(device);
  WINPR_ASSERT(format);
 
  switch (format->wFormatTag)
  {
    case WAVE_FORMAT_PCM:
      if (format->cbSize == 0 && (format->nSamplesPerSec <= PA_RATE_MAX) &&
          (format->wBitsPerSample == 8 || format->wBitsPerSample == 16) &&
          (format->nChannels >= 1 && format->nChannels <= PA_CHANNELS_MAX))
      {
        return TRUE;
      }
 
      break;
 
    default:
      break;
  }
 
  return FALSE;
}
 
WINPR_ATTR_NODISCARD
static UINT32 rdpsnd_pulse_get_volume(rdpsndDevicePlugin* device)
{
  pa_operation* o = nullptr;
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!pulse || !pulse->mainloop)
    return 0;
 
  pa_threaded_mainloop_lock(pulse->mainloop);
  /* re-check under the lock, the context is released if the connection dies */
  if (rdpsnd_check_pulse(pulse, FALSE))
  {
    o = pa_context_get_sink_info_by_index(pulse->context, 0, rdpsnd_pulse_get_sink_info, pulse);
    if (o)
      pa_operation_unref(o);
  }
  pa_threaded_mainloop_unlock(pulse->mainloop);
  return pulse->volume;
}
 
static void rdpsnd_set_volume_success_cb(WINPR_ATTR_UNUSED pa_context* c, int success,
                                         void* userdata)
{
  rdpsndPulsePlugin* pulse = userdata;
 
  if (!rdpsnd_check_pulse(pulse, TRUE))
    return;
  WINPR_ASSERT(c);
 
  WLog_INFO(TAG, "%d", success);
}
 
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_set_volume(rdpsndDevicePlugin* device, UINT32 value)
{
  pa_cvolume cv = WINPR_C_ARRAY_INIT;
  pa_volume_t left = 0;
  pa_volume_t right = 0;
  pa_operation* operation = nullptr;
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!pulse || !pulse->mainloop)
  {
    WLog_WARN(TAG, "called before pulse backend was initialized");
    return FALSE;
  }
 
  left = (pa_volume_t)(value & 0xFFFF);
  right = (pa_volume_t)((value >> 16) & 0xFFFF);
  pa_cvolume_init(&cv);
  cv.channels = 2;
  cv.values[0] = PA_VOLUME_MUTED + (left * (PA_VOLUME_NORM - PA_VOLUME_MUTED)) / PA_VOLUME_NORM;
  cv.values[1] = PA_VOLUME_MUTED + (right * (PA_VOLUME_NORM - PA_VOLUME_MUTED)) / PA_VOLUME_NORM;
  pa_threaded_mainloop_lock(pulse->mainloop);
 
  /* the stream state must be inspected under the mainloop lock - it may be
   * cleared by the state callback when the stream dies */
  if (!rdpsnd_check_pulse(pulse, TRUE))
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    WLog_WARN(TAG, "no pulse stream, not setting volume");
    return FALSE;
  }
 
  operation = pa_context_set_sink_input_volume(pulse->context, pa_stream_get_index(pulse->stream),
                                               &cv, rdpsnd_set_volume_success_cb, pulse);
 
  if (operation)
    pa_operation_unref(operation);
 
  pa_threaded_mainloop_unlock(pulse->mainloop);
  return TRUE;
}
 
/* write size bytes of data, or of silence if data is nullptr; called with the mainloop locked */
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_write(rdpsndPulsePlugin* pulse, const BYTE* data, size_t size)
{
  const size_t frame_size = pa_frame_size(&pulse->sample_spec);
  if (frame_size == 0)
    return FALSE;
 
  /* pa_stream_write only accepts whole frames, drop an incomplete one at the end */
  size -= size % frame_size;
 
  while (size > 0)
  {
    size_t length = size;
    void* pa_data = nullptr;
 
    if (pa_stream_begin_write(pulse->stream, &pa_data, &length) < 0)
      return FALSE;
 
    /* The returned length is the size of the buffer handed out, which is
     * not bound by what was asked for; never copy more than we have. */
    if (length > size)
      length = size;
 
    /* A suspended or migrating stream (e.g. the default device just
     * changed) can hand out an empty buffer; treating it as progress
     * busy-loops forever with the mainloop lock held, which also stalls
     * the channel thread calling Play. Drop the rest of the sample
     * instead. */
    if (!pa_data || (length == 0))
    {
      if (pa_data)
        pa_stream_cancel_write(pulse->stream);
      WLog_DBG(TAG, "dropping %" PRIuz " bytes, no buffer space", size);
      return FALSE;
    }
 
    if (data)
      memcpy(pa_data, data, length);
    else
      memset(pa_data, (pulse->sample_spec.format == PA_SAMPLE_U8) ? 0x80 : 0, length);
 
    /* A buffer that is neither written nor cancelled stays attached to the
     * stream and is handed out again, with its own size, by the next
     * pa_stream_begin_write. */
    if (pa_stream_write(pulse->stream, pa_data, length, nullptr, 0LL, PA_SEEK_RELATIVE) < 0)
    {
      pa_stream_cancel_write(pulse->stream);
      return FALSE;
    }
 
    if (data)
      data += length;
    size -= length;
  }
  return TRUE;
}
 
/* whether the queued audio has run out: the tick count estimate runs ahead of the device whenever
 * playback stalls (prebuffering after an underrun, bursts, clock drift); the stream's read and
 * write positions say how much is really queued, the estimate is only trusted when they are not
 * known */
WINPR_ATTR_NODISCARD
static BOOL rdpsnd_pulse_ran_dry(rdpsndPulsePlugin* pulse, UINT64 now)
{
  if (now < pulse->play_end)
    return FALSE;
 
  const pa_timing_info* ti = pa_stream_get_timing_info(pulse->stream);
  if (!ti || ti->write_index_corrupt || ti->read_index_corrupt)
    return TRUE;
 
  int64_t queued = ti->write_index - ti->read_index;
  if (ti->playing)
  {
    /* the positions are from the last timing update; take off what has been played since */
    const pa_usec_t age = pa_timeval_age(&ti->timestamp);
    queued -= (int64_t)pa_usec_to_bytes(age, &pulse->sample_spec);
  }
 
  /* less than 20 ms left: the device is about to run dry, as the estimate says */
  if (queued <= (int64_t)pa_usec_to_bytes(20000ULL, &pulse->sample_spec))
    return TRUE;
 
  /* audio is still waiting: no silence, and the estimate follows the stream again */
  pulse->play_end = now + pa_bytes_to_usec((uint64_t)queued, &pulse->sample_spec) / 1000ULL;
  return FALSE;
}
 
WINPR_ATTR_NODISCARD
static UINT rdpsnd_pulse_play(rdpsndDevicePlugin* device, const BYTE* data, size_t size)
{
  pa_usec_t latency = 0;
  int negative = 0;
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)device;
 
  if (!data)
    return 0;
 
  pa_threaded_mainloop_lock(pulse->mainloop);
 
  if (!rdpsnd_check_pulse(pulse, TRUE))
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    // Discard this playback request and just attempt to reconnect the stream
    WLog_DBG(TAG, "reconnecting playback stream");
    if (!rdpsnd_pulse_open_stream(device))
      WLog_WARN(TAG, "Failed to (re)open pulse stream, ignoring playback");
    return 0;
  }
 
  /* When everything queued so far has been played, the stream starts
   * again with this sample and nothing behind it: PipeWire plays it as
   * soon as one quantum is there, so the next sample has to be on time,
   * and PulseAudio loses some of it. Put some silence in front, so the
   * following samples can arrive a little late. */
  const UINT64 now = GetTickCount64();
  if (rdpsnd_pulse_ran_dry(pulse, now))
  {
    const UINT32 ms = (pulse->latency > 0) ? pulse->latency : PULSE_DEFAULT_LATENCY_MS;
    if (!rdpsnd_pulse_write(pulse, nullptr,
                            pa_usec_to_bytes(1000ULL * ms, &pulse->sample_spec)))
      WLog_WARN(TAG, "rdpsnd_pulse_write failed");
    pulse->play_end = now + ms;
  }
 
  if (rdpsnd_pulse_write(pulse, data, size))
    pulse->play_end += pa_bytes_to_usec(size, &pulse->sample_spec) / 1000;
 
  if (pa_stream_get_latency(pulse->stream, &latency, &negative) != 0)
    latency = 0;
 
  pa_threaded_mainloop_unlock(pulse->mainloop);
 
  const pa_usec_t val = latency / 1000;
  if (val > UINT32_MAX)
    return UINT32_MAX;
  return (UINT32)val;
}
 
WINPR_ATTR_NODISCARD
static UINT rdpsnd_pulse_parse_addin_args(rdpsndPulsePlugin* pulse, const ADDIN_ARGV* args)
{
  COMMAND_LINE_ARGUMENT_A rdpsnd_pulse_args[] = {
    { "dev", COMMAND_LINE_VALUE_REQUIRED, "<device>", nullptr, nullptr, -1, nullptr, "device" },
    { "reconnect_delay_seconds", COMMAND_LINE_VALUE_REQUIRED, "<reconnect_delay_seconds>",
      nullptr, nullptr, -1, nullptr, "reconnect_delay_seconds" },
    { "client_name", COMMAND_LINE_VALUE_REQUIRED, "<client_name>", nullptr, nullptr, -1,
      nullptr, "name of pulse client" },
    { "stream_name", COMMAND_LINE_VALUE_REQUIRED, "<stream_name>", nullptr, nullptr, -1,
      nullptr, "name of pulse stream" },
    { nullptr, 0, nullptr, nullptr, nullptr, -1, nullptr, nullptr }
  };
  const DWORD flags =
      COMMAND_LINE_SIGIL_NONE | COMMAND_LINE_SEPARATOR_COLON | COMMAND_LINE_IGN_UNKNOWN_KEYWORD;
 
  WINPR_ASSERT(pulse);
  WINPR_ASSERT(args);
 
  const int status = CommandLineParseArgumentsA(args->argc, args->argv, rdpsnd_pulse_args, flags,
                                                pulse, nullptr, nullptr);
 
  if (status < 0)
    return ERROR_INVALID_DATA;
 
  const COMMAND_LINE_ARGUMENT_A* arg = rdpsnd_pulse_args;
 
  const char* client_name = nullptr;
  const char* stream_name = nullptr;
  do
  {
    if (!(arg->Flags & COMMAND_LINE_VALUE_PRESENT))
      continue;
 
    CommandLineSwitchStart(arg) CommandLineSwitchCase(arg, "dev")
    {
      pulse->device_name = _strdup(arg->Value);
 
      if (!pulse->device_name)
        return ERROR_OUTOFMEMORY;
    }
    CommandLineSwitchCase(arg, "reconnect_delay_seconds")
    {
      unsigned long val = strtoul(arg->Value, nullptr, 0);
 
      if ((errno != 0) || (val > INT32_MAX))
        return ERROR_INVALID_DATA;
 
      pulse->reconnect_delay_seconds = (time_t)val;
    }
    CommandLineSwitchCase(arg, "client_name")
    {
      client_name = arg->Value;
    }
    CommandLineSwitchCase(arg, "stream_name")
    {
      stream_name = arg->Value;
    }
    CommandLineSwitchEnd(arg)
  } while ((arg = CommandLineFindNextArgumentA(arg)) != nullptr);
 
  if (!client_name)
    client_name = freerdp_getApplicationDetailsString();
  if (!stream_name)
    stream_name = freerdp_getApplicationDetailsString();
 
  pulse->client_name = _strdup(client_name);
  pulse->stream_name = _strdup(stream_name);
  if (!pulse->client_name || !pulse->stream_name)
    return ERROR_OUTOFMEMORY;
  return CHANNEL_RC_OK;
}
 
FREERDP_ENTRY_POINT(UINT VCAPITYPE pulse_freerdp_rdpsnd_client_subsystem_entry(
    PFREERDP_RDPSND_DEVICE_ENTRY_POINTS pEntryPoints))
{
  WINPR_ASSERT(pEntryPoints);
 
  rdpsndPulsePlugin* pulse = (rdpsndPulsePlugin*)calloc(1, sizeof(rdpsndPulsePlugin));
 
  if (!pulse)
    return CHANNEL_RC_NO_MEMORY;
 
  pulse->device.Open = rdpsnd_pulse_open;
  pulse->device.FormatSupported = rdpsnd_pulse_format_supported;
  pulse->device.GetVolume = rdpsnd_pulse_get_volume;
  pulse->device.SetVolume = rdpsnd_pulse_set_volume;
  pulse->device.Play = rdpsnd_pulse_play;
  pulse->device.Close = rdpsnd_pulse_close;
  pulse->device.Free = rdpsnd_pulse_free;
  pulse->device.DefaultFormat = rdpsnd_pulse_default_format;
 
  const ADDIN_ARGV* args = pEntryPoints->args;
  UINT ret = rdpsnd_pulse_parse_addin_args(pulse, args);
 
  if (ret != CHANNEL_RC_OK)
  {
    WLog_ERR(TAG, "error parsing arguments");
    goto error;
  }
 
  pulse->reconnect_delay_seconds = 5;
  pulse->reconnect_time = time(nullptr);
 
  ret = CHANNEL_RC_NO_MEMORY;
  pulse->mainloop = pa_threaded_mainloop_new();
 
  if (!pulse->mainloop)
    goto error;
 
  pa_threaded_mainloop_lock(pulse->mainloop);
 
  if (pa_threaded_mainloop_start(pulse->mainloop) < 0)
  {
    pa_threaded_mainloop_unlock(pulse->mainloop);
    goto error;
  }
 
  pa_threaded_mainloop_unlock(pulse->mainloop);
 
  if (!rdpsnd_pulse_context_connect((rdpsndDevicePlugin*)pulse))
    goto error;
 
  pEntryPoints->pRegisterRdpsndDevice(pEntryPoints->rdpsnd, (rdpsndDevicePlugin*)pulse);
  return CHANNEL_RC_OK;
error:
  rdpsnd_pulse_free((rdpsndDevicePlugin*)pulse);
  return ret;
}