From f325d3ee0ba7443c86abd12728a019a47e13bee9 Mon Sep 17 00:00:00 2001
From: Dan Isla <dan.isla@gmail.com>
Date: Fri, 22 Sep 2023 03:24:05 +0000
Subject: [PATCH] Video and audio in separate webrtcbins

- Reduces latency when streaming with audio enabled.
- AV sync is within reason.
- Client connects separate RTCPeerConnections for audio and video.
- Connections use the same config and STUN/TURN servers.
- Backend signal clients and pipelines are also duplicated.
- Removed option to disable audio from client and cli args.
---
 addons/gst-web/src/app.js                  | 59 +++++++--------
 addons/gst-web/src/index.html              | 10 ++-
 addons/gst-web/src/signalling.js           | 11 +--
 addons/gst-web/src/webrtc.js               | 11 ++-
 src/selkies_gstreamer/__main__.py          | 87 ++++++++++++++--------
 src/selkies_gstreamer/gstwebrtc_app.py     | 63 ++++++----------
 src/selkies_gstreamer/webrtc_input.py      |  7 --
 src/selkies_gstreamer/webrtc_signalling.py |  4 +-
 8 files changed, 130 insertions(+), 122 deletions(-)

diff --git a/addons/gst-web/src/app.js b/addons/gst-web/src/app.js
index f59db04a..72897eb7 100644
--- a/addons/gst-web/src/app.js
+++ b/addons/gst-web/src/app.js
@@ -76,7 +76,6 @@ var app = new Vue({
                 { text: '60 fps', value: 60 },
                 { text: '100 fps', value: 100 },
             ],
-            audioEnabled: false,
             audioBitRate: 32000,
             audioBitRateOptions: [
                 { text: '32 kb/s', value: 32000 },
@@ -242,12 +241,6 @@ var app = new Vue({
             webrtc.sendDataChannelMessage('_arg_fps,' + newValue);
             this.setIntParam("videoFramerate", newValue);
         },
-        audioEnabled(newValue, oldValue) {
-            if (newValue === null) return;
-            console.log("audio enabled changed from " + oldValue + " to " + newValue);
-            if (oldValue !== null && newValue !== oldValue) webrtc.sendDataChannelMessage('_arg_audio,' + newValue);
-            this.setBoolParam("audioEnabled", newValue);
-        },
         resizeRemote(newValue, oldValue) {
             if (newValue === null) return;
             console.log("resize remote changed from " + oldValue + " to " + newValue);
@@ -326,11 +319,18 @@ if (videoElement === null) {
     throw 'videoElement not found on page';
 }
 
+var audioElement = document.getElementById("audio_stream");
+if (audioElement === null) {
+    throw 'audioElement not found on page';
+}
+
 // WebRTC entrypoint, connect to the signalling server
 /*global WebRTCDemoSignalling, WebRTCDemo*/
 var protocol = (location.protocol == "http:" ? "ws://" : "wss://");
-var signalling = new WebRTCDemoSignalling(new URL(protocol + window.location.host + "/" + app.appName + "/signalling/"), 1);
-var webrtc = new WebRTCDemo(signalling, videoElement);
+var signalling = new WebRTCDemoSignalling(new URL(protocol + window.location.host + "/" + app.appName + "/signalling/"));
+var webrtc = new WebRTCDemo(signalling, videoElement, 1);
+var audio_signalling = new WebRTCDemoSignalling(new URL(protocol + window.location.host + "/" + app.appName + "/signalling/"));
+var audio_webrtc = new WebRTCDemo(audio_signalling, audioElement, 3);
 
 // Function to add timestamp to logs.
 var applyTimestamp = (msg) => {
@@ -351,15 +351,19 @@ signalling.ondisconnect = () => {
     app.status = 'connecting';
     videoElement.style.cursor = "auto";
     webrtc.reset();
+    audio_webrtc.reset();
 }
 
 // Send webrtc status and error messages to logs.
 webrtc.onstatus = (message) => { app.logEntries.push(applyTimestamp("[webrtc] " + message)) };
 webrtc.onerror = (message) => { app.logEntries.push(applyTimestamp("[webrtc] [ERROR] " + message)) };
+audio_webrtc.onstatus = (message) => { app.logEntries.push(applyTimestamp("[audio webrtc] " + message)) };
+audio_webrtc.onerror = (message) => { app.logEntries.push(applyTimestamp("[audio webrtc] [ERROR] " + message)) };
 
 if (app.debug) {
     signalling.ondebug = (message) => { app.debugEntries.push("[signalling] " + message); };
     webrtc.ondebug = (message) => { app.debugEntries.push(applyTimestamp("[webrtc] " + message)) };
+    audio_webrtc.ondebug = (message) => { app.debugEntries.push(applyTimestamp("[audio webrtc] " + message)) };
 }
 
 webrtc.ongpustats = (data) => {
@@ -409,19 +413,13 @@ webrtc.onconnectionstatechange = (state) => {
                 videoBytesReceivedStart = stats.video.bytesReceived;
 
                 // Audio stats.
-                if (app.audioEnabled) {
-                    app.connectionLatency += stats.audio.jitterBufferDelay * 1000;
-                    app.connectionPacketsReceived += stats.audio.packetsReceived;
-                    app.connectionPacketsLost += stats.audio.packetsLost;
-                    app.connectionAudioLatency = parseInt(stats.audio.jitterBufferDelay * 1000);
-                    app.connectionAudioCodecName = stats.audio.codecName;
-                    app.connectionAudioBitrate = (((stats.audio.bytesReceived - audioBytesReceivedStart) / (now - statsStart)) * 8 / 1e+3).toFixed(2);
-                    audioBytesReceivedStart = stats.audio.bytesReceived;
-                } else {
-                    app.connectionAudioBitrate = 0;
-                    app.connectionAudioCodecName = "NA";
-                    app.connectionAudioLatency = "NA";
-                }
+                app.connectionLatency += stats.audio.jitterBufferDelay * 1000;
+                app.connectionPacketsReceived += stats.audio.packetsReceived;
+                app.connectionPacketsLost += stats.audio.packetsLost;
+                app.connectionAudioLatency = parseInt(stats.audio.jitterBufferDelay * 1000);
+                app.connectionAudioCodecName = stats.audio.codecName;
+                app.connectionAudioBitrate = (((stats.audio.bytesReceived - audioBytesReceivedStart) / (now - statsStart)) * 8 / 1e+3).toFixed(2);
+                audioBytesReceivedStart = stats.audio.bytesReceived;
 
                 // Format latency
                 app.connectionLatency = parseInt(app.connectionLatency);
@@ -485,6 +483,10 @@ webrtc.onplayvideorequired = () => {
     app.showStart = true;
 }
 
+audio_webrtc.onplayvideorequired = () => {
+    app.showStart = true;
+}
+
 // Actions to take whenever window changes focus
 window.addEventListener('focus', () => {
     // reset keyboard to avoid stuck keys.
@@ -572,16 +574,6 @@ webrtc.onsystemaction = (action) => {
             // Use the server setting.
             app.audioBitRate = parseInt(action.split(",")[1]);
         }
-    } else if (action.startsWith('audio')) {
-        // Server received audio enabled setting.
-        const audioEnabledSetting = app.getBoolParam("audioEnabled" , null);
-        if (audioEnabledSetting !== null) {
-            // Prefer the user saved value.
-            app.audioEnabled = audioEnabledSetting;
-        } else {
-            // Use the server setting.
-            app.audioEnabled = (action.split(",")[1].toLowerCase() === 'true');
-        }
     } else if (action.startsWith('resize')) {
         // Remote resize enabled/disabled action.
         const resizeSetting = app.getBoolParam("resize", null);
@@ -687,6 +679,7 @@ fetch("/turn/")
     .then((config) => {
         // for debugging, force use of relay server.
         webrtc.forceTurn = app.turnSwitch;
+        audio_webrtc.forceTurn = app.turnSwitch;
 
         // get initial local resolution
         app.windowResolution = webrtc.input.getWindowResolution();
@@ -702,5 +695,7 @@ fetch("/turn/")
             app.debugEntries.push(applyTimestamp("[app] no TURN servers found."));
         }
         webrtc.rtcPeerConfig = config;
+        audio_webrtc.rtcPeerConfig = config;
         webrtc.connect();
+        audio_webrtc.connect();
     });
diff --git a/addons/gst-web/src/index.html b/addons/gst-web/src/index.html
index ba2ef2ce..a045438c 100644
--- a/addons/gst-web/src/index.html
+++ b/addons/gst-web/src/index.html
@@ -274,7 +274,7 @@
               </p>
               <p>
                 <v-select :items="audioBitRateOptions" label="Audio bit rate" menu-props="left" v-model="audioBitRate"
-                  :disabled="!audioEnabled" hint="Dynamic bit rate selection for audio encoder on server"
+                  hint="streaming bit rate selection for audio encoder on server"
                   persistent-hint>
                 </v-select>
               </p>
@@ -311,8 +311,6 @@
               <v-textarea bottom class="scrolly" label="Debug Logs" readonly :value="debugEntries.join('\n\n')">
               </v-textarea>
               <p>
-                <v-btn color="primary" v-if="!audioEnabled" small v-on:click="audioEnabled=true">Enable Audio</v-btn>
-                <v-btn color="primary" v-else small v-on:click="audioEnabled=false">Disable Audio</v-btn>
                 <v-switch v-model="resizeRemote" :label="`Resize remote to fit window: ${resizeRemote.toString()}`"></v-switch>
                 <v-switch v-model="scaleLocal" :label="`Scale to fit window: ${scaleLocal.toString()}`"></v-switch>
               </p>
@@ -355,6 +353,12 @@
         </v-container>
       </v-navigation-drawer>
 
+      <div id="audio_container" class="audio-container">
+        <audio id="audio_stream" class="audio" preload="none" playsinline>
+          Your browser doesn't support audio
+        </audio>
+      </div>
+
       <div id="video_container" class="video-container">
         <video id="stream" class="video" preload="none" playsinline>
           Your browser doesn't support video
diff --git a/addons/gst-web/src/signalling.js b/addons/gst-web/src/signalling.js
index f3d4e2b0..63b78854 100644
--- a/addons/gst-web/src/signalling.js
+++ b/addons/gst-web/src/signalling.js
@@ -43,11 +43,8 @@ class WebRTCDemoSignalling {
      *    The URL object of the signalling server to connect to, created with `new URL()`.
      *    Signalling implementation is here:
      *      https://github.com/GStreamer/gstreamer/tree/main/subprojects/gst-examples/webrtc/signalling
-     * @param {number} [peer_id]
-     *    The peer ID established during signalling that the sending peer (server) will connect to.
-     *    This can be anything, but must match what the server will attempt to connect to.
      */
-    constructor(server, peer_id) {
+    constructor(server) {
         /**
          * @private
          * @type {URL}
@@ -58,7 +55,7 @@ class WebRTCDemoSignalling {
          * @private
          * @type {number}
          */
-        this._peer_id = peer_id;
+        this.peer_id = 1;
 
         /**
          * @private
@@ -186,8 +183,8 @@ class WebRTCDemoSignalling {
             "scale": window.devicePixelRatio
         };
         this.state = 'connected';
-        this._ws_conn.send(`HELLO ${this._peer_id} ${btoa(JSON.stringify(meta))}`);
-        this._setStatus("Registering with server, peer ID: " + this._peer_id);
+        this._ws_conn.send(`HELLO ${this.peer_id} ${btoa(JSON.stringify(meta))}`);
+        this._setStatus("Registering with server, peer ID: " + this.peer_id);
         this.retry_count = 0;
     }
 
diff --git a/addons/gst-web/src/webrtc.js b/addons/gst-web/src/webrtc.js
index b0e3e677..29da461b 100644
--- a/addons/gst-web/src/webrtc.js
+++ b/addons/gst-web/src/webrtc.js
@@ -49,7 +49,7 @@ class WebRTCDemo {
      * @param {Element} [element]
      *    video element to attach stream to.
      */
-    constructor(signalling, element) {
+    constructor(signalling, element, peer_id) {
         /**
          * @type {WebRTCDemoSignalling}
          */
@@ -60,6 +60,11 @@ class WebRTCDemo {
          */
         this.element = element;
 
+        /**
+         * @type {Element}
+         */
+        this.peer_id = peer_id;
+
         /**
          * @type {boolean}
          */
@@ -313,7 +318,7 @@ class WebRTCDemo {
         this._setStatus("Received incoming " + event.track.kind + " stream from peer");
         if (!this.streams) this.streams = [];
         this.streams.push([event.track.kind, event.streams]);
-        if (event.track.kind === "video") {
+        if (event.track.kind === "video" || event.track.kind === "audio") {
             this.element.srcObject = event.streams[0];
             this.playVideo();
         }
@@ -667,6 +672,8 @@ class WebRTCDemo {
             config.iceTransportPolicy = "relay";
             this.peerConnection.setConfiguration(config);
         }
+
+        this.signalling.peer_id = this.peer_id;
         this.signalling.connect();
     }
 
diff --git a/src/selkies_gstreamer/__main__.py b/src/selkies_gstreamer/__main__.py
index f0ef6d88..04083af1 100644
--- a/src/selkies_gstreamer/__main__.py
+++ b/src/selkies_gstreamer/__main__.py
@@ -29,6 +29,7 @@
 import sys
 import time
 import urllib.parse
+import traceback
 
 from watchdog.observers import Observer
 from watchdog.events import FileClosedEvent, FileSystemEventHandler
@@ -395,9 +396,6 @@ def main():
     parser.add_argument('--video_bitrate',
                         default=os.environ.get('WEBRTC_VIDEO_BITRATE', '2000'),
                         help='Default video bitrate')
-    parser.add_argument('--enable_audio',
-                        default=os.environ.get('ENABLE_AUDIO', 'true'),
-                        help='Enable or disable audio stream')
     parser.add_argument('--audio_bitrate',
                         default=os.environ.get('WEBRTC_AUDIO_BITRATE', '64000'),
                         help='Default audio bitrate')
@@ -438,8 +436,6 @@ def main():
                     args.video_bitrate = int(v)
                 if k == "audio_bitrate":
                     args.audio_bitrate = int(v)
-                if k == "enable_audio":
-                    args.enable_audio = str((str(v).lower() == 'true')).lower()
                 if k == "enable_resize":
                     args.enable_resize = str((str(v).lower() == 'true')).lower()
                 if k == "encoder":
@@ -461,6 +457,8 @@ def main():
     # Peer id for this app, default is 0, expecting remote peer id to be 1
     my_id = 0
     peer_id = 1
+    my_audio_id = 2
+    audio_peer_id = 3
 
     # Initialize metrics server.
     metrics = Metrics(int(args.metrics_port))
@@ -475,6 +473,13 @@ def main():
         basic_auth_user=args.basic_auth_user,
         basic_auth_password=args.basic_auth_password)
 
+    # Initialize signalling client for audio connection.
+    audio_signalling = WebRTCSignalling('%s//127.0.0.1:%s/ws' % (ws_protocol, args.port), my_audio_id, audio_peer_id,
+        enable_https=using_https,
+        enable_basic_auth=using_basic_auth,
+        basic_auth_user=args.basic_auth_user,
+        basic_auth_password=args.basic_auth_password)
+
     # Handle errors from the signalling server.
     async def on_signalling_error(e):
        if isinstance(e, WebRTCSignallingErrorNoPeer):
@@ -484,12 +489,23 @@ async def on_signalling_error(e):
        else:
            logger.error("signalling error: %s", str(e))
            app.stop_pipeline()
+    async def on_audio_signalling_error(e):
+       if isinstance(e, WebRTCSignallingErrorNoPeer):
+           # Waiting for peer to connect, retry in 2 seconds.
+           time.sleep(2)
+           await audio_signalling.setup_call()
+       else:
+           logger.error("signalling error: %s", str(e))
+           audio_app.stop_pipeline()
     signalling.on_error = on_signalling_error
+    audio_signalling.on_error = on_audio_signalling_error
 
     signalling.on_disconnect = lambda: app.stop_pipeline()
+    audio_signalling.on_disconnect = lambda: audio_app.stop_pipeline()
 
     # After connecting, attempt to setup call to peer.
     signalling.on_connect = signalling.setup_call
+    audio_signalling.on_connect = audio_signalling.setup_call
 
     # [START main_setup]
     # Fetch the TURN server and credentials
@@ -533,7 +549,6 @@ async def on_signalling_error(e):
     logger.info("initial server RTC config: {}".format(rtc_config))
 
     # Extract arguments
-    enable_audio = args.enable_audio.lower() == "true"
     enable_resize = args.enable_resize.lower() == "true"
     audio_channels = int(args.audio_channels)
     curr_fps = int(args.framerate)
@@ -544,33 +559,47 @@ async def on_signalling_error(e):
     cursor_size = int(args.cursor_size)
 
     # Create instance of app
-    app = GSTWebRTCApp(stun_servers, turn_servers, enable_audio, audio_channels, curr_fps, args.encoder, curr_video_bitrate, curr_audio_bitrate)
+    app = GSTWebRTCApp(stun_servers, turn_servers, audio_channels, curr_fps, args.encoder, curr_video_bitrate, curr_audio_bitrate)
+    audio_app = GSTWebRTCApp(stun_servers, turn_servers, audio_channels, curr_fps, args.encoder, curr_video_bitrate, curr_audio_bitrate)
 
     # [END main_setup]
 
     # Send the local sdp to signalling when offer is generated.
     app.on_sdp = signalling.send_sdp
+    audio_app.on_sdp = audio_signalling.send_sdp
 
     # Send ICE candidates to the signalling server.
     app.on_ice = signalling.send_ice
+    audio_app.on_ice = audio_signalling.send_ice
 
     # Set the remote SDP when received from signalling server.
     signalling.on_sdp = app.set_sdp
+    audio_signalling.on_sdp = audio_app.set_sdp
 
     # Set ICE candidates received from signalling server.
     signalling.on_ice = app.set_ice
+    audio_signalling.on_ice = audio_app.set_ice
 
     # Start the pipeline once the session is established.
-    def on_session_handler(meta=None):
-        logger.info("starting session with meta: {}".format(meta))
-        if meta:
-            if enable_resize:
-                if meta["res"]:
-                    on_resize_handler(meta["res"])
-                if meta["scale"]:
-                    on_scaling_ratio_handler(meta["scale"])
-        app.start_pipeline()
+    def on_session_handler(session_peer_id, meta=None):
+        logger.info("starting session for peer id {} with meta: {}".format(session_peer_id, meta))
+        if str(session_peer_id) == str(peer_id):
+            if meta:
+                if enable_resize:
+                    if meta["res"]:
+                        on_resize_handler(meta["res"])
+                    if meta["scale"]:
+                        on_scaling_ratio_handler(meta["scale"])
+            logger.info("starting video pipeline")
+            app.start_pipeline()
+        elif str(session_peer_id) == str(audio_peer_id):
+            logger.info("starting audio pipeline")
+            audio_app.start_pipeline(audio_only=True)
+        else:
+            logger.error("failed to start pipeline for peer_id: %s" % peer_id)
+
     signalling.on_session = on_session_handler
+    audio_signalling.on_session = on_session_handler
 
     # Initialize the Xinput instance
     cursor_scale = 1.0
@@ -593,8 +622,7 @@ def data_channel_ready():
 
         app.send_framerate(app.framerate)
         app.send_video_bitrate(app.video_bitrate)
-        app.send_audio_bitrate(app.audio_bitrate)
-        app.send_audio_enabled(app.audio)
+        app.send_audio_bitrate(audio_app.audio_bitrate)
         app.send_resize_enabled(enable_resize)
         app.send_encoder(app.encoder)
         app.send_cursor_data(app.last_cursor_sent)
@@ -608,7 +636,7 @@ def data_channel_ready():
     webrtc_input.on_video_encoder_bit_rate = lambda bitrate: set_json_app_argument(args.json_config, "video_bitrate", bitrate) and (app.set_video_bitrate(int(bitrate)))
 
     # Send audio bitrate messages to app
-    webrtc_input.on_audio_encoder_bit_rate = lambda bitrate: set_json_app_argument(args.json_config, "audio_bitrate", bitrate) and app.set_audio_bitrate(int(bitrate))
+    webrtc_input.on_audio_encoder_bit_rate = lambda bitrate: set_json_app_argument(args.json_config, "audio_bitrate", bitrate) and audio_app.set_audio_bitrate(int(bitrate))
 
     # Send pointer visibility setting to app
     webrtc_input.on_mouse_pointer_visible = lambda visible: app.set_pointer_visible(
@@ -623,15 +651,6 @@ def set_fps_handler(fps):
         app.set_framerate(fps)
     webrtc_input.on_set_fps = lambda fps: set_fps_handler(fps)
 
-    # Write audio enabled argument to local configuration and then tell client to reload.
-    def enable_audio_handler(enabled):
-        set_json_app_argument(args.json_config, "enable_audio", enabled)
-        curr_audio = app.audio
-        app.set_enable_audio(enabled)
-        if enabled != curr_audio:
-            app.send_reload_window()
-    webrtc_input.on_set_enable_audio = lambda enabled: enable_audio_handler(enabled)
-
     # Handler for resize events.
     app.last_resize_success = True
     def on_resize_handler(res):
@@ -788,15 +807,25 @@ def mon_rtc_config(stun_servers, turn_servers, rtc_config):
 
         while True:
             asyncio.ensure_future(app.handle_bus_calls(), loop=loop)
+            asyncio.ensure_future(audio_app.handle_bus_calls(), loop=loop)
+
             loop.run_until_complete(signalling.connect())
-            loop.run_until_complete(signalling.start())            
+            loop.run_until_complete(audio_signalling.connect())
+
+            # asyncio.ensure_future(signalling.start(), loop=loop)
+            asyncio.ensure_future(audio_signalling.start(), loop=loop)
+            loop.run_until_complete(signalling.start())
+
             app.stop_pipeline()
+            audio_app.stop_pipeline()
             webrtc_input.stop_js_server()
     except Exception as e:
         logger.error("Caught exception: %s" % e)
+        traceback.print_exc()
         sys.exit(1)
     finally:
         app.stop_pipeline()
+        audio_app.stop_pipeline()
         webrtc_input.stop_clipboard()
         webrtc_input.stop_cursor_monitor()
         webrtc_input.stop_js_server()
diff --git a/src/selkies_gstreamer/gstwebrtc_app.py b/src/selkies_gstreamer/gstwebrtc_app.py
index 22b025d2..73bef0dd 100644
--- a/src/selkies_gstreamer/gstwebrtc_app.py
+++ b/src/selkies_gstreamer/gstwebrtc_app.py
@@ -44,7 +44,7 @@ class GSTWebRTCAppError(Exception):
 
 
 class GSTWebRTCApp:
-    def __init__(self, stun_servers=None, turn_servers=None, audio=True, audio_channels=2, framerate=30, encoder=None, video_bitrate=2000, audio_bitrate=64000):
+    def __init__(self, stun_servers=None, turn_servers=None, audio_channels=2, framerate=30, encoder=None, video_bitrate=2000, audio_bitrate=64000):
         """Initialize GStreamer WebRTC app.
 
         Initializes GObjects and checks for required plugins.
@@ -58,7 +58,6 @@ def __init__(self, stun_servers=None, turn_servers=None, audio=True, audio_chann
 
         self.stun_servers = stun_servers
         self.turn_servers = turn_servers
-        self.audio = audio
         self.audio_channels = audio_channels
         self.pipeline = None
         self.webrtcbin = None
@@ -85,7 +84,6 @@ def __init__(self, stun_servers=None, turn_servers=None, audio=True, audio_chann
 
         self.check_plugins()
 
-        self.pipeline = None
         self.ximagesrc = None
         self.ximagesrc_caps = None
         self.last_cursor_sent = None
@@ -776,15 +774,6 @@ def set_ice(self, mlineindex, candidate):
 
         self.webrtcbin.emit('add-ice-candidate', mlineindex, candidate)
 
-    def set_enable_audio(self, enabled):
-        """Set pipeline audio state
-
-        Arguments:
-            enabled {boolean} -- true of false indicating if audio should be enabled the next time the pipeline is built.
-        """
-        self.audio = enabled
-        logger.info("audio set to: %s" % str(enabled))
-
     def set_framerate(self, framerate):
         """Set pipeline framerate in fps
 
@@ -830,7 +819,7 @@ def set_audio_bitrate(self, bitrate):
             bitrate {integer} -- bitrate in bits per second, for example, 96000 for 96kbits/s.
         """
 
-        if self.audio:
+        if self.pipeline:
             element = Gst.Bin.get_by_name(self.pipeline, "opusenc")
             element.set_property("bitrate", bitrate)
 
@@ -920,14 +909,6 @@ def send_encoder(self, encoder):
         self.__send_data_channel_message(
             "system", {"action": "encoder,%s" % encoder})
 
-    def send_audio_enabled(self, audio_enabled):
-        """Sends the current audio state
-        """
-
-        logger.info("sending audio enabled")
-        self.__send_data_channel_message(
-            "system", {"action": "audio,"+str(audio_enabled)})
-
     def send_resize_enabled(self, resize_enabled):
         """Sends the current resize enabled state
         """
@@ -1087,7 +1068,7 @@ def bus_call(self, message):
 
         return True
 
-    def start_pipeline(self):
+    def start_pipeline(self, audio_only=False):
         """Starts the GStreamer pipeline
         """
 
@@ -1095,12 +1076,13 @@ def start_pipeline(self):
 
         self.pipeline = Gst.Pipeline.new()
 
-        # Construct the webrtcbin pipeline with video and audio.
+        # Construct the webrtcbin pipeline
         self.build_webrtcbin_pipeline()
-        self.build_video_pipeline()
 
-        if self.audio:
+        if audio_only:
             self.build_audio_pipeline()
+        else:
+            self.build_video_pipeline()
 
         # Advance the state of the pipeline to PLAYING.
         res = self.pipeline.set_state(Gst.State.PLAYING)
@@ -1108,21 +1090,22 @@ def start_pipeline(self):
             raise GSTWebRTCAppError(
                 "Failed to transition pipeline to PLAYING: %s" % res)
 
-        # Create the data channel, this has to be done after the pipeline is PLAYING.
-        options = Gst.Structure("application/data-channel")
-        options.set_value("ordered", True)
-        options.set_value("max-retransmits", 0)
-        self.data_channel = self.webrtcbin.emit(
-            'create-data-channel', "input", options)
-        self.data_channel.connect('on-open', lambda _: self.on_data_open())
-        self.data_channel.connect('on-close', lambda _: self.on_data_close())
-        self.data_channel.connect('on-error', lambda _: self.on_data_error())
-        self.data_channel.connect(
-            'on-message-string', lambda _, msg: self.on_data_message(msg))
-
-        # Enable NACKs on the transceiver, helps with retransmissions and freezing when packets are dropped.
-        transceiver = self.webrtcbin.emit("get-transceiver", 0)
-        transceiver.set_property("do-nack", True)
+        if audio_only is False:
+            # Create the data channel, this has to be done after the pipeline is PLAYING.
+            options = Gst.Structure("application/data-channel")
+            options.set_value("ordered", True)
+            options.set_value("max-retransmits", 0)
+            self.data_channel = self.webrtcbin.emit(
+                'create-data-channel', "input", options)
+            self.data_channel.connect('on-open', lambda _: self.on_data_open())
+            self.data_channel.connect('on-close', lambda _: self.on_data_close())
+            self.data_channel.connect('on-error', lambda _: self.on_data_error())
+            self.data_channel.connect(
+                'on-message-string', lambda _, msg: self.on_data_message(msg))
+
+            # Enable NACKs on the transceiver, helps with retransmissions and freezing when packets are dropped.
+            transceiver = self.webrtcbin.emit("get-transceiver", 0)
+            transceiver.set_property("do-nack", True)
 
         logger.info("pipeline started")
 
diff --git a/src/selkies_gstreamer/webrtc_input.py b/src/selkies_gstreamer/webrtc_input.py
index 16f73da4..29c33b51 100644
--- a/src/selkies_gstreamer/webrtc_input.py
+++ b/src/selkies_gstreamer/webrtc_input.py
@@ -117,8 +117,6 @@ def __init__(self, uinput_mouse_socket_path="", js_socket_path="", enable_clipbo
             'unhandled on_clipboard_read')
         self.on_set_fps = lambda fps: logger.warn(
             'unhandled on_set_fps')
-        self.on_set_enable_audio = lambda enable_audio: logger.warn(
-            'unhandled on_set_enable_audio')
         self.on_set_enable_resize = lambda enable_resize, res: logger.warn(
             'unhandled on_set_enable_resize')
         self.on_client_fps = lambda fps: logger.warn(
@@ -665,11 +663,6 @@ def on_message(self, msg):
             fps = int(toks[1])
             logger.info("Setting framerate to: %d" % fps)
             self.on_set_fps(fps)
-        elif toks[0] == "_arg_audio":
-            # Set audio enabled
-            enabled = toks[1].lower() == "true"
-            logger.info("Setting enable_audio to: %s" % str(enabled))
-            self.on_set_enable_audio(enabled)
         elif toks[0] == "_arg_resize":
             if len(toks) != 3:
                 logger.error("invalid _arg_resize command, expected 2 arguments <enabled>,<resolution>")
diff --git a/src/selkies_gstreamer/webrtc_signalling.py b/src/selkies_gstreamer/webrtc_signalling.py
index 59d3b3b8..f54e3480 100644
--- a/src/selkies_gstreamer/webrtc_signalling.py
+++ b/src/selkies_gstreamer/webrtc_signalling.py
@@ -76,7 +76,7 @@ def __init__(self, server, id, peer_id, enable_https=False, enable_basic_auth=Fa
         self.on_sdp = lambda sdp_type, sdp: logger.warn('unhandled sdp event')
         self.on_connect = lambda res, scale: logger.warn('unhandled on_connect callback')
         self.on_disconnect = lambda: logger.warn('unhandled on_disconnect callback')
-        self.on_session = lambda meta: logger.warn('unhandled on_session callback')
+        self.on_session = lambda peer_id, meta: logger.warn('unhandled on_session callback')
         self.on_error = lambda v: logger.warn(
             'unhandled on_error callback: %s', v)
 
@@ -175,7 +175,7 @@ async def start(self):
                 if len(toks) > 1:
                     meta = json.loads(base64.b64decode(toks[1]))
                 logger.info("started session with peer: %s, meta: %s", self.peer_id, json.dumps(meta))
-                self.on_session(meta)
+                self.on_session(self.peer_id, (meta))
             elif message.startswith('ERROR'):
                 if message == "ERROR peer '%s' not found" % self.peer_id:
                     await self.on_error(WebRTCSignallingErrorNoPeer("'%s' not found" % self.peer_id))