Collaboration diagram for state_audio_mixer:

Public Member Functions
	state_audio_mixer (const char *cfg)

	state_audio_mixer (state_audio_mixer const &)=delete

	~state_audio_mixer ()

state_audio_mixer &	operator= (state_audio_mixer const &)=delete

void	worker ()

Public Attributes
string	audio_codec {"PCM"}

map< sockaddr_storage, am_participant, sockaddr_storage_less >	participants

mutex	participants_lock

struct socket_udp_local *	recv_socket {}

bool	should_exit = false

Private Attributes
unique_ptr< generic_mix_algo< sample_type_source, sample_type_mixed > >	mixing_algorithm {new linear_mix_algo<sample_type_source, sample_type_mixed>()}

thread	thread_id

Detailed Description

Definition at line 227 of file mixer.cpp.

Constructor & Destructor Documentation

◆ state_audio_mixer() [1/2]

state_audio_mixer::state_audio_mixer ( const char * cfg )

inline

Definition at line 228 of file mixer.cpp.

                                           {
                if (cfg) {
                        shared_ptr<char> tmp(strdup(cfg), free);
                        char *item, *save_ptr;
                        char *copy = tmp.get();
 
                        while ((item = strtok_r(copy, ":", &save_ptr))) {
                                if (strncmp(item, "codec=", strlen("codec=")) == 0) {
                                        audio_codec = item + strlen("codec=");
                                } else if (strncmp(item, "algo=", strlen("algo=")) == 0) {
                                        string algo = item + strlen("algo=");
                                        if (algo == "linear") {
                                                mixing_algorithm = decltype(mixing_algorithm)(new linear_mix_algo<sample_type_source, sample_type_mixed>());
                                        } else if (algo == "logarithmic") {
                                                mixing_algorithm = decltype(mixing_algorithm)(new logarithmic_mix_algo<sample_type_source, sample_type_mixed>());
                                        } else {
                                                LOG(LOG_LEVEL_ERROR) << "Unknown mixing algorithm: " << algo << "\n";
                                                throw 1;
                                        }
                                } else {
                                        LOG(LOG_LEVEL_ERROR) << "Unknown option: " << item << "\n";
                                        throw 1;
                                }
                                copy = nullptr;
                        }
                }
 
                struct audio_codec_state *audio_coder =
                        audio_codec_init_cfg(audio_codec.c_str(), AUDIO_CODER);
                if (!audio_coder) {
                        LOG(LOG_LEVEL_ERROR) << "Audio coder init failed!\n";
                        throw 1;
                } else {
                        audio_codec_done(audio_coder);
                }
 
                thread_id = thread(&state_audio_mixer::worker, this);
        }

References audio_codec, audio_codec_done(), audio_codec_init_cfg(), AUDIO_CODER, LOG, LOG_LEVEL_ERROR, mixing_algorithm, thread_id, and worker().

Here is the call graph for this function:

◆ ~state_audio_mixer()

state_audio_mixer::~state_audio_mixer ( )

inline

Definition at line 266 of file mixer.cpp.

                             {
                thread_id.join();
        }

References thread_id.

◆ state_audio_mixer() [2/2]

state_audio_mixer::state_audio_mixer ( state_audio_mixer const & )

delete

Member Function Documentation

◆ operator=()

state_audio_mixer & state_audio_mixer::operator= ( state_audio_mixer const & )

delete

◆ worker()

void state_audio_mixer::worker ( )

Definition at line 284 of file mixer.cpp.

{
        set_thread_name(__func__);
        chrono::steady_clock::time_point next_frame_time = chrono::steady_clock::now();
 
        static_assert(SAMPLES_PER_FRAME * 1000ll % SAMPLE_RATE == 0, "Sample rate is not evenly divisible by number of samples in frame");
        const chrono::milliseconds interval(SAMPLES_PER_FRAME*1000ll/SAMPLE_RATE);
 
        while (!should_exit) {
                this_thread::sleep_until(next_frame_time);
                next_frame_time += interval;
                auto now = chrono::steady_clock::now();
 
                // check if we didn't overslept much
                if (next_frame_time < now) {
                        LOG(LOG_LEVEL_WARNING) << "[Audio mixer] Next frame time in past! Setting to now.\n";
                        next_frame_time = now;
                }
 
                unique_lock<mutex> plk(participants_lock);
                // check timeouts
                for (auto it = participants.cbegin(); it != participants.cend(); )
                {
                        if (duration_cast<seconds>(now - it->second.last_seen).count() > PARTICIPANT_TIMEOUT_S) {
                                it = participants.erase(it);
                        } else {
                                ++it;
                        }
                }
 
                size_t data_len_source = SAMPLES_PER_FRAME * sizeof(sample_type_source) * CHANNELS;
                vector<sample_type_mixed> mixed(SAMPLES_PER_FRAME * CHANNELS);
                vector<audio_frame2> participant_frames(participants.size());
                int participant_index = 0;
 
                // mix all together
                for (auto & p : participants) {
                        participant_frames[participant_index].init(CHANNELS, AC_PCM, BPS, SAMPLE_RATE);
                        static_assert(CHANNELS == 1, "Currently only one channel is implemented here.");
                        participant_frames[participant_index].resize(0, data_len_source);
                        char *particip_data = participant_frames[participant_index].get_data(0);
                        int ret = audio_buffer_read(p.second.m_buffer, particip_data, data_len_source);
                        memset(particip_data, 0, data_len_source - ret);
 
                        sample_type_source *src = (sample_type_source *)(void *) particip_data;
                        auto dst = mixed.begin();
                        for (int i = 0; i < SAMPLES_PER_FRAME * CHANNELS; ++i) {
                                *dst = mixing_algorithm->add_to_mix(*dst, *src);
                                ++dst;
                                ++src;
                        }
                        participant_index++;
                }
 
                // substract each source signal from the mix coming to that participant
                for (auto & pb : participant_frames) {
                        auto mix = mixed.begin();
                        sample_type_source *part = (sample_type_source *)(void *) pb.get_data(0);
                        for (int i = 0; i < SAMPLES_PER_FRAME * CHANNELS; ++i) {
                                *part = mixing_algorithm->normalize(mixing_algorithm->get_mixed_without_source_sample(*mix, *part));
                                ++mix;
                                ++part;
                        }
                }
 
                // send
                participant_index = 0;
                for (auto & p : participants) {
                        audio_frame2 *uncompressed = &participant_frames[participant_index];
                        while (audio_frame2 compressed = audio_codec_compress(p.second.m_audio_coder, uncompressed)) {
                                audio_tx_send(p.second.m_tx_session, p.second.m_network_device, &compressed);
                                uncompressed = nullptr;
                        }
 
                        participant_index++;
                }
                plk.unlock();
        }
}