fix: co-op, better audio handling

docs/modules/user/pages/configuration.adoc

@@ -283,9 +283,10 @@ In order to enable this you'll have to add the following app entry in the `confi
 title = "CO-OP session"
 # Get the parent session ID from the Wolf logs
 runner = { type = "child_session", parent_session_id = "4135727842959053255" }
-video = { source = "interpipesrc listen-to=4135727842959053255 is-live=true stream-sync=restart-ts" }
-audio = { source = "pulsesrc device=\"virtual_sink_4135727842959053255.monitor\" server=\"{server_name}\"" }
+video = { source = "interpipesrc listen-to=4135727842959053255_video is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false" }
+audio = { source = "interpipesrc listen-to=4135727842959053255_audio is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false" }
 start_virtual_compositor = false
+start_audio_server = false
 
 ....
 

src/moonlight-server/events/events.hpp

@@ -69,6 +69,7 @@ struct App {
 
   std::string opus_gst_pipeline;
   bool start_virtual_compositor;
+  bool start_audio_server;
   std::shared_ptr<Runner> runner;
   moonlight::control::pkts::CONTROLLER_TYPE joypad_type;
 };

src/moonlight-server/events/reflectors.hpp

@@ -41,6 +41,7 @@ template <> struct Reflector<events::App> {
 
     std::string opus_gst_pipeline;
     bool start_virtual_compositor;
+    bool start_audio_server;
     rfl::TaggedUnion<"type", AppCMD, AppDocker, AppChildSession> runner;
     ControllerType joypad_type;
   };
@@ -72,6 +73,7 @@ template <> struct Reflector<events::App> {
             .render_node = v.render_node,
             .opus_gst_pipeline = v.opus_gst_pipeline,
             .start_virtual_compositor = v.start_virtual_compositor,
+            .start_audio_server = v.start_audio_server,
             .runner = v.runner->serialize(),
             .joypad_type = ctrl_type};
   }

src/moonlight-server/state/configTOML.cpp

@@ -130,11 +130,16 @@ Config load_or_default(const std::string &source,
   auto default_gst_video_settings = cfg.gstreamer.video;
   if (default_gst_video_settings.default_source.find("appsrc") != std::string::npos) {
     logs::log(logs::debug, "Found appsrc in default_source, migrating to interpipesrc");
-    default_gst_video_settings.default_source =
-        "interpipesrc listen-to={session_id} is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=1 block=false";
+    default_gst_video_settings.default_source = "interpipesrc listen-to={session_id}_video is-live=true "
+                                                "stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false";
   }
 
   auto default_gst_audio_settings = cfg.gstreamer.audio;
+  if (default_gst_audio_settings.default_source.find("appsrc") != std::string::npos) {
+    logs::log(logs::debug, "Found pulsesrc in default_source, migrating to interpipesrc");
+    default_gst_audio_settings.default_source = "interpipesrc listen-to={session_id}_audio is-live=true "
+                                                "stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false";
+  }
   auto default_gst_encoder_settings = default_gst_video_settings.defaults;
 
   auto default_app_render_node = utils::get_env("WOLF_RENDER_NODE", "/dev/dri/renderD128");
@@ -235,6 +240,7 @@ Config load_or_default(const std::string &source,
 
                         .opus_gst_pipeline = opus_gst_pipeline,
                         .start_virtual_compositor = app.start_virtual_compositor.value_or(true),
+                        .start_audio_server = app.start_audio_server.value_or(true),
                         .runner = get_runner(app.runner, ev_bus, running_sessions),
                         .joypad_type = get_controller_type(app.joypad_type.value_or(ControllerType::AUTO))}};
       }) |                                                  //

src/moonlight-server/state/default/config.v4.toml

@@ -149,6 +149,7 @@ base_create_json = """
 [[apps]]
 title = "Test ball"
 start_virtual_compositor = false
+start_audio_server = false
 
 [apps.runner]
 type = "process"
@@ -170,7 +171,7 @@ source = "audiotestsrc wave=ticks is-live=true"
 
 [gstreamer.video]
 
-default_source = "interpipesrc listen-to={session_id} is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=1 block=false"
+default_source = "interpipesrc listen-to={session_id}_video is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false"
 default_sink = """
 rtpmoonlightpay_video name=moonlight_pay \
 payload_size={payload_size} fec_percentage={fec_percentage} min_required_fec_packets={min_required_fec_packets} !
@@ -374,10 +375,10 @@ video/x-av1, stream-format=obu-stream, alignment=frame, profile=main\
 ###
 [gstreamer.audio]
 default_source = """
-pulsesrc device="{sink_name}" server="{server_name}"\
+interpipesrc listen-to={session_id}_audio is-live=true stream-sync=restart-ts max-bytes=0 max-buffers=3 block=false\
 """
 
-default_audio_params = "audio/x-raw, channels={channels}, rate=48000"
+default_audio_params = "queue max-size-buffers=3 leaky=downstream ! audiorate ! audioconvert"
 
 default_opus_encoder = """
 opusenc bitrate={bitrate} bitrate-type=cbr frame-size={packet_duration} bandwidth=fullband \

src/moonlight-server/state/serialised_config.hpp

@@ -96,6 +96,7 @@ struct BaseApp {
   std::optional<BaseAppAudioOverride> audio;
   std::optional<ControllerType> joypad_type;
   std::optional<bool> start_virtual_compositor;
+  std::optional<bool> start_audio_server;
   rfl::TaggedUnion<"type", AppCMD, AppDocker, AppChildSession> runner;
 };
 

src/moonlight-server/streaming/streaming.cpp

@@ -81,11 +81,11 @@ void start_video_producer(std::size_t session_id,
                           const wolf::core::virtual_display::DisplayMode &display_mode,
                           const std::shared_ptr<events::EventBusType> &event_bus) {
   auto appsrc_state = streaming::custom_src::setup_app_src(display_mode, std::move(wl_state));
-  auto pipeline = fmt::format("appsrc is-live=true name=wolf_wayland_source ! "                        //
-                              "queue ! "                                                               //
-                              "interpipesink name={} sync=true async=false max-bytes=0 max-buffers=3", //
+  auto pipeline = fmt::format("appsrc is-live=true name=wolf_wayland_source ! "                              //
+                              "queue ! "                                                                     //
+                              "interpipesink name={}_video sync=true async=false max-bytes=0 max-buffers=3", //
                               session_id);
-  logs::log(logs::debug, "Starting pipeline: {}", pipeline);
+  logs::log(logs::debug, "[GSTREAMER] Starting video producer: {}", pipeline);
   run_pipeline(pipeline, [=](auto pipeline, auto loop) {
     if (auto app_src_el = gst_bin_get_by_name(GST_BIN(pipeline.get()), "wolf_wayland_source")) {
       appsrc_state->context = g_main_context_get_thread_default();
@@ -115,7 +115,36 @@ void start_video_producer(std::size_t session_id,
     auto stop_handler = event_bus->register_handler<immer::box<events::StopStreamEvent>>(
         [session_id, loop](const immer::box<events::StopStreamEvent> &ev) {
           if (ev->session_id == session_id) {
-            logs::log(logs::debug, "[GSTREAMER] Stopping pipeline: {}", session_id);
+            logs::log(logs::debug, "[GSTREAMER] Stopping video producer: {}", session_id);
+            g_main_loop_quit(loop.get());
+          }
+        });
+
+    return immer::array<immer::box<events::EventBusHandlers>>{std::move(stop_handler)};
+  });
+}
+
+void start_audio_producer(std::size_t session_id,
+                          const std::shared_ptr<events::EventBusType> &event_bus,
+                          int channel_count,
+                          const std::string &sink_name,
+                          const std::string &server_name) {
+  auto pipeline = fmt::format(
+      "pulsesrc device=\"{sink_name}\" server=\"{server_name}\" ! " //
+      "audio/x-raw, channels={channels}, rate=48000 ! "             //
+      "queue ! "                                                    //
+      "interpipesink name=\"{session_id}_audio\" sync=true async=false max-bytes=0 max-buffers=3",
+      fmt::arg("session_id", session_id),
+      fmt::arg("channels", channel_count),
+      fmt::arg("sink_name", sink_name),
+      fmt::arg("server_name", server_name));
+  logs::log(logs::debug, "[GSTREAMER] Starting audio producer: {}", pipeline);
+
+  run_pipeline(pipeline, [=](auto pipeline, auto loop) {
+    auto stop_handler = event_bus->register_handler<immer::box<events::StopStreamEvent>>(
+        [session_id, loop](const immer::box<events::StopStreamEvent> &ev) {
+          if (ev->session_id == session_id) {
+            logs::log(logs::debug, "[GSTREAMER] Stopping audio producer: {}", session_id);
             g_main_loop_quit(loop.get());
           }
         });
@@ -224,6 +253,7 @@ void start_streaming_audio(const immer::box<events::AudioSession> &audio_session
                            const std::string &server_name) {
   auto pipeline = fmt::format(
       fmt::runtime(audio_session->gst_pipeline),
+      fmt::arg("session_id", audio_session->session_id),
       fmt::arg("channels", audio_session->audio_mode.channels),
       fmt::arg("bitrate", audio_session->audio_mode.bitrate),
       // TODO: opusenc hardcodes those two

src/moonlight-server/streaming/streaming.hpp

@@ -24,6 +24,12 @@ void start_video_producer(std::size_t session_id,
                           const wolf::core::virtual_display::DisplayMode &display_mode,
                           const std::shared_ptr<events::EventBusType> &event_bus);
 
+void start_audio_producer(std::size_t session_id,
+                          const std::shared_ptr<events::EventBusType> &event_bus,
+                          int channel_count,
+                          const std::string &sink_name,
+                          const std::string &server_name);
+
 void start_streaming_video(const immer::box<events::VideoSession> &video_session,
                            const std::shared_ptr<events::EventBusType> &event_bus,
                            unsigned short client_port);

src/moonlight-server/wolf.cpp

@@ -239,12 +239,21 @@ auto setup_sessions_handlers(const immer::box<state::AppState> &app_state,
         logs::log(logs::debug, "[STREAM_SESSION] Create virtual audio sink");
         auto pulse_sink_name = fmt::format("virtual_sink_{}", session->session_id);
         std::shared_ptr<audio::VSink> v_device;
-        if (audio_server && audio_server->server) {
+        if (session->app->start_audio_server && audio_server && audio_server->server) {
           v_device = audio::create_virtual_sink(
               audio_server->server,
               audio::AudioDevice{.sink_name = pulse_sink_name,
                                  .mode = state::get_audio_mode(session->audio_channel_count, true)});
           session->audio_sink->store(v_device);
+
+          std::thread([session, audio_server = audio_server->server]() {
+            auto sink_name = fmt::format("virtual_sink_{}.monitor", session->session_id);
+            streaming::start_audio_producer(session->session_id,
+                                            session->event_bus,
+                                            session->audio_channel_count,
+                                            sink_name,
+                                            audio::get_server_name(audio_server));
+          }).detach();
         }
 
         session->event_bus->fire_event(immer::box<events::StartRunner>(
@@ -337,7 +346,7 @@ auto setup_sessions_handlers(const immer::box<state::AppState> &app_state,
           if (run_session->stop_stream_when_over) {
             /* App exited, cleanup */
             logs::log(logs::debug, "[STREAM_SESSION] Remove virtual audio sink");
-            if (session->audio_sink) {
+            if (session->app->start_audio_server) {
               audio::delete_virtual_sink(audio_server->server, session->audio_sink->load());
             }