/**
 * 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;
}

static void process_rewind(struct userdata *u, pa_usec_t now) {
    size_t rewind_nbytes, in_buffer;
    pa_usec_t delay;

    pa_assert(u);

    rewind_nbytes = u->sink->thread_info.rewind_nbytes;

    if (!PA_SINK_IS_OPENED(u->sink->thread_info.state) || rewind_nbytes <= 0)
        goto do_nothing;

    pa_log_debug("Requested to rewind %lu bytes.", (unsigned long) rewind_nbytes);

    if (u->timestamp <= now)
        goto do_nothing;

    delay = u->timestamp - now;
    in_buffer = pa_usec_to_bytes(delay, &u->sink->sample_spec);

    if (in_buffer <= 0)
        goto do_nothing;

    if (rewind_nbytes > in_buffer)
        rewind_nbytes = in_buffer;

    pa_sink_process_rewind(u->sink, rewind_nbytes);
    u->timestamp -= pa_bytes_to_usec(rewind_nbytes, &u->sink->sample_spec);

    pa_log_debug("Rewound %lu bytes.", (unsigned long) rewind_nbytes);
    return;

do_nothing:

    pa_sink_process_rewind(u->sink, 0);
}

static void kradpulse_stream_underflow_cb(pa_stream *s, void *userdata) {

    krad_pulse_t *kradpulse = (krad_pulse_t *)userdata;

    // We increase the latency by 50% if we get 6 underflows and latency is under 2s
    // This is very useful for over the network playback that can't handle low latencies
    printf("underflow\n");
    kradpulse->underflows++;
    if (kradpulse->underflows >= 6 && kradpulse->latency < 2000000) {
        kradpulse->latency = (kradpulse->latency*3)/2;
        kradpulse->bufattr.maxlength = pa_usec_to_bytes(kradpulse->latency, &kradpulse->ss);
        kradpulse->bufattr.tlength = pa_usec_to_bytes(kradpulse->latency, &kradpulse->ss);
        pa_stream_set_buffer_attr(s, &kradpulse->bufattr, NULL, NULL);
        kradpulse->underflows = 0;
        printf("latency increased to %d\n", kradpulse->latency);
    }
}

AudioStream::AudioStream(pa_context *c, pa_threaded_mainloop *m, const char *desc, int type, int smplrate, std::string& deviceName)
    : audiostream_(0), mainloop_(m)
{
    static const pa_channel_map channel_map = {
        1,
        { PA_CHANNEL_POSITION_MONO },
    };

    pa_sample_spec sample_spec = {
        PA_SAMPLE_S16LE, // PA_SAMPLE_FLOAT32LE,
        smplrate,
        1
    };

    assert(pa_sample_spec_valid(&sample_spec));
    assert(pa_channel_map_valid(&channel_map));

    audiostream_ = pa_stream_new(c, desc, &sample_spec, &channel_map);

    if (!audiostream_) {
        ERROR("%s: pa_stream_new() failed : %s" , desc, pa_strerror(pa_context_errno(c)));
        throw std::runtime_error("Could not create stream\n");
    }

    pa_buffer_attr attributes;
    attributes.maxlength = pa_usec_to_bytes(160 * PA_USEC_PER_MSEC, &sample_spec);
    attributes.tlength = pa_usec_to_bytes(80 * PA_USEC_PER_MSEC, &sample_spec);
    attributes.prebuf = 0;
    attributes.fragsize = pa_usec_to_bytes(80 * PA_USEC_PER_MSEC, &sample_spec);
    attributes.minreq = (uint32_t) -1;

    pa_threaded_mainloop_lock(mainloop_);

    if (type == PLAYBACK_STREAM || type == RINGTONE_STREAM)
        pa_stream_connect_playback(audiostream_, deviceName == "" ? NULL : deviceName.c_str(), &attributes,
		(pa_stream_flags_t)(PA_STREAM_ADJUST_LATENCY|PA_STREAM_AUTO_TIMING_UPDATE), NULL, NULL);
    else if (type == CAPTURE_STREAM)
        pa_stream_connect_record(audiostream_, deviceName == "" ? NULL : deviceName.c_str(), &attributes,
		(pa_stream_flags_t)(PA_STREAM_ADJUST_LATENCY|PA_STREAM_AUTO_TIMING_UPDATE));

    pa_threaded_mainloop_unlock(mainloop_);

    pa_stream_set_state_callback(audiostream_, stream_state_callback, NULL);
}

static void thread_func(void *userdata) {
    struct userdata *u = userdata;

    pa_assert(u);

    pa_log_debug("Thread starting up");

    pa_thread_mq_install(&u->thread_mq);

    u->timestamp = pa_rtclock_now();

    for (;;) {
        int ret;

        /* Generate some null data */
        if (PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
            pa_usec_t now;
            pa_memchunk chunk;

            now = pa_rtclock_now();

            if ((chunk.length = pa_usec_to_bytes(now - u->timestamp, &u->source->sample_spec)) > 0) {

                chunk.memblock = pa_memblock_new(u->core->mempool, (size_t) -1); /* or chunk.length? */
                chunk.index = 0;
                pa_source_post(u->source, &chunk);
                pa_memblock_unref(chunk.memblock);

                u->timestamp = now;
            }

            pa_rtpoll_set_timer_absolute(u->rtpoll, u->timestamp + u->latency_time * PA_USEC_PER_MSEC);
        } else
            pa_rtpoll_set_timer_disabled(u->rtpoll);

        /* Hmm, nothing to do. Let's sleep */
        if ((ret = pa_rtpoll_run(u->rtpoll, TRUE)) < 0)
            goto fail;

        if (ret == 0)
            goto finish;
    }

fail:
    /* If this was no regular exit from the loop we have to continue
     * processing messages until we received PA_MESSAGE_SHUTDOWN */
    pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
    pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);

finish:
    pa_log_debug("Thread shutting down");
}

boolean
adin_standby(int sfreq, void *dummy)
{	
	static const pa_sample_spec ss = {
		.format = PA_SAMPLE_S16LE,
		.rate = SAMPLE_RATE,
		.channels = NBCHANNEL,
	};
	
	// low-latency setting
	pa_buffer_attr ba;
	ba.tlength = pa_usec_to_bytes(50*1000, &ss);
	ba.minreq = pa_usec_to_bytes(0, &ss);
	ba.maxlength = pa_usec_to_bytes(50*1000, &ss);
	ba.fragsize = sizeof(uint32_t) -1;
	
	// input
	if (!(r = pa_simple_new(NULL, "kiku", PA_STREAM_RECORD, NULL, "voice recognition", &ss, NULL, &ba, &errorpa))) {
		return FALSE;
	}
	
  return TRUE;
}

void audio_init() {
	pa_threaded_mainloop *pa_ml=pa_threaded_mainloop_new();
	pa_mainloop_api *pa_mlapi=pa_threaded_mainloop_get_api(pa_ml);
	pa_context *pa_ctx=pa_context_new(pa_mlapi, "te");
	pa_context_connect(pa_ctx, NULL, 0, NULL);
	int pa_ready = 0;
	pa_context_set_state_callback(pa_ctx, pa_state_cb, &pa_ready);

	pa_threaded_mainloop_start(pa_ml);
	while(pa_ready==0) { ; }

	printf("audio ready\n");

	if (pa_ready == 2) {
		pa_context_disconnect(pa_ctx);
		pa_context_unref(pa_ctx);
		pa_threaded_mainloop_free(pa_ml);
	}

	pa_sample_spec ss;
	ss.rate=96000;
	ss.channels=1;
	ss.format=PA_SAMPLE_S24_32LE;
	ps=pa_stream_new(pa_ctx,"Playback",&ss,NULL);
	pa_stream_set_write_callback(ps,audio_request_cb,NULL);
	pa_stream_set_underflow_callback(ps,audio_underflow_cb,NULL);

	pa_buffer_attr bufattr;
	bufattr.fragsize = (uint32_t)-1;
	bufattr.maxlength = pa_usec_to_bytes(20000,&ss);
	bufattr.minreq = pa_usec_to_bytes(0,&ss);
	bufattr.prebuf = 0;
	bufattr.tlength = pa_usec_to_bytes(20000,&ss);

	pa_stream_connect_playback(ps,NULL,&bufattr,
		PA_STREAM_INTERPOLATE_TIMING|PA_STREAM_ADJUST_LATENCY|PA_STREAM_AUTO_TIMING_UPDATE|PA_STREAM_START_CORKED,NULL,NULL);
}

static void sink_update_requested_latency_cb(pa_sink *s) {
    struct userdata *u;
    size_t nbytes;

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

    u->block_usec = pa_sink_get_requested_latency_within_thread(s);

    if (u->block_usec == (pa_usec_t) -1)
        u->block_usec = s->thread_info.max_latency;

    nbytes = pa_usec_to_bytes(u->block_usec, &s->sample_spec);
    pa_sink_set_max_rewind_within_thread(s, nbytes);
    pa_sink_set_max_request_within_thread(s, nbytes);
}

/* Called from output thread context */
static void source_output_state_change_cb(pa_source_output *o, pa_source_output_state_t state) {
    struct userdata *u;

    pa_source_output_assert_ref(o);
    pa_source_output_assert_io_context(o);
    pa_assert_se(u = o->userdata);

    if (PA_SOURCE_OUTPUT_IS_LINKED(state) && o->thread_info.state == PA_SOURCE_OUTPUT_INIT) {

        u->skip = pa_usec_to_bytes(PA_CLIP_SUB(pa_source_get_latency_within_thread(o->source),
                                               u->latency),
                                   &o->sample_spec);

        pa_log_info("Skipping %lu bytes", (unsigned long) u->skip);
    }
}

int
sa_stream_get_position(sa_stream_t *s, sa_position_t position, int64_t *pos) {
  pa_usec_t usec;
  if (s == NULL || s->stream == NULL) {
    return SA_ERROR_NO_INIT;
  }
  if (position != SA_POSITION_WRITE_SOFTWARE) {
    return SA_ERROR_NOT_SUPPORTED;
  }
  pa_threaded_mainloop_lock(s->m);
  if(pa_stream_get_time(s->stream,  &usec) != PA_ERR_NODATA) {
    *pos = pa_usec_to_bytes(usec, &s->sample_spec);
  }
  else {
    *pos  = s->bytes_written;
  }
  pa_threaded_mainloop_unlock(s->m);
  return SA_SUCCESS;
}

int
cubeb_stream_get_position(cubeb_stream * stm, uint64_t * position)
{
  int r;
  pa_usec_t r_usec;
  uint64_t bytes;

  pa_threaded_mainloop_lock(stm->context->mainloop);
  r = pa_stream_get_time(stm->stream, &r_usec);
  pa_threaded_mainloop_unlock(stm->context->mainloop);

  if (r != 0) {
    return CUBEB_ERROR;
  }

  /* XXX might be more accurate to compute directly from get_timing_info */
  bytes = pa_usec_to_bytes(r_usec, &stm->sample_spec);
  *position = bytes / pa_frame_size(&stm->sample_spec);

  return CUBEB_OK;
}

static void process_rewind(struct context *context, pa_usec_t now)
{
	size_t rewind_nbytes, in_buffer;
	pa_usec_t delay;

	pa_assert(context);

	/* Figure out how much we shall rewind and reset the counter */
	rewind_nbytes = context->sink->thread_info.rewind_nbytes;
	context->sink->thread_info.rewind_nbytes = 0;

	if (rewind_nbytes <= 0)
		goto DO_NOTHING;

	pa_log_debug("Requested to rewind %lu bytes.",
		(unsigned long) rewind_nbytes);

	if (context->timestamp <= now)
		goto DO_NOTHING;

	delay = context->timestamp - now;
	in_buffer = pa_usec_to_bytes(delay, &context->sink->sample_spec);

	if (in_buffer <= 0)
		goto DO_NOTHING;

	if (rewind_nbytes > in_buffer)
		rewind_nbytes = in_buffer;

	pa_sink_process_rewind(context->sink, rewind_nbytes);
	context->timestamp -= pa_bytes_to_usec(rewind_nbytes, &context->sink->sample_spec);
	context->skip_bytes += rewind_nbytes;

	pa_log_debug("Rewound %lu bytes.", (unsigned long) rewind_nbytes);

	return;

DO_NOTHING:
	pa_sink_process_rewind(context->sink, 0);
}

static void sink_update_requested_latency_cb(pa_sink *s) {
    struct userdata *u;
    size_t nbytes;

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

    u->block_usec = BLOCK_USEC;
    //u->block_usec = pa_sink_get_requested_latency_within_thread(s);
    pa_log("1 block_usec %d", u->block_usec);

    u->got_max_latency = 0;
    if (u->block_usec == (pa_usec_t) -1) {
        u->block_usec = s->thread_info.max_latency;
        pa_log_debug("2 block_usec %d", u->block_usec);
        u->got_max_latency = 1;
    }

    nbytes = pa_usec_to_bytes(u->block_usec, &s->sample_spec);
    pa_sink_set_max_rewind_within_thread(s, nbytes);
    pa_sink_set_max_request_within_thread(s, nbytes);
}

static void sink_update_requested_latency_cb(pa_sink *s) {
    struct userdata *u;
    pa_operation *operation;
    size_t nbytes;
    pa_usec_t block_usec;
    pa_buffer_attr bufferattr;

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

    block_usec = pa_sink_get_requested_latency_within_thread(s);
    if (block_usec == (pa_usec_t) -1)
        block_usec = s->thread_info.max_latency;

    nbytes = pa_usec_to_bytes(block_usec, &s->sample_spec);
    pa_sink_set_max_request_within_thread(s, nbytes);

    if (u->stream) {
        switch (pa_stream_get_state(u->stream)) {
            case PA_STREAM_READY:
                if (pa_stream_get_buffer_attr(u->stream)->tlength == nbytes)
                    break;

                reset_bufferattr(&bufferattr);
                bufferattr.tlength = nbytes;
                if ((operation = pa_stream_set_buffer_attr(u->stream, &bufferattr, stream_set_buffer_attr_cb, u)))
                    pa_operation_unref(operation);
                break;
            case PA_STREAM_CREATING:
                /* we have to delay our request until stream is ready */
                u->update_stream_bufferattr_after_connect = true;
                break;
            default:
                break;
        }
    }
}

static void process_rewind(struct userdata *u, pa_usec_t now) {
    size_t rewind_nbytes, in_buffer;
    pa_usec_t delay;

    pa_assert(u);

    /* Figure out how much we shall rewind and reset the counter */
    rewind_nbytes = u->sink->thread_info.rewind_nbytes;
    u->sink->thread_info.rewind_nbytes = 0;

    pa_assert(rewind_nbytes > 0);
    pa_log_debug("Requested to rewind %lu bytes.",
                 (unsigned long) rewind_nbytes);

    if (u->timestamp <= now)
        goto do_nothing;

    delay = u->timestamp - now;
    in_buffer = pa_usec_to_bytes(delay, &u->sink->sample_spec);

    if (in_buffer <= 0)
        goto do_nothing;

    if (rewind_nbytes > in_buffer)
        rewind_nbytes = in_buffer;

    pa_sink_process_rewind(u->sink, rewind_nbytes);
    u->timestamp -= pa_bytes_to_usec(rewind_nbytes, &u->sink->sample_spec);
    u->skip_bytes += rewind_nbytes;

    pa_log_debug("Rewound %lu bytes.", (unsigned long) rewind_nbytes);
    return;

do_nothing:

    pa_sink_process_rewind(u->sink, 0);
}

static void *pulse_init(const char *device, unsigned rate, unsigned latency)
{
   pa_sample_spec spec;
   pa_t *pa;
   pa_buffer_attr buffer_attr = {0};
   const pa_buffer_attr *server_attr = NULL;

   memset(&spec, 0, sizeof(spec));
   
   pa = (pa_t*)calloc(1, sizeof(*pa));
   if (!pa)
      goto error;

   pa->mainloop = pa_threaded_mainloop_new();
   if (!pa->mainloop)
      goto error;

   pa->context = pa_context_new(pa_threaded_mainloop_get_api(pa->mainloop), "RetroArch");
   if (!pa->context)
      goto error;

   pa_context_set_state_callback(pa->context, context_state_cb, pa);

   if (pa_context_connect(pa->context, device, PA_CONTEXT_NOFLAGS, NULL) < 0)
      goto error;

   pa_threaded_mainloop_lock(pa->mainloop);
   if (pa_threaded_mainloop_start(pa->mainloop) < 0)
      goto error;

   pa_threaded_mainloop_wait(pa->mainloop);

   if (pa_context_get_state(pa->context) != PA_CONTEXT_READY)
      goto unlock_error;

   spec.format = is_little_endian() ? PA_SAMPLE_FLOAT32LE : PA_SAMPLE_FLOAT32BE;
   spec.channels = 2;
   spec.rate = rate;

   pa->stream = pa_stream_new(pa->context, "audio", &spec, NULL);
   if (!pa->stream)
      goto unlock_error;

   pa_stream_set_state_callback(pa->stream, stream_state_cb, pa);
   pa_stream_set_write_callback(pa->stream, stream_request_cb, pa);
   pa_stream_set_latency_update_callback(pa->stream, stream_latency_update_cb, pa);
   pa_stream_set_underflow_callback(pa->stream, underrun_update_cb, pa);
   pa_stream_set_buffer_attr_callback(pa->stream, buffer_attr_cb, pa);

   buffer_attr.maxlength = -1;
   buffer_attr.tlength = pa_usec_to_bytes(latency * PA_USEC_PER_MSEC, &spec);
   buffer_attr.prebuf = -1;
   buffer_attr.minreq = -1;
   buffer_attr.fragsize = -1;

   if (pa_stream_connect_playback(pa->stream, NULL, &buffer_attr, PA_STREAM_ADJUST_LATENCY, NULL, NULL) < 0)
      goto error;

   pa_threaded_mainloop_wait(pa->mainloop);

   if (pa_stream_get_state(pa->stream) != PA_STREAM_READY)
      goto unlock_error;

   server_attr = pa_stream_get_buffer_attr(pa->stream);
   if (server_attr)
   {
      pa->buffer_size = server_attr->tlength;
      RARCH_LOG("[PulseAudio]: Requested %u bytes buffer, got %u.\n",
            (unsigned)buffer_attr.tlength,
            (unsigned)pa->buffer_size);
   }
   else
      pa->buffer_size = buffer_attr.tlength;

   pa_threaded_mainloop_unlock(pa->mainloop);

   return pa;

unlock_error:
   pa_threaded_mainloop_unlock(pa->mainloop);
error:
   pulse_free(pa); 
   return NULL;
}

int pa__init(pa_module*m) {

    struct audio_buf_info info;
    struct userdata *u = NULL;
    const char *dev;
    int fd = -1;
    int nfrags, orig_frag_size, frag_size;
    int mode, caps;
    pa_bool_t record = TRUE, playback = TRUE, use_mmap = TRUE;
    pa_sample_spec ss;
    pa_channel_map map;
    pa_modargs *ma = NULL;
    char hwdesc[64];
    const char *name;
    pa_bool_t namereg_fail;
    pa_sink_new_data sink_new_data;
    pa_source_new_data source_new_data;

    pa_assert(m);

    if (!(ma = pa_modargs_new(m->argument, valid_modargs))) {
        pa_log("Failed to parse module arguments.");
        goto fail;
    }

    if (pa_modargs_get_value_boolean(ma, "record", &record) < 0 || pa_modargs_get_value_boolean(ma, "playback", &playback) < 0) {
        pa_log("record= and playback= expect boolean argument.");
        goto fail;
    }

    if (!playback && !record) {
        pa_log("Neither playback nor record enabled for device.");
        goto fail;
    }

    mode = (playback && record) ? O_RDWR : (playback ? O_WRONLY : (record ? O_RDONLY : 0));

    ss = m->core->default_sample_spec;
    map = m->core->default_channel_map;
    if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_OSS) < 0) {
        pa_log("Failed to parse sample specification or channel map");
        goto fail;
    }

    nfrags = (int) m->core->default_n_fragments;
    frag_size = (int) pa_usec_to_bytes(m->core->default_fragment_size_msec*1000, &ss);
    if (frag_size <= 0)
        frag_size = (int) pa_frame_size(&ss);

    if (pa_modargs_get_value_s32(ma, "fragments", &nfrags) < 0 || pa_modargs_get_value_s32(ma, "fragment_size", &frag_size) < 0) {
        pa_log("Failed to parse fragments arguments");
        goto fail;
    }

    if (pa_modargs_get_value_boolean(ma, "mmap", &use_mmap) < 0) {
        pa_log("Failed to parse mmap argument.");
        goto fail;
    }

    if ((fd = pa_oss_open(dev = pa_modargs_get_value(ma, "device", DEFAULT_DEVICE), &mode, &caps)) < 0)
        goto fail;

    if (use_mmap && (!(caps & DSP_CAP_MMAP) || !(caps & DSP_CAP_TRIGGER))) {
        pa_log_info("OSS device not mmap capable, falling back to UNIX read/write mode.");
        use_mmap = FALSE;
    }

    if (use_mmap && mode == O_WRONLY) {
        pa_log_info("Device opened for playback only, cannot do memory mapping, falling back to UNIX write() mode.");
        use_mmap = FALSE;
    }

    if (pa_oss_get_hw_description(dev, hwdesc, sizeof(hwdesc)) >= 0)
        pa_log_info("Hardware name is '%s'.", hwdesc);
    else
        hwdesc[0] = 0;

    pa_log_info("Device opened in %s mode.", mode == O_WRONLY ? "O_WRONLY" : (mode == O_RDONLY ? "O_RDONLY" : "O_RDWR"));

    orig_frag_size = frag_size;
    if (nfrags >= 2 && frag_size >= 1)
        if (pa_oss_set_fragments(fd, nfrags, frag_size) < 0)
            goto fail;

    if (pa_oss_auto_format(fd, &ss) < 0)
        goto fail;

    if (ioctl(fd, SNDCTL_DSP_GETBLKSIZE, &frag_size) < 0) {
        pa_log("SNDCTL_DSP_GETBLKSIZE: %s", pa_cstrerror(errno));
        goto fail;
    }
    pa_assert(frag_size > 0);

    u = pa_xnew0(struct userdata, 1);
    u->core = m->core;
    u->module = m;
    m->userdata = u;
    u->fd = fd;
    u->mixer_fd = -1;
    u->mixer_devmask = 0;
//.........这里部分代码省略.........

static void thread_loop(void *arg)
{
    struct example_sink_userdata *u = arg;
    pa_assert(u);

    pa_thread_mq_install(&u->thread_mq);

    const pa_usec_t poll_interval = 10000;

    pa_usec_t start_time = 0;
    pa_usec_t next_time = 0;

    for (;;) {
        /* process rewind */
        if (u->sink->thread_info.rewind_requested) {
            process_rewind(u);
        }

        /* process sink inputs */
        if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
            pa_usec_t now_time = pa_rtclock_now();

            if (start_time == 0) {
                start_time = now_time;
                next_time = start_time + poll_interval;
            }
            else {
                while (now_time >= next_time) {
                    uint64_t expected_bytes =
                        pa_usec_to_bytes(next_time - start_time, &u->sink->sample_spec);

                    /* render samples from sink inputs and write them to output file */
                    process_samples(u, expected_bytes);

                    /* next tick */
                    next_time += poll_interval;
                }
            }

            /* schedule set next rendering tick */
            pa_rtpoll_set_timer_absolute(u->rtpoll, next_time);
        }
        else {
            /* sleep until state change */
            start_time = 0;
            next_time = 0;
            pa_rtpoll_set_timer_disabled(u->rtpoll);
        }

        /* process events and wait next rendering tick */
#if PA_CHECK_VERSION(5, 99, 0)
        int ret = pa_rtpoll_run(u->rtpoll);
#else
        int ret = pa_rtpoll_run(u->rtpoll, true);
#endif
        if (ret < 0) {
            pa_log("[example sink] pa_rtpoll_run returned error");
            goto error;
        }

        if (ret == 0) {
            break;
        }
    }

    return;

error:
    process_error(u);
}

static int
tsmf_pulse_open_stream(TSMFPulseAudioDevice * pulse)
{
	pa_stream_state_t state;
	pa_buffer_attr buffer_attr = { 0 };

	if (!pulse->context)
		return 1;

	LLOGLN(0, ("tsmf_pulse_open_stream:"));
	pa_threaded_mainloop_lock(pulse->mainloop);
	pulse->stream = pa_stream_new(pulse->context, "freerdp",
		&pulse->sample_spec, NULL);
	if (!pulse->stream)
	{
		pa_threaded_mainloop_unlock(pulse->mainloop);
		LLOGLN(0, ("tsmf_pulse_open_stream: pa_stream_new failed (%d)",
			pa_context_errno(pulse->context)));
		return 1;
	}
	pa_stream_set_state_callback(pulse->stream,
		tsmf_pulse_stream_state_callback, pulse);
	pa_stream_set_write_callback(pulse->stream,
		tsmf_pulse_stream_request_callback, pulse);
	buffer_attr.maxlength = pa_usec_to_bytes(500000, &pulse->sample_spec);
	buffer_attr.tlength = pa_usec_to_bytes(250000, &pulse->sample_spec);
	buffer_attr.prebuf = (uint32_t) -1;
	buffer_attr.minreq = (uint32_t) -1;
	buffer_attr.fragsize = (uint32_t) -1;
	if (pa_stream_connect_playback(pulse->stream,
		pulse->device[0] ? pulse->device : NULL, &buffer_attr,
		PA_STREAM_ADJUST_LATENCY | PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_AUTO_TIMING_UPDATE,
		NULL, NULL) < 0)
	{
		pa_threaded_mainloop_unlock(pulse->mainloop);
		LLOGLN(0, ("tsmf_pulse_open_stream: pa_stream_connect_playback failed (%d)",
			pa_context_errno(pulse->context)));
		return 1;
	}

	for (;;)
	{
		state = pa_stream_get_state(pulse->stream);
		if (state == PA_STREAM_READY)
			break;
        if (!PA_STREAM_IS_GOOD(state))
		{
			LLOGLN(0, ("tsmf_pulse_open_stream: bad stream state (%d)",
				pa_context_errno(pulse->context)));
			break;
		}
		pa_threaded_mainloop_wait(pulse->mainloop);
	}
	pa_threaded_mainloop_unlock(pulse->mainloop);
	if (state == PA_STREAM_READY)
	{
		LLOGLN(0, ("tsmf_pulse_open_stream: connected"));
		return 0;
	}
	else
	{
		tsmf_pulse_close_stream(pulse);
		return 1;
	}
}

int pa__init(pa_module*m) {
    struct userdata *u = NULL;
    pa_sample_spec ss;
    pa_channel_map map;
    pa_modargs *ma = NULL;
    pa_sink_new_data data;
    size_t nbytes;

    pa_assert(m);

    pa_web_debug_func_t func = pa_get_user_data(PA_USER_WEB_FUNC);
    if (!func) {
        pa_log_error("Failed to get user web func from user data");
        goto fail;
    }

    ma = pa_modargs_new(m->argument, valid_modargs);
    if (!ma) {
        pa_log("Failed to parse module arguments.");
        goto fail;
    }

    ss = m->core->default_sample_spec;
    map = m->core->default_channel_map;
    if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
        pa_log("Invalid sample format specification or channel map");
        goto fail;
    }

    pa_modargs_get_value_u32(ma, "fixed_rate", &ss.rate);
    uint32_t c = 0;
    pa_modargs_get_value_u32(ma, "fixed_channels", &c);
    if (c)
        ss.channels = c;
    const char* fixed_fmt = pa_modargs_get_value(ma, "fixed_format", NULL);
    if (fixed_fmt)
        ss.format = pa_parse_sample_format(fixed_fmt);

    if (ss.channels != map.channels) {
        pa_log_error("sample_spec channels[%d] != channel_map channels[%d]", ss.channels, map.channels);
        pa_channel_map_init_extend(&map, ss.channels, PA_CHANNEL_MAP_DEFAULT);
        pa_log_debug("after init extend mmap.channels=%d", map.channels);
    }

    m->userdata = u = pa_xnew0(struct userdata, 1);
    u->core = m->core;
    u->module = m;
    u->rtpoll = pa_rtpoll_new();
    u->data_func = web_data_func;

    pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll);

    pa_sink_new_data_init(&data);
    data.driver = __FILE__;
    data.module = m;
    pa_sink_new_data_set_name(&data, pa_modargs_get_value(ma, "sink_name", DEFAULT_SINK_NAME));
    pa_sink_new_data_set_sample_spec(&data, &ss);
    pa_sink_new_data_set_channel_map(&data, &map);
    pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, _("Remote Output"));
    pa_proplist_sets(data.proplist, PA_PROP_DEVICE_CLASS, "abstract");

    if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
        pa_log("Invalid properties");
        pa_sink_new_data_done(&data);
        goto fail;
    }

    u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_DYNAMIC_LATENCY);
    pa_sink_new_data_done(&data);

    if (!u->sink) {
        pa_log("Failed to create sink object.");
        goto fail;
    }

    u->sink->parent.process_msg = sink_process_msg;
    u->sink->update_requested_latency = sink_update_requested_latency_cb;
    u->sink->userdata = u;

    pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
    pa_sink_set_rtpoll(u->sink, u->rtpoll);

    u->block_usec = BLOCK_USEC;
    nbytes = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
    pa_sink_set_max_rewind(u->sink, nbytes);
    pa_sink_set_max_request(u->sink, nbytes);

    // call websocket to start
    func(1, NULL, ss.rate, ss.format);

    u->thread = pa_thread_new("remote-sink", thread_func, u);
    if (!u->thread) {
        pa_log("Failed to create thread.");
        goto fail;
    }

    pa_sink_set_latency_range(u->sink, 0, BLOCK_USEC);

    pa_sink_put(u->sink);

//.........这里部分代码省略.........