Implement AudioFramesMixer class. #1

AudioFramesMixer.cpp

@@ -1,35 +1,16 @@
 #include "AudioFramesMixer.hpp"
 
+#include <boost/format.hpp>
 
-//// Reset mix buffer.
-//Arrays.fill(tempMix, 0);
-//
-//// Sum the buffers.
-//for (final AudioUser user : mix) {
-//for (int i = 0; i < tempMix.length; i++) {
-//tempMix[i] += user.lastFrame[i];
-//}
-//}
-//
-//// Clip buffer for real output.
-//for (int i = 0; i < MumbleProtocol.FRAME_SIZE; i++) {
-//clipOut[i] = (short) (Short.MAX_VALUE * (tempMix[i] < -1.0f ? -1.0f
-//: (tempMix[i] > 1.0f ? 1.0f : tempMix[i])));
-//}
+#include <climits>
 
 mixer::AudioFramesMixer::AudioFramesMixer(pj_pool_factory &poolFactory)
         : logger(log4cpp::Category::getInstance("AudioFramesMixer")) {
 
-    pool = pj_pool_create(&poolFactory, "media", 32768, 8192, nullptr);
+    pool = pj_pool_create(&poolFactory, "mixer_pool", 10 * 1024, 10 * 1024, nullptr);
     if (!pool) {
         throw mixer::Exception("error when creating memory pool");
     }
-
-    // todo calculate sizes
-    pj_status_t status = pjmedia_circ_buf_create(pool, 960 * 10, &inputBuff);
-    if (status != PJ_SUCCESS) {
-        throw mixer::Exception("error when creating circular buffer", status);
-    }
 }
 
 mixer::AudioFramesMixer::~AudioFramesMixer() {
@@ -40,22 +21,104 @@ mixer::AudioFramesMixer::~AudioFramesMixer() {
 
 void mixer::AudioFramesMixer::addFrameToBuffer(int sessionId, int sequenceNumber, int16_t *samples, int samplesLength) {
     std::unique_lock<std::mutex> lock(inBuffAccessMutex);
-    logger.debug("Pushing %d samples to in-buff.", samplesLength);
-    pjmedia_circ_buf_write(inputBuff, samples, samplesLength);
+
+    pjmedia_circ_buf *circBuf;
+    pj_status_t status;
+
+    auto it = buffersMap.find(sessionId);
+
+    if (it != buffersMap.end()) {
+        circBuf = it->second;
+    } else {
+        logger.debug("Creating circular buffer for session %d.", sessionId);
+        status = pjmedia_circ_buf_create(pool, 960 * 10, &circBuf);
+        if (status != PJ_SUCCESS) {
+            throw mixer::Exception("error when creating circular buffer", status);
+        }
+        buffersMap.insert({{sessionId, circBuf}});
+    }
+
+    logger.debug("Pushing %d samples to buffer for session %d.", samplesLength, sessionId);
+    status = pjmedia_circ_buf_write(circBuf, samples, samplesLength);
+    if (status != PJ_SUCCESS and status != PJ_ETOOBIG) {
+        throw mixer::Exception((boost::format("error when writing %d samples to circular buffer")
+                                % samplesLength).str(), status);
+    }
 }
 
 int mixer::AudioFramesMixer::getMixedSamples(int16_t *mixedSamples, int requestedLength) {
     std::unique_lock<std::mutex> lock(inBuffAccessMutex);
 
-    int availableSamples = pjmedia_circ_buf_get_len(inputBuff);
+    double mixerBuffer[MAX_BUFFER_LENGTH];
+    memset(mixerBuffer, 0, sizeof(mixerBuffer));
+
+    int longestSamples = 0;
+
+    for (auto &user: buffersMap) {
+        int16_t userBuff[MAX_BUFFER_LENGTH];
+
+        int availableSamples = pjmedia_circ_buf_get_len(user.second);
         const int samplesToRead = std::min(requestedLength, availableSamples);
 
-    logger.debug("Pulling %d samples from in-buff.", samplesToRead);
-    pjmedia_circ_buf_read(inputBuff, mixedSamples, samplesToRead);
+        longestSamples = std::max(samplesToRead, longestSamples);
 
-    return samplesToRead;
+        logger.debug("Pulling %d samples from in-buff for session ID %d.", samplesToRead, user.first);
+
+        pj_status_t status = pjmedia_circ_buf_read(user.second, userBuff, samplesToRead);
+        if (status != PJ_SUCCESS) {
+            throw mixer::Exception(
+                    (boost::format("error when pulling %d samples from buffer for session ID %d")
+                     % samplesToRead % user.first).str(), status);
         }
 
-void mixer::AudioFramesMixer::clean() {
-
+        for (int i = 0; i < samplesToRead; ++i) {
+            mixerBuffer[i] += userBuff[i];
+        }
+    }
+
+    for (auto it = buffersMap.cbegin(); it != buffersMap.cend() /* not hoisted */; /* no increment */) {
+        if (pjmedia_circ_buf_get_len(it->second) == 0) {
+            logger.debug("Removing circular buffer for session %d.", it->first);
+            pj_status_t status = pjmedia_circ_buf_reset(it->second);
+            if (status != PJ_SUCCESS) {
+                throw mixer::Exception(
+                        (boost::format("error when resetting circular buffer for session ID %d") % it->first).str(),
+                        status);
+            }
+            buffersMap.erase(it++);
+        }
+        else {
+            ++it;
+        }
+    }
+
+    double maxVal = 0;
+    for (int i = 0; i < longestSamples; ++i) {
+        maxVal = std::max(maxVal, std::abs(mixerBuffer[i]));
+    }
+
+    if (maxVal >= INT16_MAX) {
+        for (int i = 0; i < longestSamples; ++i) {
+            mixedSamples[i] = (INT16_MAX * (mixerBuffer[i] / maxVal));
+        }
+        logger.debug("Mixer overdrive, truncating to 16-bit.");
+    } else {
+        for (int i = 0; i < longestSamples; ++i) {
+            mixedSamples[i] = mixerBuffer[i];
+        }
+    }
+
+    logger.debug("Getting %d mixed samples.", longestSamples);
+
+    return longestSamples;
+}
+
+void mixer::AudioFramesMixer::clear() {
+    std::unique_lock<std::mutex> lock(inBuffAccessMutex);
+
+    for (auto &entry : buffersMap) {
+        pjmedia_circ_buf_reset(entry.second);
+    }
+
+    buffersMap.clear();
 }

AudioFramesMixer.hpp

@@ -6,16 +6,19 @@
 #include <boost/noncopyable.hpp>
 
 #include <mutex>
+#include <unordered_map>
 
 namespace mixer {
 
+    constexpr int MAX_BUFFER_LENGTH = 5000;
+
     class Exception : public std::runtime_error {
     public:
-        Exception(const char *title) : std::runtime_error(title) {
+        Exception(std::string title) : std::runtime_error(title) {
             mesg += title;
         }
 
-        Exception(const char *title, pj_status_t status) : std::runtime_error(title) {
+        Exception(std::string title, pj_status_t status) : std::runtime_error(title) {
             char errorMsgBuffer[500];
             pj_strerror(status, errorMsgBuffer, sizeof(errorMsgBuffer));
 
@@ -42,14 +45,14 @@ namespace mixer {
 
         int getMixedSamples(int16_t *mixedSamples, int requestedLength);
 
-        void clean();
+        void clear();
 
     private:
         log4cpp::Category &logger;
 
         pj_pool_t *pool = nullptr;
 
-        pjmedia_circ_buf *inputBuff;
+        std::unordered_map<int, pjmedia_circ_buf *> buffersMap;
 
         std::mutex inBuffAccessMutex;
     };

PjsuaCommunicator.cpp

@@ -142,6 +142,8 @@ namespace sip {
             if (not acc.available) {
                 auto msgText = "Call from " + address + " finished.";
 
+                communicator.mixer->clear();
+
                 communicator.logger.notice(msgText);
                 communicator.onStateChange(msgText);