video-shader-toys/apps/LoopThroughWithOpenGLCompositing/control/ControlServices.cpp

#include "ControlServices.h"

#include "ControlServer.h"
#include "OscServer.h"
#include "RuntimeControlBridge.h"
#include "RuntimeHost.h"
#include "RuntimeStore.h"
#include <windows.h>

ControlServices::ControlServices() :
	mControlServer(std::make_unique<ControlServer>()),
	mOscServer(std::make_unique<OscServer>()),
	mPollRunning(false),
	mRegistryChanged(false),
	mReloadRequested(false),
	mPollFailed(false)
{
}

ControlServices::~ControlServices()
{
	Stop();
}

bool ControlServices::Start(OpenGLComposite& composite, RuntimeHost& runtimeHost, std::string& error)
{
	Stop();

	if (!StartControlServicesBoundary(composite, runtimeHost, *this, *mControlServer, *mOscServer, error))
	{
		Stop();
		return false;
	}

	return true;
}

void ControlServices::BeginPolling(RuntimeHost& runtimeHost, RuntimeStore& runtimeStore)
{
	StartPolling(runtimeHost, runtimeStore);
}

void ControlServices::Stop()
{
	StopPolling();

	if (mOscServer)
		mOscServer->Stop();

	if (mControlServer)
		mControlServer->Stop();
}

void ControlServices::BroadcastState()
{
	if (mControlServer)
		mControlServer->BroadcastState();
}

void ControlServices::RequestBroadcastState()
{
	if (mControlServer)
		mControlServer->RequestBroadcastState();
}

bool ControlServices::QueueOscUpdate(const std::string& layerKey, const std::string& parameterKey, const std::string& valueJson, std::string& error)
{
	(void)error;

	PendingOscUpdate update;
	update.layerKey = layerKey;
	update.parameterKey = parameterKey;
	update.valueJson = valueJson;

	const std::string routeKey = layerKey + "\n" + parameterKey;
	{
		std::lock_guard<std::mutex> lock(mPendingOscMutex);
		mPendingOscUpdates[routeKey] = std::move(update);
	}
	return true;
}

bool ControlServices::ApplyPendingOscUpdates(std::vector<AppliedOscUpdate>& appliedUpdates, std::string& error)
{
	appliedUpdates.clear();

	std::map<std::string, PendingOscUpdate> pending;
	{
		std::lock_guard<std::mutex> lock(mPendingOscMutex);
		if (mPendingOscUpdates.empty())
			return true;
		pending.swap(mPendingOscUpdates);
	}

	for (const auto& entry : pending)
	{
		JsonValue targetValue;
		std::string parseError;
		if (!ParseJson(entry.second.valueJson, targetValue, parseError))
		{
			OutputDebugStringA(("OSC queued value parse failed: " + parseError + "\n").c_str());
			continue;
		}

		AppliedOscUpdate appliedUpdate;
		appliedUpdate.routeKey = entry.first;
		appliedUpdate.layerKey = entry.second.layerKey;
		appliedUpdate.parameterKey = entry.second.parameterKey;
		appliedUpdate.targetValue = targetValue;
		appliedUpdates.push_back(std::move(appliedUpdate));
	}

	(void)error;
	return true;
}

bool ControlServices::QueueOscCommit(const std::string& routeKey, const std::string& layerKey, const std::string& parameterKey, const JsonValue& value, uint64_t generation, std::string& error)
{
	(void)error;

	PendingOscCommit commit;
	commit.routeKey = routeKey;
	commit.layerKey = layerKey;
	commit.parameterKey = parameterKey;
	commit.value = value;
	commit.generation = generation;

	{
		std::lock_guard<std::mutex> lock(mPendingOscCommitMutex);
		mPendingOscCommits[routeKey] = std::move(commit);
	}
	return true;
}

void ControlServices::ClearOscState()
{
	{
		std::lock_guard<std::mutex> lock(mPendingOscMutex);
		mPendingOscUpdates.clear();
	}
	{
		std::lock_guard<std::mutex> lock(mPendingOscCommitMutex);
		mPendingOscCommits.clear();
	}
	{
		std::lock_guard<std::mutex> lock(mCompletedOscCommitMutex);
		mCompletedOscCommits.clear();
	}
}

void ControlServices::ConsumeCompletedOscCommits(std::vector<CompletedOscCommit>& completedCommits)
{
	completedCommits.clear();

	std::lock_guard<std::mutex> lock(mCompletedOscCommitMutex);
	if (mCompletedOscCommits.empty())
		return;

	completedCommits.swap(mCompletedOscCommits);
}

RuntimePollEvents ControlServices::ConsumePollEvents()
{
	RuntimePollEvents events;
	events.registryChanged = mRegistryChanged.exchange(false);
	events.reloadRequested = mReloadRequested.exchange(false);
	events.failed = mPollFailed.exchange(false);

	if (events.failed)
	{
		std::lock_guard<std::mutex> lock(mPollErrorMutex);
		events.error = mPollError;
	}

	return events;
}

void ControlServices::StartPolling(RuntimeHost& runtimeHost, RuntimeStore& runtimeStore)
{
	if (mPollRunning.exchange(true))
		return;

	mPollThread = std::thread([this, &runtimeHost, &runtimeStore]() { PollLoop(runtimeHost, runtimeStore); });
}

void ControlServices::StopPolling()
{
	if (!mPollRunning.exchange(false))
		return;

	if (mPollThread.joinable())
		mPollThread.join();
}

void ControlServices::PollLoop(RuntimeHost& runtimeHost, RuntimeStore& runtimeStore)
{
	while (mPollRunning)
	{
		std::map<std::string, PendingOscCommit> pendingCommits;
		{
			std::lock_guard<std::mutex> lock(mPendingOscCommitMutex);
			pendingCommits.swap(mPendingOscCommits);
		}
		for (const auto& entry : pendingCommits)
		{
			std::string commitError;
			if (runtimeStore.SetStoredParameterValueByControlKey(
				entry.second.layerKey,
				entry.second.parameterKey,
				entry.second.value,
				false,
				commitError))
			{
				CompletedOscCommit completedCommit;
				completedCommit.routeKey = entry.second.routeKey;
				completedCommit.generation = entry.second.generation;
				std::lock_guard<std::mutex> lock(mCompletedOscCommitMutex);
				mCompletedOscCommits.push_back(std::move(completedCommit));
			}
			else if (!commitError.empty())
			{
				OutputDebugStringA(("OSC commit failed: " + commitError + "\n").c_str());
			}
		}

		bool registryChanged = false;
		bool reloadRequested = false;
		std::string runtimeError;
		if (!runtimeHost.PollFileChanges(registryChanged, reloadRequested, runtimeError))
		{
			{
				std::lock_guard<std::mutex> lock(mPollErrorMutex);
				mPollError = runtimeError;
			}
			mPollFailed = true;
		}
		else
		{
			if (registryChanged)
				mRegistryChanged = true;
			if (reloadRequested)
				mReloadRequested = true;
		}

		for (int i = 0; i < 25 && mPollRunning; ++i)
			Sleep(10);
	}
}