From 55e3a04b8bc0aac89c0fea3b10458b38210d2e86 Mon Sep 17 00:00:00 2001
From: Julian Waller <git@julusian.co.uk>
Date: Tue, 19 Mar 2024 16:00:09 +0000
Subject: [PATCH] feat: add `CALLBG` to perform CALL on the background producer
 #1525

---
 src/core/producer/stage.cpp            | 12 ++++++++++++
 src/core/producer/stage.h              |  3 +++
 src/protocol/amcp/AMCPCommandsImpl.cpp | 24 ++++++++++++++++++++++++
 3 files changed, 39 insertions(+)

diff --git a/src/core/producer/stage.cpp b/src/core/producer/stage.cpp
index f93c199a05..6b09f8af74 100644
--- a/src/core/producer/stage.cpp
+++ b/src/core/producer/stage.cpp
@@ -384,6 +384,10 @@ struct stage::impl : public std::enable_shared_from_this<impl>
     {
         return flatten(executor_.begin_invoke([=] { return get_layer(index).foreground()->call(params).share(); }));
     }
+    std::future<std::wstring> callbg(int index, const std::vector<std::wstring>& params)
+    {
+        return flatten(executor_.begin_invoke([=] { return get_layer(index).background()->call(params).share(); }));
+    }
 
     std::unique_lock<std::mutex> get_lock() { return std::move(std::unique_lock<std::mutex>(lock_)); }
 
@@ -414,6 +418,10 @@ std::future<std::wstring> stage::call(int index, const std::vector<std::wstring>
 {
     return impl_->call(index, params);
 }
+std::future<std::wstring> stage::callbg(int index, const std::vector<std::wstring>& params)
+{
+    return impl_->callbg(index, params);
+}
 std::future<void> stage::apply_transforms(const std::vector<stage::transform_tuple_t>& transforms)
 {
     return impl_->apply_transforms(transforms);
@@ -488,6 +496,10 @@ std::future<std::wstring> stage_delayed::call(int index, const std::vector<std::
 {
     return executor_.begin_invoke([=]() -> std::wstring { return stage_->call(index, params).get(); });
 }
+std::future<std::wstring> stage_delayed::callbg(int index, const std::vector<std::wstring>& params)
+{
+    return executor_.begin_invoke([=]() -> std::wstring { return stage_->callbg(index, params).get(); });
+}
 std::future<void> stage_delayed::apply_transforms(const std::vector<stage_delayed::transform_tuple_t>& transforms)
 {
     return executor_.begin_invoke([=]() { return stage_->apply_transforms(transforms).get(); });
diff --git a/src/core/producer/stage.h b/src/core/producer/stage.h
index 593ccadcd5..f83a4f5dc7 100644
--- a/src/core/producer/stage.h
+++ b/src/core/producer/stage.h
@@ -87,6 +87,7 @@ class stage_base
     virtual std::future<void>         play(int index)                                                              = 0;
     virtual std::future<void>         stop(int index)                                                              = 0;
     virtual std::future<std::wstring> call(int index, const std::vector<std::wstring>& params)                     = 0;
+    virtual std::future<std::wstring> callbg(int index, const std::vector<std::wstring>& params)                   = 0;
     virtual std::future<void>         clear(int index)                                                             = 0;
     virtual std::future<void>         clear()                                                                      = 0;
     virtual std::future<void>         swap_layers(const std::shared_ptr<stage_base>& other, bool swap_transforms)  = 0;
@@ -136,6 +137,7 @@ class stage final : public stage_base
     std::future<void>            play(int index) override;
     std::future<void>            stop(int index) override;
     std::future<std::wstring>    call(int index, const std::vector<std::wstring>& params) override;
+    std::future<std::wstring>    callbg(int index, const std::vector<std::wstring>& params) override;
     std::future<void>            clear(int index) override;
     std::future<void>            clear() override;
     std::future<void>            swap_layers(const std::shared_ptr<stage_base>& other, bool swap_transforms) override;
@@ -191,6 +193,7 @@ class stage_delayed final : public stage_base
     std::future<void>            play(int index) override;
     std::future<void>            stop(int index) override;
     std::future<std::wstring>    call(int index, const std::vector<std::wstring>& params) override;
+    std::future<std::wstring>    callbg(int index, const std::vector<std::wstring>& params) override;
     std::future<void>            clear(int index) override;
     std::future<void>            clear() override;
     std::future<void>            swap_layers(const std::shared_ptr<stage_base>& other, bool swap_transforms) override;
diff --git a/src/protocol/amcp/AMCPCommandsImpl.cpp b/src/protocol/amcp/AMCPCommandsImpl.cpp
index 403bbe4d15..0a8c812eca 100644
--- a/src/protocol/amcp/AMCPCommandsImpl.cpp
+++ b/src/protocol/amcp/AMCPCommandsImpl.cpp
@@ -403,6 +403,29 @@ std::wstring clear_all_command(command_context& ctx)
     return L"202 CLEAR ALL OK\r\n";
 }
 
+std::future<std::wstring> callbg_command(command_context& ctx)
+{
+    const auto result = ctx.channel.stage->callbg(ctx.layer_index(), ctx.parameters).share();
+
+    // TODO: because of std::async deferred timed waiting does not work
+
+    /*auto wait_res = result.wait_for(std::chrono::seconds(2));
+    if (wait_res == std::future_status::timeout)
+    CASPAR_THROW_EXCEPTION(timed_out());*/
+
+    return std::async(std::launch::deferred, [result]() -> std::wstring {
+        std::wstring res = result.get();
+
+        std::wstringstream replyString;
+        if (res.empty())
+            replyString << L"202 CALLBG OK\r\n";
+        else
+            replyString << L"201 CALLBG OK\r\n" << res << L"\r\n";
+
+        return replyString.str();
+    });
+}
+
 std::future<std::wstring> call_command(command_context& ctx)
 {
     const auto result = ctx.channel.stage->call(ctx.layer_index(), ctx.parameters).share();
@@ -1674,6 +1697,7 @@ std::wstring osc_unsubscribe_command(command_context& ctx)
 void register_commands(std::shared_ptr<amcp_command_repository_wrapper>& repo)
 {
     repo->register_channel_command(L"Basic Commands", L"LOADBG", loadbg_command, 1);
+    repo->register_channel_command(L"Basic Commands", L"CALLBG", callbg_command, 1);
     repo->register_channel_command(L"Basic Commands", L"LOAD", load_command, 0);
     repo->register_channel_command(L"Basic Commands", L"PLAY", play_command, 0);
     repo->register_channel_command(L"Basic Commands", L"PAUSE", pause_command, 0);