WIP streaming

Source/BansheeOpenAudio/Include/BsOAAudio.h

@@ -74,6 +74,39 @@ namespace BansheeEngine
 		/** @} */
 
 	private:
+		friend class OAAudioSource;
+
+		/** Type of a command that can be queued for a streaming audio source. */
+		enum class StreamingCommandType
+		{
+			Start,
+			Stop
+		};
+
+		/** Command queued for a streaming audio source. */
+		struct StreamingCommand
+		{
+			StreamingCommand()
+				:type((StreamingCommandType)0), source(nullptr), params()
+			{ }
+
+			StreamingCommandType type;
+			OAAudioSource* source;
+			UINT32 params[2];
+		};
+
+		/** Data required for maintaining a streaming audio source. */
+		struct StreamingData
+		{
+			StreamingData()
+				:streamBuffers()
+			{ }
+
+			static const UINT32 StreamBufferCount = 3;
+			UINT32 streamBuffers[StreamBufferCount];
+			Vector<UINT32> sourceIDs;
+		};
+
 		/** @copydoc Audio::createClip */
 		SPtr<AudioClip> createClip(const SPtr<DataStream>& samples, UINT32 streamSize, UINT32 numSamples,
 			const AUDIO_CLIP_DESC& desc) override;
@@ -95,6 +128,15 @@ namespace BansheeEngine
 		/** Delete all existing OpenAL contexts. */
 		void clearContexts();
 
+		/** Streams new data to audio sources that require it. */
+		void updateStreaming();
+
+		/** Starts data streaming for the provided source. */
+		void startStreaming(OAAudioSource* source, bool startPaused);
+
+		/** Stops data streaming for the provided source. */
+		void stopStreaming(OAAudioSource* source);
+
 		float mVolume;
 
 		ALCdevice* mDevice;
@@ -106,6 +148,11 @@ namespace BansheeEngine
 		Vector<ALCcontext*> mContexts;
 		UnorderedSet<OAAudioSource*> mSources;
 
+		// Streaming thread
+		Vector<StreamingCommand> mStreamingCommandQueue;
+		UnorderedMap<OAAudioSource*, StreamingData> mStreamingSources;
+		SPtr<Task> mStreamingTask;
+		mutable Mutex mMutex;
 	};
 
 	/** Provides easier access to OAAudio. */

Source/BansheeOpenAudio/Include/BsOAAudioSource.h

@@ -74,21 +74,13 @@ namespace BansheeEngine
 
 		/** 
 		 * Returns true if the audio source is receiving audio data from a separate thread (as opposed to loading it all
-		 * at once. */
+		 * at once. 
+		 */
 		bool isStreaming() const;
 
-		/** Creates OpenAL buffers used for streaming. */
-		void createStreamBuffers();
-
-		/** Destroys OpenAL buffers used for streaming. */
-		void destroyStreamBuffers();
-
 		Vector<UINT32> mSourceIDs;
-		float mSeekPosition;
-
-		static const UINT32 StreamBufferCount = 3;
-		mutable Mutex mMutex;
-		UINT32 mStreamBuffers[StreamBufferCount];
+		UINT32 mSeekPosition;
+		bool mRequiresStreaming;
 	};
 
 	/** @} */

Source/BansheeOpenAudio/Source/BsOAAudio.cpp

@@ -5,6 +5,7 @@
 #include "BsOAAudioListener.h"
 #include "BsOAAudioSource.h"
 #include "BsMath.h"
+#include "BsTaskScheduler.h"
 #include "AL\al.h"
 
 namespace BansheeEngine
@@ -97,7 +98,14 @@ namespace BansheeEngine
 
 	void OAAudio::_update()
 	{
-		// TODO
+		auto worker = [this]() { updateStreaming(); };
+
+		// If previous task still hasn't completed, just skip streaming this frame, queuing more tasks won't help
+		if (mStreamingTask != nullptr && !mStreamingTask->isComplete())
+			return;
+
+		mStreamingTask = Task::create("AudioStream", worker, TaskPriority::VeryHigh);
+		TaskScheduler::instance().addTask(mStreamingTask);
 	}
 
 	void OAAudio::setActiveDevice(const AudioDevice& device)
@@ -155,6 +163,29 @@ namespace BansheeEngine
 		mSources.erase(source);
 	}
 
+	void OAAudio::startStreaming(OAAudioSource* source, bool startPaused)
+	{
+		Lock(mMutex);
+
+		StreamingCommand command;
+		command.type = StreamingCommandType::Start;
+		command.source = source;
+		command.params[0] = startPaused;
+
+		mStreamingCommandQueue.push_back(command);
+	}
+
+	void OAAudio::stopStreaming(OAAudioSource* source)
+	{
+		Lock(mMutex);
+
+		StreamingCommand command;
+		command.type = StreamingCommandType::Stop;
+		command.source = source;
+
+		mStreamingCommandQueue.push_back(command);
+	}
+
 	ALCcontext* OAAudio::_getContext(const OAAudioListener* listener) const
 	{
 		if (mListeners.size() > 0)
@@ -228,6 +259,61 @@ namespace BansheeEngine
 		mContexts.clear();
 	}
 
+	void OAAudio::updateStreaming()
+	{
+		{
+			Lock(mMutex);
+
+			for(auto& command : mStreamingCommandQueue)
+			{
+				switch(command.type)
+				{
+				case StreamingCommandType::Start:
+				{
+					StreamingData data;
+					data.sourceIDs = command.source->mSourceIDs;
+					alGenBuffers(StreamingData::StreamBufferCount, data.streamBuffers);
+
+					mStreamingSources.insert(std::make_pair(command.source, data));
+
+					for(auto& )
+
+					// TODO - Store source IDs, start playback (possibly paused)
+				}
+					break;
+				case StreamingCommandType::Stop:
+					auto& contexts = gOAAudio()._getContexts(); // TODO - Don't use contexts here, use the source IDs directly
+					UINT32 numContexts = (UINT32)contexts.size();
+					for (UINT32 i = 0; i < numContexts; i++)
+					{
+						if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
+							alcMakeContextCurrent(contexts[i]);
+
+						INT32 numQueuedBuffers;
+						alGetSourcei(mSourceIDs[i], AL_BUFFERS_QUEUED, &numQueuedBuffers);
+
+						UINT32 buffer;
+						for (INT32 j = 0; j < numQueuedBuffers; j++)
+							alSourceUnqueueBuffers(mSourceIDs[i], 1, &buffer);
+					}
+
+					alDeleteBuffers(StreamBufferCount, mStreamBuffers);
+
+					command.source->destroyStreamBuffers();
+					mStreamingSources.erase(command.source);
+					break;
+				default:
+					break;
+				}
+			}
+
+			mStreamingCommandQueue.clear();
+		}
+
+		for (auto& source : mStreamingSources)
+			source->stream();
+	}
+
 	OAAudio& gOAAudio()
 	{
 		return static_cast<OAAudio&>(OAAudio::instance());

Source/BansheeOpenAudio/Source/BsOAAudioSource.cpp

@@ -3,12 +3,12 @@
 #include "BsOAAudioSource.h"
 #include "BsOAAudio.h"
 #include "BsOAAudioClip.h"
-#include "AL\al.h"
+#include "AL/al.h"
 
 namespace BansheeEngine
 {
 	OAAudioSource::OAAudioSource()
-		:mSeekPosition(0.0f), mStreamBuffers()
+		:mSeekPosition(0), mRequiresStreaming(false)
 	{
 		gOAAudio()._registerSource(this);
 		rebuild();
@@ -16,16 +16,17 @@ namespace BansheeEngine
 
 	OAAudioSource::~OAAudioSource()
 	{
+		clear();
 		gOAAudio()._unregisterSource(this);
-		destroyStreamBuffers();
 	}
 
 	void OAAudioSource::setClip(const HAudioClip& clip)
 	{
-		bool wasStreaming = isStreaming();
-
 		AudioSource::setClip(clip);
 
+		stop(); // TODO: I should wait until all buffers are unqueued before proceeding
+		mRequiresStreaming = isStreaming();
+
 		auto& contexts = gOAAudio()._getContexts();
 		UINT32 numContexts = (UINT32)contexts.size();
 		for (UINT32 i = 0; i < numContexts; i++)
@@ -35,7 +36,7 @@ namespace BansheeEngine
 
 			alSourcei(mSourceIDs[i], AL_SOURCE_RELATIVE, !is3D());
 
-			if (!isStreaming())
+			if (!mRequiresStreaming)
 			{
 				UINT32 oaBuffer = 0;
 				if (clip.isLoaded())
@@ -46,18 +47,7 @@ namespace BansheeEngine
 				
 				alSourcei(mSourceIDs[i], AL_BUFFER, oaBuffer);
 			}
-			else
-				alSourcei(mSourceIDs[i], AL_BUFFER, 0);
 		}
-
-		if(wasStreaming && !isStreaming())
-			destroyStreamBuffers();
-
-		if(!wasStreaming && isStreaming())
-			createStreamBuffers();
-
-		//if(isStreaming())
-		//	stream(); // TODO
 	}
 
 	void OAAudioSource::setPosition(const Vector3& position)
@@ -180,25 +170,38 @@ namespace BansheeEngine
 
 	void OAAudioSource::play()
 	{
+		AudioSourceState state = getState();
 		AudioSource::play();
 
-		auto& contexts = gOAAudio()._getContexts();
-		UINT32 numContexts = (UINT32)contexts.size();
-		for (UINT32 i = 0; i < numContexts; i++)
+		if (state == AudioSourceState::Playing)
+			return;
+		
+		if(!mRequiresStreaming || state == AudioSourceState::Paused)
 		{
-			if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
-				alcMakeContextCurrent(contexts[i]);
+			auto& contexts = gOAAudio()._getContexts();
+			UINT32 numContexts = (UINT32)contexts.size();
+			for (UINT32 i = 0; i < numContexts; i++)
+			{
+				if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
+					alcMakeContextCurrent(contexts[i]);
 
-			alSourcePlay(mSourceIDs[i]);
+				alSourcePlay(mSourceIDs[i]);
+			}
+		} 
+		else // Streaming and currently stopped
+		{
+			gOAAudio().startStreaming(this, false);
 		}
-
-		// TODO - Update streaming thread state
 	}
 
 	void OAAudioSource::pause()
 	{
+		AudioSourceState state = getState();
 		AudioSource::pause();
 
+		if (state == AudioSourceState::Paused)
+			return;
+
 		auto& contexts = gOAAudio()._getContexts();
 		UINT32 numContexts = (UINT32)contexts.size();
 		for (UINT32 i = 0; i < numContexts; i++)
@@ -208,14 +211,16 @@ namespace BansheeEngine
 
 			alSourcePause(mSourceIDs[i]);
 		}
-
-		// TODO - Update streaming thread state
 	}
 
 	void OAAudioSource::stop()
 	{
+		AudioSourceState state = getState();
 		AudioSource::stop();
 
+		if (state == AudioSourceState::Stopped)
+			return;
+
 		auto& contexts = gOAAudio()._getContexts();
 		UINT32 numContexts = (UINT32)contexts.size();
 		for (UINT32 i = 0; i < numContexts; i++)
@@ -226,37 +231,78 @@ namespace BansheeEngine
 			alSourceStop(mSourceIDs[i]);
 		}
 
-		// TODO - Update streaming thread state, seek to zero
+		if (mRequiresStreaming)
+			gOAAudio().stopStreaming(this);
 	}
 
 	void OAAudioSource::seek(float position)
 	{
-		auto& contexts = gOAAudio()._getContexts();
-		UINT32 numContexts = (UINT32)contexts.size();
-		for (UINT32 i = 0; i < numContexts; i++)
+		if (mAudioClip.isLoaded())
+			mSeekPosition = position * mAudioClip->getFrequency() * mAudioClip->getNumChannels();
+		else
+			mSeekPosition = 0;
+
+		if (!mRequiresStreaming)
 		{
-			if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
-				alcMakeContextCurrent(contexts[i]);
+			auto& contexts = gOAAudio()._getContexts();
+			UINT32 numContexts = (UINT32)contexts.size();
+			for (UINT32 i = 0; i < numContexts; i++)
+			{
+				if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
+					alcMakeContextCurrent(contexts[i]);
 
-			alSourcef(mSourceIDs[i], AL_SEC_OFFSET, position);
+				alSourcef(mSourceIDs[i], AL_SEC_OFFSET, position);
+			}
+		}
+		else
+		{
+			AudioSourceState state = getState();
+			if(state != AudioSourceState::Stopped)
+			{
+				gOAAudio().stopStreaming(this);
+				gOAAudio().startStreaming(this, state == AudioSourceState::Paused);
+			}
 		}
-
-		mSeekPosition = position;
-
-		// TODO - Update streaming thread position
 	}
 
 	void OAAudioSource::clear()
 	{
-		for (auto& source : mSourceIDs)
-			alSourcei(source, AL_BUFFER, 0);
+		if (mAudioClip.isLoaded())
+			mSeekPosition = tell() * mAudioClip->getFrequency() * mAudioClip->getNumChannels();
+		else
+			mSeekPosition = 0;
+
+		auto destroySources = [this]()
+		{
+			for (auto& source : mSourceIDs)
+				alSourcei(source, AL_BUFFER, 0);
+
+			alDeleteSources((UINT32)mSourceIDs.size(), mSourceIDs.data());
+			mSourceIDs.clear();
+		};
 
-		alDeleteSources((UINT32)mSourceIDs.size(), mSourceIDs.data());
-		mSourceIDs.clear();
+		if(mRequiresStreaming)
+		{
+			AudioSourceState state = getState();
+			if (state != AudioSourceState::Stopped)
+			{
+				gOAAudio().stopStreaming(this);
+				gOAAudio().queueCommand(destroySources);
+			}
+		}
+		else
+		{
+			// Still used by the streaming thread
+			if (gOAAudio().isStreaming(this)) // Assuming stopStreaming was already called
+				gOAAudio().queueCommand(destroySources);
+			else
+				destroySources();
+		}
 	}
 
 	void OAAudioSource::rebuild()
 	{
+		AudioSourceState state = getState();
 		auto& contexts = gOAAudio()._getContexts();
 
 		mSourceIDs.resize(contexts.size());
@@ -287,7 +333,7 @@ namespace BansheeEngine
 				alSource3f(mSourceIDs[i], AL_VELOCITY, 0.0f, 0.0f, 0.0f);
 			}
 
-			if (!isStreaming())
+			if (!mRequiresStreaming)
 			{
 				UINT32 oaBuffer = 0;
 				if (mAudioClip.isLoaded())
@@ -297,13 +343,27 @@ namespace BansheeEngine
 				}
 
 				alSourcei(mSourceIDs[i], AL_BUFFER, oaBuffer);
+
+				float offset = 0.0f;
+				if (mAudioClip.isLoaded())
+					offset = (float)mSeekPosition / mAudioClip->getFrequency() / mAudioClip->getNumChannels();
+
+				alSourcef(mSourceIDs[i], AL_SEC_OFFSET, offset);
 			}
 			else
-			{
-				alSourcei(mSourceIDs[i], AL_BUFFER, 0);
+				alSourcef(mSourceIDs[i], AL_SEC_OFFSET, 0.0f); // Offset handled by the streaming system
 
-				//applyStreamBuffers(); // TODO
-			}
+			if (state == AudioSourceState::Playing || state == AudioSourceState::Paused)
+				alSourcePlay(mSourceIDs[i]);
+			
+			if (state == AudioSourceState::Paused)
+				alSourcePause(mSourceIDs[i]);
+		}
+
+		if(mRequiresStreaming)
+		{
+			if (state != AudioSourceState::Stopped)
+				gOAAudio().startStreaming(this, state == AudioSourceState::Paused);
 		}
 	}
 
@@ -325,29 +385,4 @@ namespace BansheeEngine
 
 		return (readMode == AudioReadMode::Stream) || isCompressed;
 	}
-
-	void OAAudioSource::createStreamBuffers()
-	{
-		alGenBuffers(StreamBufferCount, mStreamBuffers);
-	}
-
-	void OAAudioSource::destroyStreamBuffers()
-	{
-		auto& contexts = gOAAudio()._getContexts();
-		UINT32 numContexts = (UINT32)contexts.size();
-		for (UINT32 i = 0; i < numContexts; i++)
-		{
-			if (contexts.size() > 1) // If only one context is available it is guaranteed it is always active, so we can avoid setting it
-				alcMakeContextCurrent(contexts[i]);
-
-			INT32 numQueuedBuffers;
-			alGetSourcei(mSourceIDs[i], AL_BUFFERS_QUEUED, &numQueuedBuffers);
-
-			UINT32 buffer;
-			for (INT32 j = 0; j < numQueuedBuffers; j++)
-				alSourceUnqueueBuffers(mSourceIDs[i], 1, &buffer);
-		}
-
-		alDeleteBuffers(StreamBufferCount, mStreamBuffers);
-	}
 }