TeaSpeak-Client/native/serverconnection/src/audio/AudioInput.cpp

#include <cstring>
#include <string>
#include <misc/spin_lock.h>
#include "./AudioInput.h"
#include "./AudioReframer.h"
#include "./AudioResampler.h"
#include "../logger.h"
#include "AudioGain.h"

using namespace std;
using namespace tc;
using namespace tc::audio;

#if false
class AudioInputSource {
    public:
        constexpr static auto kChannelCount{2};
        constexpr static auto kSampleRate{48000};

        explicit AudioInputSource(PaHostApiIndex index) : device_index{index} {}
        ~AudioInputSource() = default;

        /* its blocking! */
        void register_consumer(AudioInput* consumer) {
            std::lock_guard lock{this->registered_inputs_lock};
            if(find(this->registered_inputs.begin(), this->registered_inputs.end(), consumer) != this->registered_inputs.end())
                return;

            this->registered_inputs.push_back(consumer);
        }

        /* its blocking */
        void remove_consumer(AudioInput* consumer) {
            std::lock_guard lock{this->registered_inputs_lock};

            auto index = find(this->registered_inputs.begin(), this->registered_inputs.end(), consumer);
            if(index == this->registered_inputs.end())
                return;

            this->registered_inputs.erase(index);
            if(!this->registered_inputs.empty())
                return;
        }

        /* this could take a bit longer! */
        bool begin_recording(std::string& error) {
            std::lock_guard lock{this->state_lock};
            if(this->state == RECORDING) return true;

            if(this->state != STOPPED) {
                if(this->state == DELETED) {
                    error = "stream has been deleted";
                    return false;
                }
                error = "invalid state";
                return false;
            }

            this->current_device = Pa_GetDeviceInfo(this->device_index);
            if(!this->current_device) {
                error = "failed to get device info";
                return false;
            }

            PaStreamParameters parameters{};
            memset(&parameters, 0, sizeof(parameters));
            parameters.channelCount = (int) kChannelCount;
            parameters.device = this->device_index;
            parameters.sampleFormat = paFloat32;
            parameters.suggestedLatency = this->current_device->defaultLowOutputLatency;
            auto err = Pa_OpenStream(
                    &this->input_stream,
                    &parameters,
                    nullptr,
                    (double) kSampleRate,
                    paFramesPerBufferUnspecified,
                    paClipOff,
                    &AudioInputSource::pa_audio_callback,
                    this);

            if(err != paNoError) {
                this->input_stream = nullptr;
                error = to_string(err) + "/" + Pa_GetErrorText(err);
                return false;
            }

            err = Pa_StartStream(this->input_stream);
            if(err != paNoError) {
                error = "recording failed " + to_string(err) + "/" + Pa_GetErrorText(err);
                err = Pa_CloseStream(this->input_stream);
                if(err != paNoError)
                    log_critical(category::audio, tr("Failed to close opened pa stream. This will cause memory leaks. Error: {}/{}"), err, Pa_GetErrorText(err));
                return false;
            }
            this->state = RECORDING;

            return true;
        }

        void stop_recording_if_possible() {
            std::lock_guard lock{this->state_lock};
            if(this->state != RECORDING) return;

            {
                std::lock_guard client_lock{this->registered_inputs_lock};
                if(!this->registered_inputs.empty()) return;
            }
            this->state = STOPPED;

            if(Pa_IsStreamActive(this->input_stream))
                Pa_AbortStream(this->input_stream);

            auto error = Pa_CloseStream(this->input_stream);
            if(error != paNoError)
                log_error(category::audio, tr("Failed to close PA stream: {}"), error);
            this->input_stream = nullptr;
        }

        const PaDeviceIndex device_index;
    private:
        static int pa_audio_callback(const void *input, void *output, unsigned long frameCount, const PaStreamCallbackTimeInfo* timeInfo, PaStreamCallbackFlags statusFlags, void* _input_source) {
            if(!input) return 0; /* this should never happen */
            auto input_source = (AudioInputSource*) _input_source;

            std::lock_guard lock{input_source->registered_inputs_lock};
            for(auto& client : input_source->registered_inputs)
                client->consume(input, frameCount, 2);
            return 0;
        }

        std::mutex state_lock{};
        enum _state {
            STOPPED,
            RECORDING,
            DELETED
        } state{STOPPED};
        PaStream* input_stream{nullptr};
        const PaDeviceInfo* current_device = nullptr;

        std::mutex registered_inputs_lock{};
        std::vector<AudioInput*> registered_inputs{};
};
std::mutex input_sources_lock{};
static std::deque<std::shared_ptr<AudioInputSource>> input_sources{};

std::shared_ptr<AudioInputSource> get_input_source(PaDeviceIndex device_index, bool create = true) {
    std::lock_guard sources_lock{input_sources_lock};
    for(const auto& input : input_sources)
        if(input->device_index == device_index)
            return input;
    if(!create)
        return nullptr;

    auto input = std::make_shared<AudioInputSource>(device_index);
    input_sources.push_back(std::make_shared<AudioInputSource>(device_index));
    return input;
}
#endif

AudioConsumer::AudioConsumer(tc::audio::AudioInput *handle, size_t channel_count, size_t sample_rate, size_t frame_size) :
	handle(handle),
	channel_count(channel_count),
	sample_rate(sample_rate) ,
	frame_size(frame_size) {
	if(this->frame_size > 0) {
		this->reframer = make_unique<Reframer>(channel_count, frame_size);
		this->reframer->on_frame = [&](const void* buffer) { this->handle_framed_data(buffer, this->frame_size); };
	}
}

void AudioConsumer::handle_framed_data(const void *buffer, size_t samples) {
	unique_lock read_callback_lock(this->on_read_lock);
	auto function = this->on_read; /* copy */
	read_callback_lock.unlock();
	if(!function)
		return;

	function(buffer, samples);
}

void AudioConsumer::process_data(const void *buffer, size_t samples) {
	if(this->reframer)
		this->reframer->process(buffer, samples);
	else
		this->handle_framed_data(buffer, samples);
}

AudioInput::AudioInput(size_t channels, size_t rate) : _channel_count(channels), _sample_rate(rate) {}
AudioInput::~AudioInput() {
	this->close_device();
    {
        lock_guard lock(this->consumers_lock);
        for(const auto& consumer : this->_consumers)
            consumer->handle = nullptr;
    }
	free(this->resample_buffer);
}

void AudioInput::set_device(const std::shared_ptr<AudioDevice> &device) {
    lock_guard lock(this->input_source_lock);
    if(device == this->input_device) return;

    this->close_device();
    this->input_device = device;
}

void AudioInput::close_device() {
    lock_guard lock(this->input_source_lock);
    if(this->input_recorder) {
        this->input_recorder->remove_consumer(this);
        this->input_recorder->stop_if_possible();
        this->input_recorder.reset();
    }
    this->_resampler = nullptr;
    this->input_device = nullptr;
}

bool AudioInput::record(std::string& error) {
    lock_guard lock(this->input_source_lock);
    if(!this->input_device) {
        error = "no device";
        return false;
    }
    if(this->input_recorder) return true;

    this->input_recorder = this->input_device->record();
    if(!this->input_recorder) {
        error = "failed to get recorder";
        return false;
    }

    if(this->input_recorder->sample_rate() != this->sample_rate()) {
        this->_resampler = std::make_unique<AudioResampler>(this->input_recorder->sample_rate(), this->sample_rate(), this->_channel_count);
    }

    this->input_recorder->register_consumer(this);
    if(!this->input_recorder->start(error)) {
        this->input_recorder->remove_consumer(this);
        this->input_recorder.reset();
        return false;
    }
    return true;
}

bool AudioInput::recording() {
    return !!this->input_recorder;
}

void AudioInput::stop() {
    if(!this->input_recorder) return;

    this->input_recorder->remove_consumer(this);
    this->input_recorder->stop_if_possible();
    this->input_recorder.reset();
}

std::shared_ptr<AudioConsumer> AudioInput::create_consumer(size_t frame_length) {
	auto result = shared_ptr<AudioConsumer>(new AudioConsumer(this, this->_channel_count, this->_sample_rate, frame_length));
	{
		lock_guard lock(this->consumers_lock);
		this->_consumers.push_back(result);
	}
	return result;
}

void AudioInput::delete_consumer(const std::shared_ptr<AudioConsumer> &source) {
	{
		lock_guard lock(this->consumers_lock);
		auto it = find(this->_consumers.begin(), this->_consumers.end(), source);
		if(it != this->_consumers.end())
			this->_consumers.erase(it);
	}

	source->handle = nullptr;
}

void AudioInput::consume(const void *input, size_t frameCount, size_t channels) {
    if(channels != this->_channel_count) {
        log_critical(category::audio, tr("Channel count miss match (input)! Fixme!"));
        return;
    }

	if(this->_resampler) {
	    const auto expected_size = this->_resampler->estimated_output_size(frameCount);
	    const auto expected_byte_size = expected_size * this->_channel_count * sizeof(float);

	    if(this->resample_buffer_size < expected_byte_size) {
	        free(this->resample_buffer);
	        this->resample_buffer = malloc(expected_byte_size);
	        this->resample_buffer_size = expected_byte_size;
	    }

	    auto result = this->_resampler->process(this->resample_buffer, input, frameCount);
	    if(result < 0) {
	        log_error(category::audio, tr("Failed to resample input audio: {}"), result);
	        return;
	    }

	    frameCount = (size_t) result;
	    input = this->resample_buffer;

	    audio::apply_gain(this->resample_buffer, this->_channel_count, frameCount, this->_volume);
	} else if(this->_volume != 1) {
        const auto byte_size = frameCount * this->_channel_count * sizeof(float);
        if(this->resample_buffer_size < byte_size) {
            free(this->resample_buffer);
            this->resample_buffer = malloc(byte_size);
            this->resample_buffer_size = byte_size;
        }

        memcpy(this->resample_buffer, input, byte_size);
        input = this->resample_buffer;

        audio::apply_gain(this->resample_buffer, this->_channel_count, frameCount, this->_volume);
	}

	auto begin = chrono::system_clock::now();
	for(const auto& consumer : this->consumers())
		consumer->process_data(input, frameCount);

	auto end = chrono::system_clock::now();
	auto ms = chrono::duration_cast<chrono::milliseconds>(end - begin).count();
	if(ms > 5) {
		log_warn(category::audio, tr("Processing of audio input needed {}ms. This could be an issue!"), chrono::duration_cast<chrono::milliseconds>(end - begin).count());
	}
}