feat: decoupled stream event from runner, added API endpoint

docs/modules/dev/partials/spec.json

@@ -215,6 +215,45 @@
         }
       }
     },
+    "/api/v1/runners/start": {
+      "post": {
+        "summary": "Start a runner in a given session",
+        "description": "",
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/wolf__api__RunnerStartRequest"
+              }
+            }
+          },
+          "description": "",
+          "required": true
+        },
+        "responses": {
+          "200": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/wolf__api__GenericSuccessResponse"
+                }
+              }
+            },
+            "description": ""
+          },
+          "500": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/wolf__api__GenericErrorResponse"
+                }
+              }
+            },
+            "description": ""
+          }
+        }
+      }
+    },
     "/api/v1/sessions/add": {
       "post": {
         "summary": "Create a new stream session",
@@ -347,6 +386,16 @@
           "hevc_gst_pipeline": {
             "type": "string"
           },
+          "icon_png_path": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ]
+          },
           "id": {
             "type": "string"
           },
@@ -559,6 +608,35 @@
           "success"
         ]
       },
+      "wolf__api__RunnerStartRequest": {
+        "type": "object",
+        "properties": {
+          "runner": {
+            "anyOf": [
+              {
+                "$ref": "#/components/schemas/wolf__config__AppCMD__tagged"
+              },
+              {
+                "$ref": "#/components/schemas/wolf__config__AppDocker__tagged"
+              },
+              {
+                "$ref": "#/components/schemas/wolf__config__AppChildSession__tagged"
+              }
+            ]
+          },
+          "session_id": {
+            "type": "string"
+          },
+          "stop_stream_when_over": {
+            "type": "boolean"
+          }
+        },
+        "required": [
+          "runner",
+          "session_id",
+          "stop_stream_when_over"
+        ]
+      },
       "wolf__api__StreamSessionListResponse": {
         "type": "object",
         "properties": {

src/moonlight-server/api/api.hpp

@@ -62,6 +62,12 @@ struct StreamSessionStopRequest {
   std::string session_id;
 };
 
+struct RunnerStartRequest {
+  bool stop_stream_when_over;
+  rfl::TaggedUnion<"type", wolf::config::AppCMD, wolf::config::AppDocker, wolf::config::AppChildSession> runner;
+  std::string session_id;
+};
+
 struct UnixSocket {
   boost::asio::local::stream_protocol::socket socket;
   bool is_alive = true;
@@ -93,6 +99,8 @@ class UnixSocketServer {
   void endpoint_StreamSessionPause(const HTTPRequest &req, std::shared_ptr<UnixSocket> socket);
   void endpoint_StreamSessionStop(const HTTPRequest &req, std::shared_ptr<UnixSocket> socket);
 
+  void endpoint_RunnerStart(const HTTPRequest &req, std::shared_ptr<UnixSocket> socket);
+
   void sse_broadcast(const std::string &payload);
   void sse_keepalive(const boost::system::error_code &e);
 

src/moonlight-server/api/endpoints.cpp

@@ -202,4 +202,28 @@ void UnixSocketServer::endpoint_StreamSessionStop(const HTTPRequest &req, std::s
   }
 }
 
+void UnixSocketServer::endpoint_RunnerStart(const wolf::api::HTTPRequest &req, std::shared_ptr<UnixSocket> socket) {
+  auto event = rfl::json::read<RunnerStartRequest>(req.body);
+  if (event) {
+    auto session = state::get_session_by_id(this->state_->app_state->running_sessions->load(),
+                                            std::stoul(event.value().session_id));
+    if (!session) {
+      logs::log(logs::warning, "[API] Invalid session_id: {}", event.value().session_id);
+      auto res = GenericErrorResponse{.error = "Invalid session_id"};
+      send_http(socket, 500, rfl::json::write(res));
+      return;
+    }
+
+    auto runner = state::get_runner(event.value().runner, this->state_->app_state->event_bus);
+    state_->app_state->event_bus->fire_event(immer::box<events::StartRunner>(
+        events::StartRunner{.stop_stream_when_over = event.value().stop_stream_when_over,
+                            .runner = runner,
+                            .stream_session = immer::box<events::StreamSession>(*session)}));
+  } else {
+    logs::log(logs::warning, "[API] Invalid event: {} - {}", req.body, event.error()->what());
+    auto res = GenericErrorResponse{.error = event.error()->what()};
+    send_http(socket, 500, rfl::json::write(res));
+  }
+}
+
 } // namespace wolf::api

src/moonlight-server/api/openapi.cpp

@@ -9,7 +9,7 @@ struct OpenAPIComponents {
 struct OpenAPISchema {
   std::string openapi = "3.1.0";
   rfl::Object<rfl::Generic> info = {};
-  rfl::Object<rfl::Generic> server = {};
+  std::vector<rfl::Object<rfl::Generic>> servers = {};
   rfl::Object<rfl::Object<rfl::Generic>> paths = {};
   OpenAPIComponents components = {};
 };
@@ -27,8 +27,10 @@ template <typename T> std::string HTTPServer<T>::openapi_schema() const {
   schema.info["description"] = "API for the Wolf server";
   schema.info["version"] = "0.1";
 
-  schema.server["url"] = "http://localhost/";
-  schema.server["description"] = "Unix socket server";
+  rfl::Object<rfl::Generic> server = {};
+  server["url"] = "http://localhost/";
+  server["description"] = "Local development server";
+  schema.servers.push_back(server);
 
   /**
    * Takes a json schema in string form and returns a valid OpenAPI json object for that schema.

src/moonlight-server/api/unix_socket_server.cpp

@@ -147,6 +147,16 @@ UnixSocketServer::UnixSocketServer(boost::asio::io_context &io_context,
           .handler = [this](auto req, auto socket) { endpoint_StreamSessionStop(req, socket); },
       });
 
+  state_->http.add(HTTPMethod::POST,
+                   "/api/v1/runners/start",
+                   {
+                       .summary = "Start a runner in a given session",
+                       .request_description = APIDescription{.json_schema = rfl::json::to_schema<RunnerStartRequest>()},
+                       .response_description = {{200, {.json_schema = rfl::json::to_schema<GenericSuccessResponse>()}},
+                                                {500, {.json_schema = rfl::json::to_schema<GenericErrorResponse>()}}},
+                       .handler = [this](auto req, auto socket) { endpoint_RunnerStart(req, socket); },
+                   });
+
   /**
    * OpenAPI schema
    */

src/moonlight-server/events/events.hpp

@@ -48,7 +48,7 @@ struct Runner {
 
   virtual void run(std::size_t session_id,
                    std::string_view app_state_folder,
-                   std::shared_ptr<devices_atom_queue> plugged_devices_queue,
+                   const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
                    const immer::array<std::string> &virtual_inputs,
                    const immer::array<std::pair<std::string, std::string>> &paths,
                    const immer::map<std::string, std::string> &env_variables,
@@ -161,6 +161,12 @@ struct RTPAudioPingEvent {
 
 struct StreamSession;
 
+struct StartRunner {
+  bool stop_stream_when_over = false;
+  std::shared_ptr<Runner> runner;
+  immer::box<StreamSession> stream_session;
+};
+
 using EventBusHandlers = dp::handler_registration<immer::box<PlugDeviceEvent>,
                                                   immer::box<PairSignal>,
                                                   immer::box<UnplugDeviceEvent>,
@@ -172,7 +178,8 @@ using EventBusHandlers = dp::handler_registration<immer::box<PlugDeviceEvent>,
                                                   immer::box<ResumeStreamEvent>,
                                                   immer::box<StopStreamEvent>,
                                                   immer::box<RTPVideoPingEvent>,
-                                                  immer::box<RTPAudioPingEvent>>;
+                                                  immer::box<RTPAudioPingEvent>,
+                                                  immer::box<StartRunner>>;
 using EventBusType = dp::event_bus<immer::box<PlugDeviceEvent>,
                                    immer::box<PairSignal>,
                                    immer::box<UnplugDeviceEvent>,
@@ -184,7 +191,8 @@ using EventBusType = dp::event_bus<immer::box<PlugDeviceEvent>,
                                    immer::box<ResumeStreamEvent>,
                                    immer::box<StopStreamEvent>,
                                    immer::box<RTPVideoPingEvent>,
-                                   immer::box<RTPAudioPingEvent>>;
+                                   immer::box<RTPAudioPingEvent>,
+                                   immer::box<StartRunner>>;
 using EventsVariant = std::variant<immer::box<PlugDeviceEvent>,
                                    immer::box<PairSignal>,
                                    immer::box<UnplugDeviceEvent>,
@@ -196,7 +204,8 @@ using EventsVariant = std::variant<immer::box<PlugDeviceEvent>,
                                    immer::box<ResumeStreamEvent>,
                                    immer::box<StopStreamEvent>,
                                    immer::box<RTPVideoPingEvent>,
-                                   immer::box<RTPAudioPingEvent>>;
+                                   immer::box<RTPAudioPingEvent>,
+                                   immer::box<StartRunner>>;
 
 /**
  * A StreamSession is created when a Moonlight user call `launch`
@@ -230,6 +239,9 @@ struct StreamSession {
   std::shared_ptr<immer::atom<virtual_display::wl_state_ptr>> wayland_display =
       std::make_shared<immer::atom<virtual_display::wl_state_ptr>>();
 
+  std::shared_ptr<immer::atom<std::shared_ptr<audio::VSink>>> audio_sink =
+      std::make_shared<immer::atom<std::shared_ptr<audio::VSink>>>();
+
   // virtual devices
   std::shared_ptr<std::optional<MouseTypes>> mouse = std::make_shared<std::optional<MouseTypes>>();
   std::shared_ptr<std::optional<KeyboardTypes>> keyboard = std::make_shared<std::optional<KeyboardTypes>>();

src/moonlight-server/events/reflectors.hpp

@@ -77,6 +77,20 @@ template <> struct Reflector<events::App> {
   }
 };
 
+template <> struct Reflector<events::StartRunner> {
+  struct ReflType {
+    bool stop_stream_when_over;
+    rfl::TaggedUnion<"type", AppCMD, AppDocker, AppChildSession> runner;
+    std::string session_id;
+  };
+
+  static ReflType from(const events::StartRunner &v) {
+    return {.stop_stream_when_over = v.stop_stream_when_over,
+            .runner = v.runner->serialize(),
+            .session_id = std::to_string(v.stream_session->session_id)};
+  }
+};
+
 template <> struct Reflector<events::StreamSession> {
   struct ReflType {
     std::string app_id;

src/moonlight-server/runners/child_session.hpp

@@ -19,7 +19,7 @@ class RunChildSession : public events::Runner {
 
   void run(std::size_t session_id,
            std::string_view app_state_folder,
-           std::shared_ptr<events::devices_atom_queue> plugged_devices_queue,
+           const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
            const immer::array<std::string> &virtual_inputs,
            const immer::array<std::pair<std::string, std::string>> &paths,
            const immer::map<std::string, std::string> &env_variables,

src/moonlight-server/runners/docker.cpp

@@ -15,7 +15,7 @@ void create_udev_hw_files(std::filesystem::path base_hw_db_path,
 
 void RunDocker::run(std::size_t session_id,
                     std::string_view app_state_folder,
-                    std::shared_ptr<events::devices_atom_queue> plugged_devices_queue,
+                    const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
                     const immer::array<std::string> &virtual_inputs,
                     const immer::array<std::pair<std::string, std::string>> &paths,
                     const immer::map<std::string, std::string> &env_variables,

src/moonlight-server/runners/docker.hpp

@@ -86,7 +86,7 @@ class RunDocker : public events::Runner {
 
   void run(std::size_t session_id,
            std::string_view app_state_folder,
-           std::shared_ptr<events::devices_atom_queue> plugged_devices_queue,
+           const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
            const immer::array<std::string> &virtual_inputs,
            const immer::array<std::pair<std::string, std::string>> &paths,
            const immer::map<std::string, std::string> &env_variables,

src/moonlight-server/runners/process.cpp

@@ -12,7 +12,7 @@ using namespace wolf::core::events;
 
 void RunProcess::run(std::size_t session_id,
                      std::string_view app_state_folder,
-                     std::shared_ptr<events::devices_atom_queue> plugged_devices_queue,
+                     const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
                      const immer::array<std::string> &virtual_inputs,
                      const immer::array<std::pair<std::string, std::string>> &paths,
                      const immer::map<std::string, std::string> &env_variables,

src/moonlight-server/runners/process.hpp

@@ -19,7 +19,7 @@ class RunProcess : public events::Runner {
 
   void run(std::size_t session_id,
            std::string_view app_state_folder,
-           std::shared_ptr<events::devices_atom_queue> plugged_devices_queue,
+           const std::shared_ptr<events::devices_atom_queue> &plugged_devices_queue,
            const immer::array<std::string> &virtual_inputs,
            const immer::array<std::pair<std::string, std::string>> &paths,
            const immer::map<std::string, std::string> &env_variables,