#include "ControlServices.h" #include "ControlServer.h" #include "OscServer.h" #include "RuntimeControlBridge.h" #include "RuntimeStore.h" #include ControlServices::ControlServices() : mControlServer(std::make_unique()), mOscServer(std::make_unique()), mPollRunning(false) { } ControlServices::~ControlServices() { Stop(); } bool ControlServices::Start(OpenGLComposite& composite, RuntimeStore& runtimeStore, std::string& error) { Stop(); if (!StartControlServicesBoundary(composite, runtimeStore, *this, *mControlServer, *mOscServer, error)) { Stop(); return false; } return true; } void ControlServices::BeginPolling(RuntimeCoordinator& runtimeCoordinator) { StartPolling(runtimeCoordinator); } 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 lock(mPendingOscMutex); mPendingOscUpdates[routeKey] = std::move(update); } return true; } bool ControlServices::ApplyPendingOscUpdates(std::vector& appliedUpdates, std::string& error) { appliedUpdates.clear(); std::map pending; { std::lock_guard 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 lock(mPendingOscCommitMutex); mPendingOscCommits[routeKey] = std::move(commit); } return true; } void ControlServices::ClearOscState() { { std::lock_guard lock(mPendingOscMutex); mPendingOscUpdates.clear(); } { std::lock_guard lock(mPendingOscCommitMutex); mPendingOscCommits.clear(); } { std::lock_guard lock(mCompletedOscCommitMutex); mCompletedOscCommits.clear(); } } void ControlServices::ConsumeCompletedOscCommits(std::vector& completedCommits) { completedCommits.clear(); std::lock_guard lock(mCompletedOscCommitMutex); if (mCompletedOscCommits.empty()) return; completedCommits.swap(mCompletedOscCommits); } void ControlServices::ConsumeRuntimeCoordinatorResults(std::vector& results) { results.clear(); std::lock_guard lock(mRuntimeCoordinatorResultMutex); if (mRuntimeCoordinatorResults.empty()) return; results.swap(mRuntimeCoordinatorResults); } void ControlServices::StartPolling(RuntimeCoordinator& runtimeCoordinator) { if (mPollRunning.exchange(true)) return; mPollThread = std::thread([this, &runtimeCoordinator]() { PollLoop(runtimeCoordinator); }); } void ControlServices::StopPolling() { if (!mPollRunning.exchange(false)) return; if (mPollThread.joinable()) mPollThread.join(); } void ControlServices::PollLoop(RuntimeCoordinator& runtimeCoordinator) { while (mPollRunning) { std::map pendingCommits; { std::lock_guard lock(mPendingOscCommitMutex); pendingCommits.swap(mPendingOscCommits); } for (const auto& entry : pendingCommits) { const RuntimeCoordinatorResult result = runtimeCoordinator.CommitOscParameterByControlKey( entry.second.layerKey, entry.second.parameterKey, entry.second.value); if (result.accepted) { CompletedOscCommit completedCommit; completedCommit.routeKey = entry.second.routeKey; completedCommit.generation = entry.second.generation; std::lock_guard lock(mCompletedOscCommitMutex); mCompletedOscCommits.push_back(std::move(completedCommit)); QueueRuntimeCoordinatorResult(result); } else if (!result.errorMessage.empty()) { OutputDebugStringA(("OSC commit failed: " + result.errorMessage + "\n").c_str()); } } bool registryChanged = false; const RuntimeCoordinatorResult pollResult = runtimeCoordinator.PollRuntimeStoreChanges(registryChanged); if (pollResult.runtimeStateBroadcastRequired || pollResult.shaderBuildRequested || pollResult.compileStatusChanged) QueueRuntimeCoordinatorResult(pollResult, pollResult.compileStatusChanged && !pollResult.compileStatusSucceeded && !pollResult.compileStatusMessage.empty()); for (int i = 0; i < 25 && mPollRunning; ++i) Sleep(10); } } void ControlServices::QueueRuntimeCoordinatorResult(const RuntimeCoordinatorResult& result, bool failed) { RuntimeCoordinatorServiceResult serviceResult; serviceResult.result = result; serviceResult.failed = failed; std::lock_guard lock(mRuntimeCoordinatorResultMutex); mRuntimeCoordinatorResults.push_back(std::move(serviceResult)); }