static void __pulseaudio_stream_write_cb(pa_stream *stream, size_t length, void *user_data)
{
	sf_count_t read_length;
	short *data;
	SOUND_INFO *info = NULL;

	mmf_return_if_fail(user_data);

	info = (SOUND_INFO *)user_data;

	_mmcam_dbg_log("START");

	data = pa_xmalloc(length);

	read_length = (sf_count_t)(length/pa_frame_size(&(info->sample_spec)));

	if ((sf_readf_short(info->infile, data, read_length)) != read_length) {
		pa_xfree(data);
		return;
	}

	pa_stream_write(stream, data, length, pa_xfree, 0, PA_SEEK_RELATIVE);

	info->sample_length -= length;

	if (info->sample_length <= 0) {
		pa_stream_set_write_callback(info->sample_stream, NULL, NULL);
		pa_stream_finish_upload(info->sample_stream);

		pa_threaded_mainloop_signal(info->pulse_mainloop, 0);
		_mmcam_dbg_log("send signal DONE");
	}

	_mmcam_dbg_log("DONE read_length %d", read_length);
}

int pa_sound_file_too_big_to_cache(const char *fname) {

    SNDFILE*sf = NULL;
    SF_INFO sfi;
    pa_sample_spec ss;

    pa_assert(fname);

    pa_zero(sfi);
    if (!(sf = sf_open(fname, SFM_READ, &sfi))) {
        pa_log("Failed to open file %s", fname);
        return -1;
    }

    if (pa_sndfile_read_sample_spec(sf, &ss) < 0) {
        pa_log("Failed to determine file sample format.");
        sf_close(sf);
        return -1;
    }

    sf_close(sf);

    if ((pa_frame_size(&ss) * (size_t) sfi.frames) > PA_SCACHE_ENTRY_SIZE_MAX) {
        pa_log("File too large: %s", fname);
        return 1;
    }

    return 0;
}

int pa_play_memchunk(
        pa_sink *sink,
        const pa_sample_spec *ss,
        const pa_channel_map *map,
        const pa_memchunk *chunk,
        pa_cvolume *volume,
        pa_proplist *p,
        uint32_t *sink_input_index) {

    pa_memblockq *q;
    int r;

    pa_assert(sink);
    pa_assert(ss);
    pa_assert(chunk);

    q = pa_memblockq_new(0, chunk->length, 0, pa_frame_size(ss), 1, 1, 0, NULL);
    pa_assert_se(pa_memblockq_push(q, chunk) >= 0);

    if ((r = pa_play_memblockq(sink, ss, map, q, volume, p, sink_input_index)) < 0) {
        pa_memblockq_free(q);
        return r;
    }

    return 0;
}

/**
 * Start recording
 *
 * We request the default format used by pulse here because the data will be
 * converted and possibly re-sampled by obs anyway.
 *
 * For now we request a buffer length of 25ms although pulse seems to ignore
 * this setting for monitor streams. For "real" input streams this should work
 * fine though.
 */
static int_fast32_t pulse_start_recording(struct pulse_data *data)
{
	if (pulse_get_server_info(pulse_server_info, (void *) data) < 0) {
		blog(LOG_ERROR, "Unable to get server info !");
		return -1;
	}

	if (pulse_get_source_info(pulse_source_info, data->device,
			(void *) data) < 0) {
		blog(LOG_ERROR, "Unable to get source info !");
		return -1;
	}

	pa_sample_spec spec;
	spec.format   = data->format;
	spec.rate     = data->samples_per_sec;
	spec.channels = data->channels;

	if (!pa_sample_spec_valid(&spec)) {
		blog(LOG_ERROR, "Sample spec is not valid");
		return -1;
	}

	data->speakers = pulse_channels_to_obs_speakers(spec.channels);
	data->bytes_per_frame = pa_frame_size(&spec);

	data->stream = pulse_stream_new(obs_source_get_name(data->source),
		&spec, NULL);
	if (!data->stream) {
		blog(LOG_ERROR, "Unable to create stream");
		return -1;
	}

	pulse_lock();
	pa_stream_set_read_callback(data->stream, pulse_stream_read,
		(void *) data);
	pulse_unlock();

	pa_buffer_attr attr;
	attr.fragsize  = pa_usec_to_bytes(25000, &spec);
	attr.maxlength = (uint32_t) -1;
	attr.minreq    = (uint32_t) -1;
	attr.prebuf    = (uint32_t) -1;
	attr.tlength   = (uint32_t) -1;

	pa_stream_flags_t flags = PA_STREAM_ADJUST_LATENCY;

	pulse_lock();
	int_fast32_t ret = pa_stream_connect_record(data->stream, data->device,
		&attr, flags);
	pulse_unlock();
	if (ret < 0) {
		pulse_stop_recording(data);
		blog(LOG_ERROR, "Unable to connect to stream");
		return -1;
	}

	blog(LOG_INFO, "Started recording from '%s'", data->device);
	return 0;
}

FileReader::Status FileReader::writeToStream(pa_stream *p, size_t nbytes) {
  int frameSize = pa_frame_size(&m_s);
  int len = qMin(nbytes, (nbytes/frameSize) * nbytes);
  void *buff = pa_xmalloc(len);
  if (!buff) {
    qWarning() << "Failed to allocate buffer";
    return StatusError;
  }

  sf_count_t read = sf_read_raw(m_file, buff, len);

  if (pa_stream_write(p, buff, read, pa_xfree, 0, PA_SEEK_RELATIVE) < 0) {
    qWarning() << "Failed to write to pulse audio stream";
    pa_xfree(buff);
    return StatusError;
  }

  m_pos += read;
  if (m_pos == size()) {
    // Over.
    return StatusEof;
  }

  if (read != len) {
    return StatusError;
  }

  return StatusOk;
}

int pa_play_memchunk(
        pa_sink *sink,
        const pa_sample_spec *ss,
        const pa_channel_map *map,
        const pa_memchunk *chunk,
        pa_cvolume *volume,
        pa_proplist *p,
        uint32_t *sink_input_index) {

    pa_memblockq *q;
    int r;
    pa_memchunk silence;

    pa_assert(sink);
    pa_assert(ss);
    pa_assert(chunk);

    pa_silence_memchunk_get(&sink->core->silence_cache, sink->core->mempool, &silence, ss, 0);
    q = pa_memblockq_new(0, chunk->length, 0, pa_frame_size(ss), 1, 1, 0, &silence);
    pa_memblock_unref(silence.memblock);

    pa_assert_se(pa_memblockq_push(q, chunk) >= 0);

    if ((r = pa_play_memblockq(sink, ss, map, q, volume, p, sink_input_index)) < 0) {
        pa_memblockq_free(q);
        return r;
    }

    return 0;
}

/* Called from main context */
static void adjust_rates(struct userdata *u) {
    size_t buffer, fs;
    uint32_t old_rate, base_rate, new_rate;
    pa_usec_t buffer_latency;

    pa_assert(u);
    pa_assert_ctl_context();

    pa_asyncmsgq_send(u->source_output->source->asyncmsgq, PA_MSGOBJECT(u->source_output), SOURCE_OUTPUT_MESSAGE_LATENCY_SNAPSHOT, NULL, 0, NULL);
    pa_asyncmsgq_send(u->sink_input->sink->asyncmsgq, PA_MSGOBJECT(u->sink_input), SINK_INPUT_MESSAGE_LATENCY_SNAPSHOT, NULL, 0, NULL);

    buffer =
        u->latency_snapshot.sink_input_buffer +
        u->latency_snapshot.source_output_buffer;

    if (u->latency_snapshot.recv_counter <= u->latency_snapshot.send_counter)
        buffer += (size_t) (u->latency_snapshot.send_counter - u->latency_snapshot.recv_counter);
    else
        buffer += PA_CLIP_SUB(buffer, (size_t) (u->latency_snapshot.recv_counter - u->latency_snapshot.send_counter));

    buffer_latency = pa_bytes_to_usec(buffer, &u->sink_input->sample_spec);

    pa_log_debug("Loopback overall latency is %0.2f ms + %0.2f ms + %0.2f ms = %0.2f ms",
                (double) u->latency_snapshot.sink_latency / PA_USEC_PER_MSEC,
                (double) buffer_latency / PA_USEC_PER_MSEC,
                (double) u->latency_snapshot.source_latency / PA_USEC_PER_MSEC,
                ((double) u->latency_snapshot.sink_latency + buffer_latency + u->latency_snapshot.source_latency) / PA_USEC_PER_MSEC);

    pa_log_debug("Should buffer %zu bytes, buffered at minimum %zu bytes",
                u->latency_snapshot.max_request*2,
                u->latency_snapshot.min_memblockq_length);

    fs = pa_frame_size(&u->sink_input->sample_spec);
    old_rate = u->sink_input->sample_spec.rate;
    base_rate = u->source_output->sample_spec.rate;

    if (u->latency_snapshot.min_memblockq_length < u->latency_snapshot.max_request*2)
        new_rate = base_rate - (((u->latency_snapshot.max_request*2 - u->latency_snapshot.min_memblockq_length) / fs) *PA_USEC_PER_SEC)/u->adjust_time;
    else
        new_rate = base_rate + (((u->latency_snapshot.min_memblockq_length - u->latency_snapshot.max_request*2) / fs) *PA_USEC_PER_SEC)/u->adjust_time;

    if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) {
        pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate, new_rate);
        new_rate = base_rate;
    } else {
        if (base_rate < new_rate + 20 && new_rate < base_rate + 20)
          new_rate = base_rate;
        /* Do the adjustment in small steps; 2‰ can be considered inaudible */
        if (new_rate < (uint32_t) (old_rate*0.998) || new_rate > (uint32_t) (old_rate*1.002)) {
            pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate, old_rate);
            new_rate = PA_CLAMP(new_rate, (uint32_t) (old_rate*0.998), (uint32_t) (old_rate*1.002));
        }
    }

    pa_sink_input_set_rate(u->sink_input, new_rate);
    pa_log_debug("[%s] Updated sampling rate to %lu Hz.", u->sink_input->sink->name, (unsigned long) new_rate);

    pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
}

/* This is called whenever new data may be written to the stream */
static void stream_write_callback(pa_stream *s, size_t length, void *userdata) {
    pa_assert(s);
    pa_assert(length > 0);

    if (raw) {
        pa_assert(!sndfile);

        if (stdio_event)
            mainloop_api->io_enable(stdio_event, PA_IO_EVENT_INPUT);

        if (!buffer)
            return;

        do_stream_write(length);

    } else {
        sf_count_t bytes;
        void *data;

        pa_assert(sndfile);

        for (;;) {
            size_t data_length = length;

            if (pa_stream_begin_write(s, &data, &data_length) < 0) {
                pa_log(_("pa_stream_begin_write() failed: %s"), pa_strerror(pa_context_errno(context)));
                quit(1);
                return;
            }

            if (readf_function) {
                size_t k = pa_frame_size(&sample_spec);

                if ((bytes = readf_function(sndfile, data, (sf_count_t) (data_length/k))) > 0)
                    bytes *= (sf_count_t) k;

            } else
                bytes = sf_read_raw(sndfile, data, (sf_count_t) data_length);

            if (bytes > 0)
                pa_stream_write(s, data, (size_t) bytes, NULL, 0, PA_SEEK_RELATIVE);
            else
                pa_stream_cancel_write(s);

            /* EOF? */
            if (bytes < (sf_count_t) data_length) {
                start_drain();
                break;
            }

            /* Request fulfilled */
            if ((size_t) bytes >= length)
                break;

            length -= bytes;
        }
    }
}

bool pa_frame_aligned(size_t l, const pa_sample_spec *ss) {
    size_t fs;

    pa_assert(ss);

    fs = pa_frame_size(ss);

    return l % fs == 0;
}

size_t pa_frame_align(size_t l, const pa_sample_spec *ss) {
    size_t fs;

    pa_assert(ss);

    fs = pa_frame_size(ss);

    return (l/fs) * fs;
}

/*
 * start recording
 */
static int_fast32_t pulse_start_recording(struct pulse_data *data)
{
	if (pulse_get_server_info(pulse_server_info, (void *) data) < 0) {
		blog(LOG_ERROR, "pulse-input: Unable to get server info !");
		return -1;
	}

	pa_sample_spec spec;
	spec.format   = data->format;
	spec.rate     = data->samples_per_sec;
	spec.channels = data->channels;

	if (!pa_sample_spec_valid(&spec)) {
		blog(LOG_ERROR, "pulse-input: Sample spec is not valid");
		return -1;
	}

	data->bytes_per_frame = pa_frame_size(&spec);
	blog(LOG_DEBUG, "pulse-input: %u bytes per frame",
	     (unsigned int) data->bytes_per_frame);

	data->stream = pulse_stream_new(obs_source_getname(data->source),
		&spec, NULL);
	if (!data->stream) {
		blog(LOG_ERROR, "pulse-input: Unable to create stream");
		return -1;
	}

	pulse_lock();
	pa_stream_set_read_callback(data->stream, pulse_stream_read,
		(void *) data);
	pulse_unlock();

	pa_buffer_attr attr;
	attr.fragsize  = get_buffer_size(data, 250);
	attr.maxlength = (uint32_t) -1;
	attr.minreq    = (uint32_t) -1;
	attr.prebuf    = (uint32_t) -1;
	attr.tlength   = (uint32_t) -1;

	pa_stream_flags_t flags =
		PA_STREAM_INTERPOLATE_TIMING
		| PA_STREAM_AUTO_TIMING_UPDATE
		| PA_STREAM_ADJUST_LATENCY;

	pulse_lock();
	int_fast32_t ret = pa_stream_connect_record(data->stream, data->device,
		&attr, flags);
	pulse_unlock();
	if (ret < 0) {
		blog(LOG_ERROR, "pulse-input: Unable to connect to stream");
		return -1;
	}

	blog(LOG_DEBUG, "pulse-input: Recording started");
	return 0;
}

pa_bool_t pa_speex_ec_init(pa_core *c, pa_echo_canceller *ec,
                           pa_sample_spec *source_ss, pa_channel_map *source_map,
                           pa_sample_spec *sink_ss, pa_channel_map *sink_map,
                           uint32_t *blocksize, const char *args)
{
    int framelen, y, rate;
    uint32_t frame_size_ms, filter_size_ms;
    pa_modargs *ma;

    if (!(ma = pa_modargs_new(args, valid_modargs))) {
        pa_log("Failed to parse submodule arguments.");
        goto fail;
    }

    filter_size_ms = DEFAULT_FILTER_SIZE_MS;
    if (pa_modargs_get_value_u32(ma, "filter_size_ms", &filter_size_ms) < 0 || filter_size_ms < 1 || filter_size_ms > 2000) {
        pa_log("Invalid filter_size_ms specification");
        goto fail;
    }

    frame_size_ms = DEFAULT_FRAME_SIZE_MS;
    if (pa_modargs_get_value_u32(ma, "frame_size_ms", &frame_size_ms) < 0 || frame_size_ms < 1 || frame_size_ms > 200) {
        pa_log("Invalid frame_size_ms specification");
        goto fail;
    }

    pa_speex_ec_fixate_spec(source_ss, source_map, sink_ss, sink_map);

    rate = source_ss->rate;
    framelen = (rate * frame_size_ms) / 1000;
    /* framelen should be a power of 2, round down to nearest power of two */
    y = 1 << ((8 * sizeof (int)) - 2);
    while (y > framelen)
      y >>= 1;
    framelen = y;

    *blocksize = framelen * pa_frame_size (source_ss);

    pa_log_debug ("Using framelen %d, blocksize %u, channels %d, rate %d", framelen, *blocksize, source_ss->channels, source_ss->rate);

    ec->params.priv.speex.state = speex_echo_state_init_mc (framelen, (rate * filter_size_ms) / 1000, source_ss->channels, source_ss->channels);

    if (!ec->params.priv.speex.state)
	goto fail;

    speex_echo_ctl(ec->params.priv.speex.state, SPEEX_ECHO_SET_SAMPLING_RATE, &rate);

    pa_modargs_free(ma);
    return TRUE;

fail:
    if (ma)
	pa_modargs_free(ma);
    return FALSE;
}

/* Called from I/O thread context */
static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk) {
    struct userdata *u;
    float *src, *dst;
    size_t fs;
    unsigned n, h, c;
    pa_memchunk tchunk;

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

    /* Hmm, process any rewind request that might be queued up */
    pa_sink_process_rewind(u->sink, 0);

    while (pa_memblockq_peek(u->memblockq, &tchunk) < 0) {
        pa_memchunk nchunk;

        pa_sink_render(u->sink, nbytes, &nchunk);
        pa_memblockq_push(u->memblockq, &nchunk);
        pa_memblock_unref(nchunk.memblock);
    }

    tchunk.length = PA_MIN(nbytes, tchunk.length);
    pa_assert(tchunk.length > 0);

    fs = pa_frame_size(&i->sample_spec);
    n = (unsigned) (PA_MIN(tchunk.length, u->block_size) / fs);

    pa_assert(n > 0);

    chunk->index = 0;
    chunk->length = n*fs;
    chunk->memblock = pa_memblock_new(i->sink->core->mempool, chunk->length);

    pa_memblockq_drop(u->memblockq, chunk->length);

    src = (float*) ((uint8_t*) pa_memblock_acquire(tchunk.memblock) + tchunk.index);
    dst = (float*) pa_memblock_acquire(chunk->memblock);

    for (h = 0; h < (u->channels / u->max_ladspaport_count); h++) {
        for (c = 0; c < u->input_count; c++)
            pa_sample_clamp(PA_SAMPLE_FLOAT32NE, u->input[c], sizeof(float), src+ h*u->max_ladspaport_count + c, u->channels*sizeof(float), n);
        u->descriptor->run(u->handle[h], n);
        for (c = 0; c < u->output_count; c++)
            pa_sample_clamp(PA_SAMPLE_FLOAT32NE, dst + h*u->max_ladspaport_count + c, u->channels*sizeof(float), u->output[c], sizeof(float), n);
    }

    pa_memblock_release(tchunk.memblock);
    pa_memblock_release(chunk->memblock);

    pa_memblock_unref(tchunk.memblock);

    return 0;
}

static ALCenum pulse_capture_samples(ALCdevice *device, ALCvoid *buffer, ALCuint samples)
{
    pulse_data *data = device->ExtraData;
    ALCuint todo = samples * pa_frame_size(&data->spec);

    pa_threaded_mainloop_lock(data->loop);
    /* Capture is done in fragment-sized chunks, so we loop until we get all
     * that's available */
    data->last_readable -= todo;
    while(todo > 0)
    {
        size_t rem = todo;

        if(data->cap_len == 0)
        {
            pa_stream_state_t state;

            state = pa_stream_get_state(data->stream);
            if(!PA_STREAM_IS_GOOD(state))
            {
                aluHandleDisconnect(device);
                break;
            }
            if(pa_stream_peek(data->stream, &data->cap_store, &data->cap_len) < 0)
            {
                ERR("pa_stream_peek() failed: %s\n",
                    pa_strerror(pa_context_errno(data->context)));
                aluHandleDisconnect(device);
                break;
            }
            data->cap_remain = data->cap_len;
        }
        if(rem > data->cap_remain)
            rem = data->cap_remain;

        memcpy(buffer, data->cap_store, rem);

        buffer = (ALbyte*)buffer + rem;
        todo -= rem;

        data->cap_store = (ALbyte*)data->cap_store + rem;
        data->cap_remain -= rem;
        if(data->cap_remain == 0)
        {
            pa_stream_drop(data->stream);
            data->cap_len = 0;
        }
    }
    if(todo > 0)
        memset(buffer, ((device->FmtType==DevFmtUByte) ? 0x80 : 0), todo);
    pa_threaded_mainloop_unlock(data->loop);

    return ALC_NO_ERROR;
}

size_t pa_usec_to_bytes_round_up(pa_usec_t t, const pa_sample_spec *spec) {
    uint64_t u;
    pa_assert(spec);

    u = (uint64_t) t * (uint64_t) spec->rate;

    u = (u + PA_USEC_PER_SEC - 1) / PA_USEC_PER_SEC;

    u *= pa_frame_size(spec);

    return (size_t) u;
}

/*
 * Create a new pulse audio stream and connect to it
 *
 * Return a negative value on error
 */
static int pulse_connect_stream(struct pulse_data *data)
{
	pa_sample_spec spec;
	spec.format = data->format;
	spec.rate = data->samples_per_sec;
	spec.channels = get_audio_channels(data->speakers);

	if (!pa_sample_spec_valid(&spec)) {
		blog(LOG_ERROR, "pulse-input: Sample spec is not valid");
		return -1;
	}

	data->bytes_per_frame = pa_frame_size(&spec);
	blog(LOG_DEBUG, "pulse-input: %u bytes per frame",
	     (unsigned int) data->bytes_per_frame);

	pa_buffer_attr attr;
	attr.fragsize = get_buffer_size(data, 250);
	attr.maxlength = (uint32_t) -1;
	attr.minreq = (uint32_t) -1;
	attr.prebuf = (uint32_t) -1;
	attr.tlength = (uint32_t) -1;

	data->stream = pa_stream_new_with_proplist(data->context,
		obs_source_getname(data->source), &spec, NULL, data->props);
	if (!data->stream) {
		blog(LOG_ERROR, "pulse-input: Unable to create stream");
		return -1;
	}
	pa_stream_flags_t flags =
		PA_STREAM_INTERPOLATE_TIMING
		| PA_STREAM_AUTO_TIMING_UPDATE
		| PA_STREAM_ADJUST_LATENCY;
	if (pa_stream_connect_record(data->stream, NULL, &attr, flags) < 0) {
		blog(LOG_ERROR, "pulse-input: Unable to connect to stream");
		return -1;
	}

	for (;;) {
		pulse_iterate(data);
		pa_stream_state_t state = pa_stream_get_state(data->stream);
		if (state == PA_STREAM_READY) {
			blog(LOG_DEBUG, "pulse-input: Stream ready");
			break;
		}
		if (!PA_STREAM_IS_GOOD(state)) {
			blog(LOG_ERROR, "pulse-input: Stream connect failed");
			return -1;
		}
	}

	return 0;
}

pa_usec_t pa_bytes_to_usec_round_up(uint64_t length, const pa_sample_spec *spec) {
    size_t fs;
    pa_usec_t usec;

    pa_assert(spec);

    fs = pa_frame_size(spec);
    length = (length + fs - 1) / fs;

    usec = (pa_usec_t) length * PA_USEC_PER_SEC;

    return (usec + spec->rate - 1) / spec->rate;
}

size_t pa_context_get_tile_size(pa_context *c, const pa_sample_spec *ss) {
    size_t fs, mbs;

    pa_assert(c);
    pa_assert(PA_REFCNT_VALUE(c) >= 1);

    PA_CHECK_VALIDITY_RETURN_ANY(c, !pa_detect_fork(), PA_ERR_FORKED, (size_t) -1);
    PA_CHECK_VALIDITY_RETURN_ANY(c, !ss || pa_sample_spec_valid(ss), PA_ERR_INVALID, (size_t) -1);

    fs = ss ? pa_frame_size(ss) : 1;
    mbs = PA_ROUND_DOWN(pa_mempool_block_size_max(c->mempool), fs);
    return PA_MAX(mbs, fs);
}

APULSE_EXPORT
const pa_timing_info *
pa_stream_get_timing_info(pa_stream *s)
{
    trace_info_f("F %s s=%p\n", __func__, s);

    snd_pcm_sframes_t delay;

    if (snd_pcm_delay(s->ph, &delay) != 0)
        delay = 0;
    s->timing_info.read_index = s->timing_info.write_index - delay * pa_frame_size(&s->ss);

    return &s->timing_info;
}

static void
stream_request_callback(pa_stream * s, size_t nbytes, void * u)
{
  cubeb_stream * stm;
  void * buffer;
  size_t size;
  int r;
  long got;
  size_t towrite;
  size_t frame_size;

  stm = u;

  if (stm->shutdown)
    return;

  frame_size = pa_frame_size(&stm->sample_spec);

  assert(nbytes % frame_size == 0);

  towrite = nbytes;

  while (towrite) {
    size = towrite;
    r = pa_stream_begin_write(s, &buffer, &size);
    assert(r == 0);
    assert(size > 0);
    assert(size % frame_size == 0);

    got = stm->data_callback(stm, stm->user_ptr, buffer, size / frame_size);
    if (got < 0) {
      pa_stream_cancel_write(s);
      stm->shutdown = 1;
      return;
    }

    r = pa_stream_write(s, buffer, got * frame_size, NULL, 0, PA_SEEK_RELATIVE);
    assert(r == 0);

    if ((size_t) got < size / frame_size) {
      stm->draining = pa_stream_drain(s, stream_drain_success_callback, stm);
      stm->shutdown = 1;
      return;
    }

    towrite -= size;
  }

  assert(towrite == 0);
}