AsyncAudioThreadSource.h

Go to the documentation of this file.

 
#ifndef ASYNC_AUDIO_THREAD_SOURCE_INCLUDED
#define ASYNC_AUDIO_THREAD_SOURCE_INCLUDED
 
 
/****************************************************************************
 *
 * System Includes
 *
 ****************************************************************************/
 
#include <vector>
#include <mutex>
#include <condition_variable>
//#include <iostream>
 
 
/****************************************************************************
 *
 * Project Includes
 *
 ****************************************************************************/
 
#include <AsyncApplication.h>
#include <AsyncAudioSink.h>
#include <AsyncAudioSource.h>
#include <AsyncTaskRunner.h>
 
 
/****************************************************************************
 *
 * Local Includes
 *
 ****************************************************************************/
 
 
 
/****************************************************************************
 *
 * Forward declarations
 *
 ****************************************************************************/
 
 
 
/****************************************************************************
 *
 * Namespace
 *
 ****************************************************************************/
 
namespace Async
{
 
 
/****************************************************************************
 *
 * Forward declarations of classes inside of the declared namespace
 *
 ****************************************************************************/
 
 
 
/****************************************************************************
 *
 * Defines & typedefs
 *
 ****************************************************************************/
 
 
 
/****************************************************************************
 *
 * Exported Global Variables
 *
 ****************************************************************************/
 
 
 
/****************************************************************************
 *
 * Class definitions
 *
 ****************************************************************************/
 
class AudioThreadSource : public Async::AudioSource
{
  public:
    AudioThreadSource(void) {}
 
    AudioThreadSource(const AudioThreadSource&) = delete;
 
    virtual ~AudioThreadSource(void)
    {
      assert(std::this_thread::get_id() == Async::Application::app().threadId());
    }
 
    /*
     * @brief   No assignment operator
     */
    AudioThreadSource& operator=(const AudioThreadSource&) = delete;
 
  protected:
    virtual int sinkWriteSamples(const float *samples, int count)
    {
      {
        std::lock_guard<std::mutex> lk(m_in_buf_mu);
        m_flush = false;
        m_in_buf.insert(m_in_buf.end(), samples, samples+count);
        if (!m_pending_processing)
        {
          m_pending_processing = true;
          m_runner(&AudioThreadSource::resumeOutput, this);
        }
      }
      return count;
    }
 
    virtual void sinkFlushSamples(void)
    {
      {
        std::lock_guard<std::mutex> lk(m_in_buf_mu);
        m_flush = true;
        if (!m_pending_processing)
        {
          m_pending_processing = true;
          m_runner(&AudioThreadSource::resumeOutput, this);
        }
      }
    }
 
    void waitForAllSamplesFlushed(void)
    {
      std::unique_lock<std::mutex> lk(m_all_flushed_mu);
      m_all_flushed_cond.wait(lk, [this]{ return m_all_flushed; });
    }
 
  private:
    std::mutex                    m_in_buf_mu;
    std::vector<float>            m_in_buf;
    std::vector<float>            m_out_buf;
    std::vector<float>::iterator  m_out_buf_it = m_out_buf.begin();
    bool                          m_flush = false;
    bool                          m_all_flushed = true;
    std::mutex                    m_all_flushed_mu;
    std::condition_variable_any   m_all_flushed_cond;
    bool                          m_pending_processing = false;
    Async::TaskRunner             m_runner;
 
    void resumeOutput(void)
    {
      for (;;)
      {
        {
          std::lock_guard<std::mutex> lk(m_in_buf_mu);
          m_pending_processing = false;
          if (m_out_buf.empty() && !m_in_buf.empty())
          {
            m_out_buf.swap(m_in_buf);
            m_out_buf_it = m_out_buf.begin();
          }
        }
        if (!m_out_buf.empty())
        {
          int write_cnt = m_out_buf.end()-m_out_buf_it;
          int cnt = AudioSource::sinkWriteSamples(&(*m_out_buf_it), write_cnt);
          if (cnt > 0)
          {
            m_out_buf_it += std::min(cnt, write_cnt);
            if (m_out_buf_it == m_out_buf.end())
            {
              m_out_buf.clear();
              m_out_buf_it = m_out_buf.begin();
            }
          }
          else
          {
            break;
          }
        }
        else
        {
          if (m_flush)
          {
            {
              std::lock_guard<std::mutex> lk(m_in_buf_mu);
              m_flush = false;
            }
            {
              std::lock_guard<std::mutex> lk(m_all_flushed_mu);
              m_all_flushed = false;
            }
            AudioSource::sinkFlushSamples();
          }
          break;
        }
      }
    }
 
    void allSamplesFlushed(void)
    {
      {
        std::lock_guard<std::mutex> lk(m_all_flushed_mu);
        m_all_flushed = true;
      }
      m_all_flushed_cond.notify_all();
    }
};  /* class AudioThreadSource */
 
 
} /* namespace */
 
#endif /* ASYNC_AUDIO_THREAD_SOURCE_INCLUDED */
 
/*
 * This file has not been truncated
 */