JNIEXPORT jint JNICALL
Java_org_jitsi_impl_neomedia_pulseaudio_PA_stream_1write
    (JNIEnv *env, jclass clazz, jlong s, jbyteArray data, jint dataOffset,
        jint dataLength, jobject freeCb, jlong offset, jint seek)
{
    pa_stream *stream = (pa_stream *) (intptr_t) s;
    jbyte *bytes = NULL;
    size_t nbytes = dataLength;
    int ret;

    pa_stream_begin_write(stream, (void **) &bytes, &nbytes);
    if (bytes && nbytes)
    {
        if (nbytes < dataLength)
            dataLength = nbytes;
        (*env)->GetByteArrayRegion(env, data, dataOffset, dataLength, bytes);
        if ((*env)->ExceptionCheck(env))
            ret = 0;
        else
        {
            pa_stream_write(
                    stream,
                    bytes,
                    (size_t) dataLength,
                    NULL,
                    (int64_t) offset,
                    (pa_seek_mode_t) seek);
            ret = dataLength;
        }
    }
    else
    {
        bytes = (*env)->GetByteArrayElements(env, data, NULL);
        if ((*env)->ExceptionCheck(env))
            ret = 0;
        else
        {
            pa_stream_write(
                    stream,
                    bytes + dataOffset,
                    (size_t) dataLength,
                    NULL,
                    (int64_t) offset,
                    (pa_seek_mode_t) seek);
            (*env)->ReleaseByteArrayElements(env, data, bytes, JNI_ABORT);
            ret = dataLength;
        }
    }
    return ret;
}

unsigned int CAESinkPULSE::AddPackets(uint8_t **data, unsigned int frames, unsigned int offset)
{
  if (!m_IsAllocated)
    return 0;

  if (m_IsStreamPaused)
  {
    Pause(false);
  }

  pa_threaded_mainloop_lock(m_MainLoop);

  unsigned int available = frames * m_format.m_frameSize;
  unsigned int length = 0;
  void *buffer = data[0]+offset*m_format.m_frameSize;
  // care a bit for fragmentation
  while ((length = pa_stream_writable_size(m_Stream)) < m_periodSize)
    pa_threaded_mainloop_wait(m_MainLoop);

  length =  std::min((unsigned int)length, available);

  int error = pa_stream_write(m_Stream, buffer, length, NULL, 0, PA_SEEK_RELATIVE);
  pa_threaded_mainloop_unlock(m_MainLoop);

  if (error)
  {
    CLog::Log(LOGERROR, "CPulseAudioDirectSound::AddPackets - pa_stream_write failed\n");
    return 0;
  }
  unsigned int res = (unsigned int)(length / m_format.m_frameSize);

  return res;
}

static ssize_t pulse_write(void *data, const void *buf_, size_t size)
{
   pa_t *pa = (pa_t*)data;
   const uint8_t *buf = (const uint8_t*)buf_;
   size_t written = 0;

   pa_threaded_mainloop_lock(pa->mainloop);
   while (size)
   {
      size_t writable = MIN(size, pa_stream_writable_size(pa->stream));

      if (writable)
      {
         pa_stream_write(pa->stream, buf, writable, NULL, 0, PA_SEEK_RELATIVE);
         buf += writable;
         size -= writable;
         written += writable;
      }
      else if (!pa->nonblock)
         pa_threaded_mainloop_wait(pa->mainloop);
      else
         break;
   }

   pa_threaded_mainloop_unlock(pa->mainloop);

   return written;
}

static void pulse_write(void* ptr, int length) {
    int writeoffs, remain, writable;

    CHECK_CONNECTED();

    pa_threaded_mainloop_lock(mainloop);
    CHECK_DEAD_GOTO(fail, 1);

    /* break large fragments into smaller fragments. --nenolod */
    for (writeoffs = 0, remain = length;
         writeoffs < length;
         writeoffs += writable, remain -= writable)
    {
         void * pptr = (char *) ptr + writeoffs;

         writable = length - writeoffs;
         size_t fragsize = pa_stream_writable_size(stream);

         /* don't write more than what PA is willing to handle right now. */
         if (writable > fragsize)
             writable = fragsize;

         if (pa_stream_write(stream, pptr, writable, NULL, PA_SEEK_RELATIVE, 0) < 0) {
             AUDDBG("pa_stream_write() failed: %s", pa_strerror(pa_context_errno(context)));
             goto fail;
         }
    }

    do_trigger = 0;
    written += length;

fail:
    pa_threaded_mainloop_unlock(mainloop);
}

static void stream_request_cb(pa_stream *s, size_t length, void *data) {
    M6502 *mpu = (M6502 *) data;
	int max = 0;

    redraw = 0;

    pa_stream_begin_write(s, (void**) &sndbuf, &length);

   	memset(sndbuf, 128, length);

    max = length * 6250 / 539;
    if (max > 20454)
		fprintf(stderr, "Cycles = %u!\n", (unsigned int) max);

    cycles = 0;
    while (cycles < max) {
        if (trace)
            M6502_disassemble(mpu);

        cycles += M6502_run(mpu);
    }

   	pa_stream_write(s, sndbuf, length, NULL, 0, PA_SEEK_RELATIVE);

    redraw = 1;
}

void PulseAudioDriver::stream_write_callback(pa_stream* stream, size_t bytes, void* udata)
{
	PulseAudioDriver* self = (PulseAudioDriver*)udata;

	void* vdata;
	pa_stream_begin_write(stream, &vdata, &bytes);
	if (!vdata) return;

	short* out = (short*)vdata;

	unsigned num_samples = bytes / 4;

	while (num_samples)
	{
		int n = std::min(self->m_buffer_size, num_samples);
		self->m_callback(n, 0);
		for (int i = 0; i < n; ++i)
		{
			*out++ = FLOAT_TO_SHORT(self->m_outL[i]);
			*out++ = FLOAT_TO_SHORT(self->m_outR[i]);
		}

		num_samples -= n;
	}

	pa_stream_write(stream, vdata, (bytes / 4) * 4, 0, 0, PA_SEEK_RELATIVE);
}

static void audio_request_cb(pa_stream *s, size_t length, void *userdata) {
	int i;
	uint32_t buf[length/4];

	if(pa_stream_is_corked(s)) return;

	for(i=0;i<length/4;i++) {
		if((offset+i-1)/ticksize != (offset+i)/ticksize) {
			int ntickno=(offset+i)/ticksize;
			if(ntickno!=tickno) {
				g_mutex_lock(tickmutex);
				tickno=ntickno;
				printf("signal %u %u (%u)\n",beatno(),tickinbeat(),offset+i);
				g_cond_signal(tickcond);
				g_mutex_unlock(tickmutex);
			}
		}


		int k;
		int32_t v=0;
		for(k=0;k<stack.len;k++) {
			v=stack.action[k].f(v,stack.action+k,offset+i);
		}
		buf[i]=v;
	}
	offset+=i;
	pa_stream_write(s,buf,length,0,0,PA_SEEK_RELATIVE);
}

static int outstream_end_write_pa(SoundIoPrivate *si, SoundIoOutStreamPrivate *os) {
    SoundIoOutStreamPulseAudio *ospa = &os->backend_data.pulseaudio;
    pa_stream *stream = ospa->stream;
    if (pa_stream_write(stream, ospa->write_ptr, ospa->write_byte_count, nullptr, 0, PA_SEEK_RELATIVE))
        return SoundIoErrorStreaming;
    return 0;
}

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);
}

/* Write some data to the stream */
static void do_stream_write(size_t length) {
    size_t l;
    pa_assert(length);

    if (!buffer || !buffer_length)
        return;

    l = length;
    if (l > buffer_length)
        l = buffer_length;

    if (pa_stream_write(stream, (uint8_t*) buffer + buffer_index, l, NULL, 0, PA_SEEK_RELATIVE) < 0) {
        pa_log(_("pa_stream_write() failed: %s"), pa_strerror(pa_context_errno(context)));
        quit(1);
        return;
    }

    buffer_length -= l;
    buffer_index += l;

    if (!buffer_length) {
        pa_xfree(buffer);
        buffer = NULL;
        buffer_index = buffer_length = 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;
}

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

static int
rdpsnd_pulse_play(rdpsndDevicePlugin * devplugin, char * data, int size)
{
	struct pulse_device_data * pulse_data;
	int len;
	int ret;
	uint8 * decoded_data;
	char * src;
	int decoded_size;

	pulse_data = (struct pulse_device_data *) devplugin->device_data;
	if (!pulse_data->stream)
		return 1;

	if (pulse_data->format == 0x11)
	{
		decoded_data = pulse_data->pDecodeImaAdpcm(&pulse_data->adpcm,
			(uint8 *) data, size, pulse_data->sample_spec.channels, pulse_data->block_size, &decoded_size);
		size = decoded_size;
		src = (char *) decoded_data;
	}
	else
	{
		decoded_data = NULL;
		src = data;
	}

	LLOGLN(10, ("rdpsnd_pulse_play: size %d", size));

	pa_threaded_mainloop_lock(pulse_data->mainloop);
	while (size > 0)
	{
		while ((len = pa_stream_writable_size(pulse_data->stream)) == 0)
		{
			LLOGLN(10, ("rdpsnd_pulse_play: waiting"));
			pa_threaded_mainloop_wait(pulse_data->mainloop);
		}
		if (len < 0)
			break;
		if (len > size)
			len = size;
		ret = pa_stream_write(pulse_data->stream, src, len, NULL, 0LL, PA_SEEK_RELATIVE);
		if (ret < 0)
		{
			LLOGLN(0, ("rdpsnd_pulse_play: pa_stream_write failed (%d)",
				pa_context_errno(pulse_data->context)));
			break;
		}
		src += len;
		size -= len;
	}
	pa_threaded_mainloop_unlock(pulse_data->mainloop);

	if (decoded_data)
		free(decoded_data);

	return 0;
}

/** Play the specified data to the pulseaudio server */
static int play(void* data, int len, int flags) {
    pa_threaded_mainloop_lock(mainloop);
    if (pa_stream_write(stream, data, len, NULL, 0, PA_SEEK_RELATIVE) < 0) {
        GENERIC_ERR_MSG(context, "pa_stream_write() failed");
        len = -1;
    }
    pa_threaded_mainloop_unlock(mainloop);
    return len;
}

void AudioOutputPulseAudio::WriteAudio(uchar *aubuf, int size)
{
    QString fn_log_tag = "WriteAudio, ";
    pa_stream_state_t sstate = pa_stream_get_state(pstream);

    VBAUDIOTS(fn_log_tag + QString("writing %1 bytes").arg(size));

    /* NB This "if" check can be replaced with PA_STREAM_IS_GOOD() in
       PulseAudio API from 0.9.11. As 0.9.10 is still widely used
       we use the more verbose version for now */

    if (sstate == PA_STREAM_CREATING || sstate == PA_STREAM_READY)
    {
        int write_status = PA_ERR_INVALID;
        size_t to_write = size;
        unsigned char *buf_ptr = aubuf;

        pa_threaded_mainloop_lock(mainloop);
        while (to_write > 0)
        {
            write_status = 0;
            size_t writable = pa_stream_writable_size(pstream);
            if (writable > 0)
            {
                size_t write = min(to_write, writable);
                write_status = pa_stream_write(pstream, buf_ptr, write,
                                               NULL, 0, PA_SEEK_RELATIVE);

                if (0 != write_status)
                    break;

                buf_ptr += write;
                to_write -= write;
            }
            else
            {
                pa_threaded_mainloop_wait(mainloop);
            }
        }
        pa_threaded_mainloop_unlock(mainloop);

        if (to_write > 0)
        {
            if (write_status != 0)
                VBERROR(fn_log_tag + QString("stream write failed: %1")
                                     .arg(write_status == PA_ERR_BADSTATE
                                                ? "PA_ERR_BADSTATE"
                                                : "PA_ERR_INVALID"));

            VBERROR(fn_log_tag + QString("short write, %1 of %2")
                                 .arg(size - to_write).arg(size));
        }
    }
    else
        VBERROR(fn_log_tag + QString("stream state not good: %1")
                             .arg(sstate,0,16));
}

static int outstream_end_write_pa(SoundIoPrivate *si, SoundIoOutStreamPrivate *os) {
    SoundIoOutStreamPulseAudio *ospa = &os->backend_data.pulseaudio;
    pa_stream *stream = ospa->stream;

    pa_seek_mode_t seek_mode = ospa->clear_buffer_flag.test_and_set() ? PA_SEEK_RELATIVE : PA_SEEK_RELATIVE_ON_READ;
    if (pa_stream_write(stream, ospa->write_ptr, ospa->write_byte_count, nullptr, 0, seek_mode))
        return SoundIoErrorStreaming;

    return 0;
}

void PulseAudio::paCallback(pa_stream* s, size_t len, void* data)
      {
      PulseAudio* pa = (PulseAudio*)data;
      size_t n = FRAMES * 2 * sizeof(float);
      if (len > n)
            len = n;
      float* p = pa->buffer;
      pa->seq->process(len / (2 * sizeof(float)), p);
      pa_stream_write(s, p, len, NULL, 0LL, PA_SEEK_RELATIVE);
      }

static int outstream_end_write_pa(struct SoundIoPrivate *si, struct SoundIoOutStreamPrivate *os) {
    struct SoundIoOutStreamPulseAudio *ospa = &os->backend_data.pulseaudio;
    pa_stream *stream = ospa->stream;

    pa_seek_mode_t seek_mode = SOUNDIO_ATOMIC_FLAG_TEST_AND_SET(ospa->clear_buffer_flag) ? PA_SEEK_RELATIVE : PA_SEEK_RELATIVE_ON_READ;
    if (pa_stream_write(stream, ospa->write_ptr, ospa->write_byte_count, NULL, 0, seek_mode))
        return SoundIoErrorStreaming;

    return 0;
}

void PulseAudio::paCallback(pa_stream* s, size_t len, void* data)
      {
      PulseAudio* pa = (PulseAudio*)data;
      constexpr size_t n = FRAMES * 2 * sizeof(float);
      if (len > n) {
            qDebug("PulseAudio:: buffer too large!");
            len = n;
            }
      pa->seq->process(len / (2 * sizeof(float)), pa->buffer);
      pa_stream_write(s, pa->buffer, len, nullptr, int64_t(0), PA_SEEK_RELATIVE);
      }

void PulseAudio::WriteCallback(pa_stream* s, size_t length)
{
	// fetch dst buffer directly from pulseaudio, so no memcpy is needed
	void* buffer;
	m_pa_error = pa_stream_begin_write(s, &buffer, &length);

	if (!buffer || m_pa_error < 0)
		return; // error will be printed from main loop

	m_mixer->Mix((s16*) buffer, length / sizeof(s16) / CHANNEL_COUNT);
	m_pa_error = pa_stream_write(s, buffer, length, nullptr, 0, PA_SEEK_RELATIVE);
}