Give the local tracks their controls

Our own tracks are LocalAudioMediaTrack and LocalVideoMediaTrack, with setEnabled and setDevice, and on video facingMode$, switchFacingMode and setProcessor. switchCamera on the local media goes, as do the device and processor behaviors in LocalMediaInputs: a host calls the track when its setting changes instead of the SDK watching behaviors it does not own. The input device ids and the processor stay as initial values for the tracks created at the join. setMicrophoneEnabled and setCameraEnabled stay for the case of no track yet.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

sdk/src/api.ts

@@ -84,19 +84,19 @@ export interface VideoCaptureSettings {
 }
 
 /**
- * What the local member publishes. The enabled flags are the initial state;
- * from then on `setMicrophoneEnabled` and `setCameraEnabled` on the client
- * change it, and report what the device allowed.
+ * What the local member publishes when it joins. Initial values: from then on
+ * `setMicrophoneEnabled` and `setCameraEnabled` on the client, and the
+ * controls on the local tracks, change it.
  */
 export interface LocalMediaInputs {
   microphoneEnabled: boolean;
   cameraEnabled: boolean;
-  audioInputDeviceId$: Behavior<string | undefined>;
-  videoInputDeviceId$: Behavior<string | undefined>;
+  audioInputDeviceId?: string;
+  videoInputDeviceId?: string;
+  /** Background blur and the like. */
+  videoProcessor?: TrackProcessor<Track.Kind.Video>;
   /** Undefined where the host routes audio itself, or to leave the browser's choice. */
   audioOutputDeviceId$: Behavior<string | undefined>;
-  /** Background blur and the like. */
-  videoProcessor$: Behavior<TrackProcessor<Track.Kind.Video> | undefined>;
 }
 
 export type ConnectionStatus =
@@ -300,8 +300,35 @@ export interface AudioMediaTrack extends MediaTrack {
 
 export interface VideoMediaTrack extends MediaTrack {
   kind: "video";
-  /** Undefined for remote tracks. */
-  facingMode$?: Behavior<"user" | "environment" | undefined>;
+}
+
+/** The controls a member has over a track it publishes itself. */
+export interface LocalMediaTrack {
+  /**
+   * Mutes or unmutes. Resolves with the state that resulted, which differs
+   * from the request where the device could not be used.
+   */
+  setEnabled(enabled: boolean): Promise<boolean>;
+  /** Captures from another device. Rejects for a screen share, which has none. */
+  setDevice(deviceId: string): Promise<void>;
+}
+
+export interface LocalAudioMediaTrack
+  extends AudioMediaTrack, LocalMediaTrack {}
+
+export interface LocalVideoMediaTrack extends VideoMediaTrack, LocalMediaTrack {
+  /** For mirroring; undefined where the camera does not say which way it faces. */
+  facingMode$: Behavior<"user" | "environment" | undefined>;
+  /**
+   * Restarts the camera facing the other way, on devices with a front and a
+   * back camera, and resolves with the id of the device now in use. Does
+   * nothing where the facing mode is unknown.
+   */
+  switchFacingMode(): Promise<string | undefined>;
+  /** Background blur and the like; undefined removes the processor. */
+  setProcessor(
+    processor: TrackProcessor<Track.Kind.Video> | undefined,
+  ): Promise<void>;
 }
 
 export type EncryptionError = "MissingKey" | "InvalidKey";
@@ -325,11 +352,5 @@ export interface MemberMedia {
 
 export interface LocalMemberMedia extends MemberMedia {
   local: true;
-  /**
-   * Restarts the camera facing the other way, on devices with a front and a
-   * back camera, and resolves with the id of the device now in use. Does
-   * nothing without a camera track or where the facing mode is unknown (see
-   * `VideoMediaTrack.facingMode$`).
-   */
-  switchCamera(): Promise<string | undefined>;
+  tracks$: Behavior<(LocalAudioMediaTrack | LocalVideoMediaTrack)[]>;
 }

sdk/src/media/LivekitMediaTrack.test.ts

@@ -7,6 +7,7 @@ Please see LICENSE in the repository root for full details.
 
 import { EventEmitter } from "events";
 import {
+  type LocalParticipant,
   ParticipantEvent,
   RemoteAudioTrack,
   type RemoteParticipant,
@@ -19,7 +20,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import { ObservableScope } from "../reactive/ObservableScope";
 import { type AudioMediaTrack } from "../api";
-import { createLivekitMediaTrack } from "./LivekitMediaTrack";
+import {
+  createLivekitMediaTrack,
+  createLocalLivekitMediaTrack,
+} from "./LivekitMediaTrack";
 
 describe("createLivekitMediaTrack", () => {
   let scope: ObservableScope;
@@ -127,6 +131,27 @@ describe("createLivekitMediaTrack", () => {
     emitter.emit(ParticipantEvent.TrackMuted, publication);
     expect(track.isActive$.value).toBe(false);
   });
+
+  it("gives our own track its controls", async () => {
+    const { participant, publication, room } = fakes({
+      kind: Track.Kind.Audio,
+    });
+    const setEnabled = vi.fn().mockResolvedValue(false);
+    const switchActiveDevice = vi.fn().mockResolvedValue(true);
+    const track = createLocalLivekitMediaTrack(
+      scope,
+      Object.assign(participant, {
+        isLocal: true,
+      }) as unknown as LocalParticipant,
+      publication as unknown as TrackPublication,
+      Object.assign(room, { switchActiveDevice }),
+      setEnabled,
+    );
+    await expect(track.setEnabled(false)).resolves.toBe(false);
+    expect(setEnabled).toHaveBeenCalledWith("microphone", false);
+    await track.setDevice("mic-2");
+    expect(switchActiveDevice).toHaveBeenCalledWith("audioinput", "mic-2");
+  });
 });
 
 type FakeTrack = Track & {

sdk/src/media/LivekitMediaTrack.ts

@@ -12,6 +12,7 @@ import {
 } from "@livekit/components-core";
 import {
   facingModeFromLocalTrack,
+  type LocalParticipant,
   LocalTrack,
   LocalVideoTrack,
   type Participant,
@@ -41,11 +42,14 @@ import { type Behavior } from "../reactive/Behavior";
 import { type ObservableScope } from "../reactive/ObservableScope";
 import {
   type AudioMediaTrack,
+  type LocalAudioMediaTrack,
+  type LocalVideoMediaTrack,
   type MediaSource,
   type MediaStreamStats,
   type MediaTrack,
   type VideoMediaTrack,
 } from "../api";
+import { MatrixRTCError } from "../errors";
 import { LazyBehavior } from "../utils/LazyBehavior";
 
 const sources: Record<Track.Source, MediaSource> = {
@@ -56,6 +60,14 @@ const sources: Record<Track.Source, MediaSource> = {
   [Track.Source.Unknown]: "unknown",
 };
 
+export const livekitSources: Record<MediaSource, Track.Source> = {
+  microphone: Track.Source.Microphone,
+  camera: Track.Source.Camera,
+  screenShare: Track.Source.ScreenShare,
+  screenShareAudio: Track.Source.ScreenShareAudio,
+  unknown: Track.Source.Unknown,
+};
+
 // One timer for every track so that a large session does not keep hundreds of
 // them, each firing a statistics request, in the event loop.
 const refreshStats$ = interval(1000).pipe(startWith(0), share());
@@ -72,17 +84,7 @@ export function createLivekitMediaTrack(
   room: LivekitRoom,
 ): AudioMediaTrack | VideoMediaTrack {
   const mediaChanged$ = observeParticipantMedia(participant);
-  const track$: Behavior<Track | undefined> = scope.behavior(
-    merge(
-      mediaChanged$,
-      fromEvent(publication, TrackEvent.Subscribed),
-      fromEvent(publication, TrackEvent.Unsubscribed),
-    ).pipe(
-      map(() => publication.track),
-      startWith(publication.track),
-      distinctUntilChanged(),
-    ),
-  );
+  const track$ = publicationTrack$(scope, participant, publication);
 
   const attached = new Set<HTMLMediaElement>();
   let audioContext: AudioContext | undefined;
@@ -172,13 +174,85 @@ export function createLivekitMediaTrack(
       },
     };
 
+  return { ...base, kind: "video" };
+}
+
+/**
+ * One of our own publications, with the controls over it. `setEnabled` goes
+ * through the local member so that what it remembers for a reconnection
+ * stays in step.
+ */
+export function createLocalLivekitMediaTrack(
+  scope: ObservableScope,
+  participant: LocalParticipant,
+  publication: TrackPublication,
+  room: LivekitRoom,
+  setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>,
+): LocalAudioMediaTrack | LocalVideoMediaTrack {
+  const base = createLivekitMediaTrack(scope, participant, publication, room);
+  const local = {
+    setEnabled: async (enabled: boolean) => setEnabled(base.source, enabled),
+    setDevice: async (deviceId: string): Promise<void> => {
+      if (base.source !== "microphone" && base.source !== "camera")
+        throw new MatrixRTCError(`A ${base.source} track has no device`);
+      await room.switchActiveDevice(
+        base.kind === "audio" ? "audioinput" : "videoinput",
+        deviceId,
+      );
+    },
+  };
+  if (base.kind === "audio") return { ...base, ...local };
+
+  const videoTrack = (): LocalVideoTrack | undefined =>
+    publication.track instanceof LocalVideoTrack
+      ? publication.track
+      : undefined;
   return {
     ...base,
-    kind: "video",
-    ...(participant.isLocal && { facingMode$: facingMode$(scope, track$) }),
+    ...local,
+    facingMode$: facingMode$(
+      scope,
+      publicationTrack$(scope, participant, publication),
+    ),
+    switchFacingMode: async () => {
+      const track = videoTrack();
+      if (!track) return;
+      const { facingMode } = facingModeFromLocalTrack(track);
+      if (facingMode !== "user" && facingMode !== "environment") return;
+      await track.restartTrack({
+        facingMode: facingMode === "user" ? "environment" : "user",
+      });
+      return track.mediaStreamTrack.getSettings().deviceId;
+    },
+    setProcessor: async (processor) => {
+      const track = videoTrack();
+      // A processor cannot be built on a track that has already ended
+      if (!track || track.mediaStreamTrack.readyState === "ended") return;
+      if (processor) await track.setProcessor(processor);
+      else if (track.getProcessor()) await track.stopProcessor();
+    },
   };
 }
 
+/** The track behind a publication; a remote one arrives on subscription. */
+function publicationTrack$(
+  scope: ObservableScope,
+  participant: Participant,
+  publication: TrackPublication,
+): Behavior<Track | undefined> {
+  return scope.behavior(
+    merge(
+      observeParticipantMedia(participant),
+      fromEvent(publication, TrackEvent.Subscribed),
+      fromEvent(publication, TrackEvent.Unsubscribed),
+    ).pipe(
+      map(() => publication.track),
+      startWith(publication.track),
+      distinctUntilChanged(),
+    ),
+  );
+}
+
 async function rtpStreamStats(
   publication: TrackPublication,
   local: boolean,

sdk/src/media/LivekitMemberMedia.ts

@@ -10,13 +10,11 @@ import {
   roomEventSelector,
 } from "@livekit/components-core";
 import {
-  facingModeFromLocalTrack,
   type LocalParticipant,
-  LocalVideoTrack,
   type Participant,
   type Room as LivekitRoom,
   RoomEvent,
-  Track,
+  type TrackPublication,
 } from "livekit-client";
 import { filter, map, type Observable } from "rxjs";
 
@@ -24,12 +22,17 @@ import { type ObservableScope } from "../reactive/ObservableScope";
 import { generateItems } from "../reactive/observable";
 import { E2eeType } from "../encryption";
 import { type EncryptionSystem } from "../encryption";
+import { type Behavior } from "../reactive/Behavior";
 import {
   type EncryptionError,
   type LocalMemberMedia,
+  type MediaSource,
   type MemberMedia,
 } from "../api";
-import { createLivekitMediaTrack } from "./LivekitMediaTrack";
+import {
+  createLivekitMediaTrack,
+  createLocalLivekitMediaTrack,
+} from "./LivekitMediaTrack";
 
 /**
  * A participant as `MemberMedia`. The only place, with `LivekitMediaTrack`,
@@ -43,26 +46,9 @@ export function createLivekitMemberMedia(
   encryptionSystem: EncryptionSystem,
 ): MemberMedia {
   return {
-    local: participant.isLocal,
-    // One track per publication, each living as long as its publication is
-    // in the participant's map
-    tracks$: scope.behavior(
-      observeParticipantMedia(participant).pipe(
-        generateItems(
-          `${participant.identity} tracks$`,
-          function* () {
-            for (const publication of participant.trackPublications.values())
-              yield { keys: [publication.trackSid], data: publication };
-          },
-          (trackScope, publication$) =>
-            createLivekitMediaTrack(
-              trackScope,
-              participant,
-              publication$.value,
-              room,
-            ),
-        ),
-      ),
+    local: false,
+    tracks$: memberTracks$(scope, participant, (trackScope, publication) =>
+      createLivekitMediaTrack(trackScope, participant, publication, room),
     ),
     encryptionError$: encryptionErrors$(
       scope,
@@ -78,23 +64,48 @@ export function createLocalLivekitMemberMedia(
   participant: LocalParticipant,
   room: LivekitRoom,
   encryptionSystem: EncryptionSystem,
+  setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>,
 ): LocalMemberMedia {
   return {
-    ...createLivekitMemberMedia(scope, participant, room, encryptionSystem),
     local: true,
-    switchCamera: async () => {
-      const track = participant.getTrackPublication(Track.Source.Camera)?.track;
-      if (!(track instanceof LocalVideoTrack)) return;
-      const { facingMode } = facingModeFromLocalTrack(track);
-      if (facingMode !== "user" && facingMode !== "environment") return;
-      await track.restartTrack({
-        facingMode: facingMode === "user" ? "environment" : "user",
-      });
-      return track.mediaStreamTrack.getSettings().deviceId;
-    },
+    tracks$: memberTracks$(scope, participant, (trackScope, publication) =>
+      createLocalLivekitMediaTrack(
+        trackScope,
+        participant,
+        publication,
+        room,
+        setEnabled,
+      ),
+    ),
+    encryptionError$: encryptionErrors$(
+      scope,
+      participant,
+      room,
+      encryptionSystem,
+    ),
   };
 }
 
+/** One track per publication, each living as long as its publication is in the participant's map. */
+function memberTracks$<T>(
+  scope: ObservableScope,
+  participant: Participant,
+  factory: (scope: ObservableScope, publication: TrackPublication) => T,
+): Behavior<T[]> {
+  return scope.behavior(
+    observeParticipantMedia(participant).pipe(
+      generateItems(
+        `${participant.identity} tracks$`,
+        function* () {
+          for (const publication of participant.trackPublications.values())
+            yield { keys: [publication.trackSid], data: publication };
+        },
+        (trackScope, publication$) => factory(trackScope, publication$.value),
+      ),
+    ),
+  );
+}
+
 function encryptionErrors$(
   scope: ObservableScope,
   participant: Participant,

sdk/src/session/ConnectionFactory.ts

@@ -138,8 +138,8 @@ function roomOptions(
     ...base,
     videoCaptureDefaults: {
       ...base.videoCaptureDefaults,
-      deviceId: localMedia.videoInputDeviceId$.value,
-      processor: localMedia.videoProcessor$.value,
+      deviceId: localMedia.videoInputDeviceId,
+      processor: localMedia.videoProcessor,
       ...(camera?.resolution && { resolution: camera.resolution }),
     },
     publishDefaults: {
@@ -154,7 +154,7 @@ function roomOptions(
     },
     audioCaptureDefaults: {
       ...base.audioCaptureDefaults,
-      deviceId: localMedia.audioInputDeviceId$.value,
+      deviceId: localMedia.audioInputDeviceId,
       ...capture?.audio,
     },
     audioOutput: { deviceId: localMedia.audioOutputDeviceId$.value },

sdk/src/session/LocalMember.ts

@@ -46,7 +46,11 @@ import {
   type MatrixRTCMode,
   type SessionTimings,
 } from "../config";
-import { type DisconnectReason, type VideoCaptureSettings } from "../api";
+import {
+  type DisconnectReason,
+  type MediaSource,
+  type VideoCaptureSettings,
+} from "../api";
 import {
   FailToStartLivekitConnection,
   type MatrixRTCError,
@@ -58,6 +62,7 @@ import { type Connection, ConnectionState } from "./Connection";
 import { type IConnectionManager } from "./ConnectionManager";
 import { type LocalTransport } from "./LocalTransport";
 import { getSFUConfigWithOpenID } from "./openIDSFU";
+import { livekitSources } from "../media/LivekitMediaTrack";
 import { type DesiredMedia, type Publisher } from "./Publisher";
 
 export enum TransportState {
@@ -118,6 +123,8 @@ export interface LocalMembership {
   disconnectReason$: Behavior<DisconnectReason | null>;
   setMicrophoneEnabled: (enabled: boolean) => Promise<boolean>;
   setCameraEnabled: (enabled: boolean) => Promise<boolean>;
+  /** Mutes or unmutes one of our publications; the result is the state that resulted. */
+  setEnabled: (source: MediaSource, enabled: boolean) => Promise<boolean>;
   sharingScreen$: Behavior<boolean>;
   toggleScreenSharing: (() => void) | null;
   screenShareError$: Behavior<Error | null>;
@@ -392,22 +399,40 @@ export function createLocalMembership$({
     });
 
   const setEnabled = async (
-    source: Track.Source.Microphone | Track.Source.Camera,
+    source: MediaSource,
     enabled: boolean,
   ): Promise<boolean> => {
+    if (source !== "microphone" && source !== "camera")
+      return setPublicationEnabled(source, enabled);
     const desired$ =
-      source === Track.Source.Microphone
-        ? desired.microphone$
-        : desired.camera$;
+      source === "microphone" ? desired.microphone$ : desired.camera$;
     desired$.next(enabled);
     // Without a publisher the request waits for the tracks to be created
     const publisher = publisher$.value;
     if (publisher === null) return enabled;
-    const result = await publisher.setEnabled(source, enabled);
+    const result = await publisher.setEnabled(
+      source === "microphone" ? Track.Source.Microphone : Track.Source.Camera,
+      enabled,
+    );
     if (result !== enabled) desired$.next(result);
     return result;
   };
 
+  // A publication the publisher does not recreate on reconnection, so there
+  // is nothing to remember for it
+  const setPublicationEnabled = async (
+    source: MediaSource,
+    enabled: boolean,
+  ): Promise<boolean> => {
+    const publication = participant$.value?.getTrackPublication(
+      livekitSources[source],
+    );
+    if (!publication) return false;
+    if (enabled) await publication.unmute();
+    else await publication.mute();
+    return !publication.isMuted;
+  };
+
   const sharingScreen$ = scope.behavior(
     participant$.pipe(
       switchMap((p) => (p === null ? of(false) : observeSharingScreen$(p))),
@@ -451,10 +476,9 @@ export function createLocalMembership$({
     connected$,
     reconnecting$,
     disconnectReason$,
-    setMicrophoneEnabled: async (enabled) =>
-      setEnabled(Track.Source.Microphone, enabled),
-    setCameraEnabled: async (enabled) =>
-      setEnabled(Track.Source.Camera, enabled),
+    setMicrophoneEnabled: async (enabled) => setEnabled("microphone", enabled),
+    setCameraEnabled: async (enabled) => setEnabled("camera", enabled),
+    setEnabled,
     sharingScreen$,
     toggleScreenSharing,
     screenShareError$,

sdk/src/session/Members.ts

@@ -93,6 +93,7 @@ export function createLocalRTCMember(
           participant,
           room,
           context.encryptionSystem,
+          localMembership.setEnabled,
         ),
     ),
     sharingScreen$: localMembership.sharingScreen$,

sdk/src/session/Publisher.ts

@@ -5,24 +5,16 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { observeParticipantMedia } from "@livekit/components-core";
 import {
   ConnectionState as LivekitConnectionState,
   type LocalTrackPublication,
-  LocalVideoTrack,
   ParticipantEvent,
   type Room as LivekitRoom,
   Track,
 } from "livekit-client";
 import { type Logger } from "matrix-js-sdk/lib/logger";
-import {
-  type BehaviorSubject,
-  combineLatest,
-  distinctUntilChanged,
-  map,
-} from "rxjs";
+import { type BehaviorSubject } from "rxjs";
 
-import { type Behavior } from "../reactive/Behavior";
 import { ObservableScope } from "../reactive/ObservableScope";
 import { type LocalMediaInputs } from "../api";
 
@@ -57,8 +49,7 @@ export class Publisher {
     room.setE2EEEnabled(room.options.e2ee !== undefined)?.catch((e: Error) => {
       this.logger.error("Failed to enable E2EE on the room", e);
     });
-    this.followVideoProcessor();
-    this.followDevices();
+    this.followAudioOutput();
     this.onLocalTrackPublished = this.onLocalTrackPublished.bind(this);
     room.localParticipant.on(
       ParticipantEvent.LocalTrackPublished,
@@ -188,54 +179,19 @@ export class Publisher {
     }
   }
 
-  private followDevices(): void {
-    const sync = (
-      kind: MediaDeviceKind,
-      deviceId$: Behavior<string | undefined>,
-    ): void => {
-      deviceId$.pipe(this.scope.bind()).subscribe((deviceId) => {
+  private followAudioOutput(): void {
+    this.inputs.audioOutputDeviceId$
+      .pipe(this.scope.bind())
+      .subscribe((deviceId) => {
         if (
           deviceId === undefined ||
           this.room.state !== LivekitConnectionState.Connected ||
-          this.room.getActiveDevice(kind) === deviceId
+          this.room.getActiveDevice("audiooutput") === deviceId
         )
           return;
         this.room
-          .switchActiveDevice(kind, deviceId)
-          .catch((e) => this.logger.error(`Failed to switch ${kind}`, e));
-      });
-    };
-    sync("audioinput", this.inputs.audioInputDeviceId$);
-    sync("audiooutput", this.inputs.audioOutputDeviceId$);
-    sync("videoinput", this.inputs.videoInputDeviceId$);
-  }
-
-  private followVideoProcessor(): void {
-    const participant = this.room.localParticipant;
-    const cameraTrack$ = observeParticipantMedia(participant).pipe(
-      map(() => {
-        const track = participant.getTrackPublication(
-          Track.Source.Camera,
-        )?.track;
-        return track instanceof LocalVideoTrack ? track : undefined;
-      }),
-      distinctUntilChanged(),
-    );
-    combineLatest([cameraTrack$, this.inputs.videoProcessor$])
-      .pipe(this.scope.bind())
-      .subscribe(([track, processor]) => {
-        if (!track) return;
-        if (processor && !track.getProcessor()) {
-          // A processor cannot be built on a track that has already ended
-          if (track.mediaStreamTrack.readyState === "ended") return;
-          track.setProcessor(processor).catch((e) => {
-            this.logger.warn("Failed to attach the video processor", e);
-          });
-        } else if (!processor && track.getProcessor()) {
-          track.stopProcessor().catch((e) => {
-            this.logger.warn("Failed to stop the video processor", e);
-          });
-        }
+          .switchActiveDevice("audiooutput", deviceId)
+          .catch((e) => this.logger.error("Failed to switch audiooutput", e));
       });
   }
 }