Skip to content

Refactor ring$ observable #3504

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
toger5 merged 3 commits into from
Sep 19, 2025
Merged

Refactor ring$ observable #3504

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c3f49fa
Refactor ring$ observable
toger5 Sep 19, 2025
e38c04b
fix ci
toger5 Sep 19, 2025
dd978b0
fix regression test
toger5 Sep 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions src/room/InCallView.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ import { prefetchSounds } from "../soundUtils";
import { useAudioContext } from "../useAudioContext";
import ringtoneMp3 from "../sound/ringtone.mp3?url";
import ringtoneOgg from "../sound/ringtone.ogg?url";
import { ObservableScope } from "../state/ObservableScope.ts";

const canScreenshare = "getDisplayMedia" in (navigator.mediaDevices ?? {});

Expand All @@ -144,8 +145,13 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
sfuConfig,
props.e2eeSystem,
);
const connStateObservable$ = useObservable(
(inputs$) => inputs$.pipe(map(([connState]) => connState)),
const observableScope = useInitial(() => new ObservableScope());
const connStateBehavior$ = useObservable(
(inputs$) =>
observableScope.behavior(
inputs$.pipe(map(([connState]) => connState)),
connState,
),
[connState],
);
const [vm, setVm] = useState<CallViewModel | null>(null);
Expand Down Expand Up @@ -188,7 +194,7 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
waitForCallPickup:
waitForCallPickup && sendNotificationType === "ring",
},
connStateObservable$,
connStateBehavior$,
reactionsReader.raisedHands$,
reactionsReader.reactions$,
);
Expand All @@ -204,7 +210,7 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
livekitRoom,
mediaDevices,
props.e2eeSystem,
connStateObservable$,
connStateBehavior$,
autoLeaveWhenOthersLeft,
sendNotificationType,
waitForCallPickup,
Expand Down
8 changes: 5 additions & 3 deletions src/state/CallViewModel.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,7 @@ const mockLegacyRingEvent = {} as { event_id: string } & ICallNotifyContent;
interface CallViewModelInputs {
remoteParticipants$: Behavior<RemoteParticipant[]>;
rtcMembers$: Behavior<Partial<CallMembership>[]>;
livekitConnectionState$: Observable<ECConnectionState>;
livekitConnectionState$: Behavior<ECConnectionState>;
speaking: Map<Participant, Observable<boolean>>;
mediaDevices: MediaDevices;
initialSyncState: SyncState;
Expand All @@ -276,7 +276,9 @@ function withCallViewModel(
{
remoteParticipants$ = constant([]),
rtcMembers$ = constant([localRtcMember]),
livekitConnectionState$: connectionState$ = of(ConnectionState.Connected),
livekitConnectionState$: connectionState$ = constant(
ConnectionState.Connected,
),
speaking = new Map(),
mediaDevices = mockMediaDevices({}),
initialSyncState = SyncState.Syncing,
Expand Down Expand Up @@ -1272,7 +1274,7 @@ describe("waitForCallPickup$", () => {
},
});

expectObservable(vm.callPickupState$).toBe("a 9ms b 29ms c", {
expectObservable(vm.callPickupState$).toBe("a 9ms b 19ms c", {
Copy link
Copy Markdown
Contributor Author

@toger5 toger5 Sep 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This was actually wrong before the refactor.

Before the timer was only ever started once connected. But it might have sent the notification event a couple of seconds before (if the connection to the sfu is very poor)
The test has passed previously because we only were running ring$.pipe(...) once we were connected. Now we subscribe to ring$ right in the combineLatest so we start the timer instantly.
Basically even though the observable with the timer was already created it was only subscribed later. (the inner observable at least. the other one was subscribed by the behavior scope immediately that is why the didSendCallNotification$ was still caught.

I hope this makes sense. The more direct justification for this change is:

 // Fire a call notification IMMEDIATELY (its important for this test, that this happens before the livekitConnectionState$ emits)
          schedule("n", {
            n: () => {
              rtcSession.emit(
                MatrixRTCSessionEvent.DidSendCallNotification,
                mockRingEvent("$notif1", 30),
                mockLegacyRingEvent,
              );
            },
          });

Says we fire the event immediately and it rings for 30ms -> it should stop after 30ms -> c=timeout should happen after 30ms
before: "a 9ms b 29ms c" 1ms+9ms+1ms+29ms = 40ms is just wrong
now: "a 9ms b 19ms c" 1ms+9ms+1ms+19ms = 30ms is what we would expect with a 30ms notificaiton event

Copy link
Copy Markdown
Member

@Half-Shot Half-Shot Sep 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That makes sense to me 👍 , good catch.

a: "unknown",
b: "ringing",
c: "timeout",
Expand Down
115 changes: 54 additions & 61 deletions src/state/CallViewModel.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -898,58 +898,59 @@ export class CallViewModel extends ViewModel {
// A behavior will emit the latest observable with the running timer to new subscribers.
// see also: callPickupState$ and in particular the line: `return this.ring$.pipe(mergeAll());` here we otherwise might get an EMPTY observable if
// `ring$` would not be a behavior.
private readonly ring$: Behavior<
Observable<"ringing" | "timeout" | "decline"> | Observable<never>
> = this.scope.behavior(
this.didSendCallNotification$.pipe(
filter(
([notificationEvent]) => notificationEvent.notification_type === "ring",
),
map(([notificationEvent]) => {
const lifetimeMs = notificationEvent?.lifetime ?? 0;
return concat(
lifetimeMs === 0
? // If no lifetime, skip the ring state
EMPTY
: // Ring until lifetime ms have passed
timer(lifetimeMs).pipe(
ignoreElements(),
startWith("ringing" as const),
),
// The notification lifetime has timed out, meaning ringing has likely
// stopped on all receiving clients.
of("timeout" as const),
NEVER,
).pipe(
takeUntil(
(
fromEvent(this.matrixRoom, RoomEvent.Timeline) as Observable<
Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
>
).pipe(
filter(
([event]) =>
event.getType() === EventType.RTCDecline &&
event.getRelation()?.rel_type === "m.reference" &&
event.getRelation()?.event_id ===
notificationEvent.event_id &&
event.getSender() !== this.userId,
private readonly ring$: Behavior<"ringing" | "timeout" | "decline" | null> =
this.scope.behavior(
this.didSendCallNotification$.pipe(
filter(
([notificationEvent]) =>
notificationEvent.notification_type === "ring",
),
switchMap(([notificationEvent]) => {
const lifetimeMs = notificationEvent?.lifetime ?? 0;
return concat(
lifetimeMs === 0
? // If no lifetime, skip the ring state
of(null)
: // Ring until lifetime ms have passed
timer(lifetimeMs).pipe(
ignoreElements(),
startWith("ringing" as const),
),
// The notification lifetime has timed out, meaning ringing has likely
// stopped on all receiving clients.
of("timeout" as const),
// This makes sure we will not drop into the `endWith("decline" as const)` state
NEVER,
).pipe(
takeUntil(
(
fromEvent(this.matrixRoom, RoomEvent.Timeline) as Observable<
Parameters<EventTimelineSetHandlerMap[RoomEvent.Timeline]>
>
).pipe(
filter(
([event]) =>
event.getType() === EventType.RTCDecline &&
event.getRelation()?.rel_type === "m.reference" &&
event.getRelation()?.event_id ===
notificationEvent.event_id &&
event.getSender() !== this.userId,
),
),
),
),
endWith("decline" as const),
);
}),
),
EMPTY,
);
endWith("decline" as const),
);
}),
),
null,
);

/**
* Whether some Matrix user other than ourself is joined to the call.
*/
private readonly someoneElseJoined$ = this.memberships$.pipe(
map((ms) => ms.some((m) => m.sender !== this.userId)),
);
) as Behavior<boolean>;

/**
* The current call pickup state of the call.
Expand All @@ -968,27 +969,19 @@ export class CallViewModel extends ViewModel {
? this.scope.behavior<
"unknown" | "ringing" | "timeout" | "decline" | "success"
>(
combineLatest([
this.livekitConnectionState$,
this.someoneElseJoined$,
]).pipe(
switchMap(([livekitConnectionState, someoneElseJoined]) => {
combineLatest(
[this.livekitConnectionState$, this.someoneElseJoined$, this.ring$],
(livekitConnectionState, someoneElseJoined, ring) => {
if (livekitConnectionState === ConnectionState.Disconnected) {
// Do not ring until we're connected.
return of("unknown" as const);
return "unknown" as const;
} else if (someoneElseJoined) {
return of("success" as const);
return "success" as const;
}
// Show the ringing state of the most recent ringing attempt.
// ring$ is a behavior so it will emit the latest observable which very well might already have a running timer.
// this is important in case livekitConnectionState$ after didSendCallNotification$ has already emitted.
return this.ring$.pipe(switchAll());
}),
// The state starts as 'unknown' because we don't know if the RTC
// session will actually send a notify event yet. It will only be
// known once we send our own membership and see that we were the
// first one to join.
startWith("unknown" as const),
// as long as we have not yet sent an RTC notification event, ring will be null -> callPickupState$ = unknown.
return ring ?? ("unknown" as const);
},
),
)
: constant(null);
Expand Down Expand Up @@ -1700,7 +1693,7 @@ export class CallViewModel extends ViewModel {
private readonly livekitRoom: LivekitRoom,
private readonly mediaDevices: MediaDevices,
private readonly options: CallViewModelOptions,
public readonly livekitConnectionState$: Observable<ECConnectionState>,
public readonly livekitConnectionState$: Behavior<ECConnectionState>,
private readonly handsRaisedSubject$: Observable<
Record<string, RaisedHandInfo>
>,
Expand Down
3 changes: 2 additions & 1 deletion src/utils/test-viewmodel.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ import {
localRtcMember,
} from "./test-fixtures";
import { type RaisedHandInfo, type ReactionInfo } from "../reactions";
import { constant } from "../state/Behavior";

export function getBasicRTCSession(
members: RoomMember[],
Expand Down Expand Up @@ -154,7 +155,7 @@ export function getBasicCallViewModelEnvironment(
encryptionSystem: { kind: E2eeType.PER_PARTICIPANT },
...callViewModelOptions,
},
of(ConnectionState.Connected),
constant(ConnectionState.Connected),
handRaisedSubject$,
reactionsSubject$,
);
Expand Down