refactor:calculate encoder pts based on decoder pts

include/AudioBuffer.h

@@ -25,23 +25,26 @@ class AudioBuffer
 	void	 SetClockRate(DWORD clockRate) { this->clockRate = clockRate; }
 
 	const int16_t* GetData() const { return pcmBuffer.data(); }
-
-	uint16_t SetSamples(int16_t* in, uint16_t numSamples)
+	uint16_t SetSamples(int16_t* in, uint16_t numSamples, size_t offset=0)
 	{
 		if (!in)
 			return Error("AudioBuffer::SetSamples() empty input buffer\n");
 
-		uint16_t totalResampled = numSamples * numChannels;
-		if (totalResampled != pcmBuffer.size())
-		{
-			Debug("AudioBuffer::SetSamples buffer resized, resized to =%d\n", totalResampled);
-			pcmBuffer.resize(totalResampled);
-		}
-
-		memcpy((int16_t*)pcmBuffer.data(), in, sizeof(int16_t) * totalResampled);
+		if (numSamples > pcmBuffer.size())
+			return Error("AudioBuffer::SetSamples() not enough buffer size\n");
+		memcpy((int16_t*)pcmBuffer.data()+offset, in, sizeof(int16_t) * numSamples);
 		return numSamples;
 	}
 
+	bool Resize(size_t samples) 
+	{
+		if (samples*numChannels > pcmBuffer.size())
+			return Error("AudioBuffer::Resize() resize to larger size\n");
+
+		pcmBuffer.resize(samples*numChannels);
+		return true;
+	}
+
 	uint16_t SetPCMData(uint8_t** pcmData, uint16_t numSamples)
 	{
 		if (!pcmData || !*pcmData)
@@ -76,7 +79,7 @@ class AudioBuffer
 		return true;
 	}
 	uint8_t  GetNumChannels() const { return numChannels; }
-	uint16_t GetNumSamples() const { return pcmBuffer.size() / numChannels; }
+	uint16_t GetNumSamples() const { return pcmBuffer.size(); }
 
 private:
 	uint16_t numSamplesPerFrame;

include/AudioBufferPool.h

@@ -0,0 +1,91 @@
+#ifndef AUDIOBUFFERPOOL_H_
+#define AUDIOBUFFERPOOL_H_
+
+#include "concurrentqueue.h"
+#include "AudioBuffer.h"
+
+class AudioBufferPool
+{
+public:
+	AudioBufferPool(std::size_t preallocate, std::size_t maxallocate) :
+		preallocate(preallocate),
+		maxallocate(maxallocate)
+	{
+	}
+
+	void SetSize(DWORD numSamples, DWORD numChannels)
+	{
+		//Make sure we have a new size
+		if (this->numSamples==numSamples && this->numChannels==numChannels)
+			//Do nothing
+			return;
+
+		//Try deallocate old buffers now, but we will check the size when allocating a new one
+		AudioBuffer* buffer;
+
+		//Get all the object from the pool
+		while (pool->try_dequeue(buffer))
+			//Delete them
+			delete(buffer);
+
+		//Store new size
+		this->numSamples = numSamples;
+		this->numChannels = numChannels;
+		//Allocate some buffer objects by default
+		for (std::size_t i = 0; i < preallocate; ++i)
+			pool->enqueue(new AudioBuffer(numSamples, numChannels));
+	}
+
+	~AudioBufferPool()
+	{
+		AudioBuffer* buffer;
+
+		//Get all the object from the pool
+		while (pool->try_dequeue(buffer))
+			//Delete them
+			delete(buffer);
+	}
+
+	AudioBuffer::shared Acquire()
+	{
+		AudioBuffer* buffer = nullptr;
+		//Try to get one from the pool
+		if (!pool->try_dequeue(buffer))
+		{
+			//Create a new one
+			buffer = new AudioBuffer(numSamples, numChannels);
+		}
+		//Make sure that it is from same size
+		else if (!buffer || buffer->GetNumSamples()!=numSamples*numChannels || buffer->GetNumChannels()!=numChannels)
+		{
+			//Delete old one
+			delete (buffer);
+			//Create a new one
+			buffer = new AudioBuffer(numSamples, numChannels);
+		}
+		//We need to create a new one
+		return AudioBuffer::shared(buffer, [maxallocate = maxallocate, weak = std::weak_ptr<moodycamel::ConcurrentQueue<AudioBuffer*>>(pool) ](auto p) {
+			//Try to get a reference to the pool, as it may have been already deleted
+			auto pool = weak.lock();
+			//Ensure we are not overallocating
+			if (maxallocate && pool && pool->size_approx() < maxallocate)
+			{
+				// Enqueue it back
+				pool->enqueue(p);
+			} else {
+				//Delete it
+				delete (p);
+			}
+		});
+	}
+
+private:
+	std::shared_ptr<moodycamel::ConcurrentQueue<AudioBuffer*>> pool = std::make_shared<moodycamel::ConcurrentQueue<AudioBuffer*>>();
+
+	std::size_t preallocate = 0;
+	std::size_t maxallocate = 0;
+
+	DWORD numSamples = 0;
+	DWORD numChannels = 0;
+};
+#endif // !AUDIOBUFFERPOOL_H_

include/AudioPipe.h

@@ -6,6 +6,7 @@
 #include "AudioTransrater.h"
 #include <AudioBuffer.h>
 #include <limits>
+#include "CircularQueue.h"
 
 
 class AudioPipe : 
@@ -34,11 +35,17 @@ class AudioPipe :
 	virtual DWORD GetNumChannels()		{ return numChannels;	}
 
 private:
+	static constexpr int quantizationErrToleranceInMs = 5;
 	//Los mutex y condiciones
 	pthread_mutex_t mutex;
 	pthread_cond_t  cond;
+	// track available num samples
+	int availableNumSamples = 0;
+
+	CircularQueue<AudioBuffer::const_shared> queue;
 	//Members
 	fifo<SWORD,48000*4>	rateBlockBuffer;
+	QWORD rateBlockBufferPTS;
 
 	bool			recording = false;
 	bool			playing = false;
@@ -50,7 +57,6 @@ class AudioPipe :
 	DWORD			recordRate = 0;
 	DWORD			nativeRate = 0;
 	DWORD			numChannels = 1;
-	DWORD			cache = 0;
 };
 
 #endif /* AUDIOPIPE_H */

include/AudioTransrater.h

@@ -1,6 +1,7 @@
 #ifndef _AUDIOTRANSRATER_H_
 #define _AUDIOTRANSRATER_H_
 #include "speex/speex_resampler.h"
+#include "AudioBufferPool.h"
 
 class AudioTransrater
 {
@@ -9,13 +10,19 @@ class AudioTransrater
 	~AudioTransrater();
 
 	int Open(DWORD inputRate, DWORD outputRate, DWORD numChannels = 1);
-	int ProcessBuffer(SWORD* in, DWORD sizeIn, SWORD* out, DWORD* sizeOut);
+	AudioBuffer::shared ProcessBuffer(const AudioBuffer::shared& audioBuffer);
 	void Close();
 
 	bool IsOpen()	{ return resampler!=NULL; }
 
 private:
+	static constexpr size_t InitialPoolSize = 30;
+	static constexpr size_t MaxPoolSize = InitialPoolSize + 2;
 	SpeexResamplerState *resampler;
+	DWORD inputRate=0;
+	DWORD outputRate=0;
+	std::optional<QWORD> playPTSOffset;
+	AudioBufferPool audioBufferPool;
 };
 
 #endif

include/VideoBufferPool.h

@@ -46,7 +46,7 @@ class VideoBufferPool
 			delete(buffer);
 	}
 
-	VideoBuffer::shared allocate()
+	VideoBuffer::shared Acquire()
 	{
 		VideoBuffer* buffer = nullptr;