Skip to content
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

[Enhancement]Debounce SetPublisher requests #709

Closed
wants to merge 2 commits into from
+4 −2
Closed

[Enhancement]Debounce SetPublisher requests #709

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
55b6267
[Enhancement]Debounce SetPublisher requests
ipavlidakis Mar 12, 2025
b4ac0c6
Update tests
ipavlidakis Mar 13, 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
2 changes: 2 additions & 0 deletions Sources/StreamVideo/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator.swift
View file
Original file line number Diff line number Diff line change
@@ -36,6 +36,7 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
private let subsystem: LogSubsystem
private let disposableBag: DisposableBag = .init()
private let dispatchQueue = DispatchQueue(label: "io.getstream.peerconnection.serial.offer.queue")
private let debouncedIntervalInMilliseconds = 250 // The interval used to debounce the SetPublisher requests.

/// `SetPublisher` and `HandleSubscriberOffer` are expected from the SFU to be sent/handled
/// in a serial manner. The processing queues below ensure that the respective tasks are being executed
@@ -206,6 +207,7 @@ class RTCPeerConnectionCoordinator: @unchecked Sendable {
peerConnection
.publisher(eventType: StreamRTCPeerConnection.ShouldNegotiateEvent.self)
.log(.debug) { _ in "Publisher will negotiate" }
.debounce(for: .milliseconds(250), scheduler: RunLoop.main)
.receive(on: dispatchQueue)
.map { _ in () }
.sinkTask(queue: setPublisherProcessingQueue) { [weak self] in await self?.negotiate() }
4 changes: 2 additions & 2 deletions StreamVideoTests/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator_Tests.swift
View file
Original file line number Diff line number Diff line change
@@ -307,7 +307,7 @@
XCTAssertEqual(mockPeerConnection?.timesCalled(.setRemoteDescription), 0)
}

func test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferEveryTime(
func test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferOnlyOnce(
) async throws {
_ = subject
let offerA = RTCSessionDescription(
@@ -341,7 +341,7 @@
}

await fulfillment { [mockPeerConnection] in
mockPeerConnection?.timesCalled(.setLocalDescription) == 2
mockPeerConnection?.timesCalled(.setLocalDescription) == 1
}

XCTAssertEqual(
@@ -351,7 +351,7 @@
)?.first?.sdp,
offerA.sdp
)
XCTAssertEqual(

Check failure on line 354 in StreamVideoTests/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator_Tests.swift

GitHub Actions / Test LLC (Debug) 

    test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferOnlyOnce, XCTAssertEqual failed: ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerA")") is not equal to ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerB")")

Check failure on line 354 in StreamVideoTests/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator_Tests.swift

GitHub Actions / Test LLC (Debug) 

    test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferOnlyOnce, XCTAssertEqual failed: ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerA")") is not equal to ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerB")")

Check failure on line 354 in StreamVideoTests/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator_Tests.swift

GitHub Actions / Test LLC (Debug) 

    test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferOnlyOnce, XCTAssertEqual failed: ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerA")") is not equal to ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerB")")

Check failure on line 354 in StreamVideoTests/WebRTC/v2/PeerConnection/RTCPeerConnectionCoordinator_Tests.swift

GitHub Actions / Test LLC (Debug) 

    test_negotiate_subjectIsPublisher_multipleRequestsExecuteSerially_callSetPublisherOnSFUWithCorrectOfferOnlyOnce, XCTAssertEqual failed: ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerA")") is not equal to ("Optional("useinbandfec=1;\r\n00:11 opus/;\r\n12:13: red/48000/2l;offerB")")
mockPeerConnection.recordedInputPayload(
RTCSessionDescription.self,
for: .setLocalDescription

Unchanged files with check annotations Beta

View file
...Tests/WebRTC/v2/StateMachine/Stages/WebRTCCoordinatorStateMachine_JoiningStageTests.swift
func test_transition_fromConnectedWithRejoin_sendsExpectedJoinRequest() async throws {
throw XCTSkip("To be fixed")
subject.context.coordinator = mockCoordinatorStack.coordinator

Check warning on line 418 in StreamVideoTests/WebRTC/v2/StateMachine/Stages/WebRTCCoordinatorStateMachine_JoiningStageTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
let previousSessionId = String.unique
subject.context.isRejoiningFromSessionID = previousSessionId
subject.context.reconnectAttempts = 11
func test_transition_fromFastReconnected_sendsExpectedJoinRequest() async throws {
throw XCTSkip("To be fixed")
subject.context.coordinator = mockCoordinatorStack.coordinator

Check warning on line 802 in StreamVideoTests/WebRTC/v2/StateMachine/Stages/WebRTCCoordinatorStateMachine_JoiningStageTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
subject.context.reconnectAttempts = 11
let mockRTCCoordinator = try MockRTCPeerConnectionCoordinator(
sessionId: "stub-session-id",
func test_transition_fromMigrated_sendsExpectedJoinRequest() async throws {
throw XCTSkip("To be fixed")
subject.context.coordinator = mockCoordinatorStack.coordinator

Check warning on line 1001 in StreamVideoTests/WebRTC/v2/StateMachine/Stages/WebRTCCoordinatorStateMachine_JoiningStageTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
subject.context.reconnectAttempts = 11
subject.context.migratingFromSFU = "test-sfu"
let mockRTCCoordinator = try MockRTCPeerConnectionCoordinator(
View file
StreamVideoTests/IntegrationTests/CallCRUDTests.swift
func test_muteUserById() async throws {
throw XCTSkip("https://github.com/GetStream/ios-issues-tracking/issues/541")
let firstUserCall = client.call(callType: String.audioRoom, callId: randomCallId)

Check warning on line 387 in StreamVideoTests/IntegrationTests/CallCRUDTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
try await firstUserCall.create(memberIds: [user1, user2])
try await firstUserCall.goLive()
func test_muteAllUsers() async throws {
throw XCTSkip("https://github.com/GetStream/ios-issues-tracking/issues/541")
let firstUserCall = client.call(callType: String.audioRoom, callId: randomCallId)

Check warning on line 431 in StreamVideoTests/IntegrationTests/CallCRUDTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
try await firstUserCall.create(memberIds: [user1, user2])
try await firstUserCall.goLive()
func test_grantPermissions() async throws {
throw XCTSkip("https://github.com/GetStream/ios-issues-tracking/issues/541")
let call = client.call(callType: defaultCallType, callId: randomCallId)

Check warning on line 505 in StreamVideoTests/IntegrationTests/CallCRUDTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
try await call.create(memberIds: [user1])
let expectedPermissions: [Permission] = [.sendAudio, .sendVideo, .screenshare]
func test_grantPermissionsByRequest() async throws {
throw XCTSkip("https://github.com/GetStream/ios-issues-tracking/issues/541")
let firstUserCall = client.call(callType: String.audioRoom, callId: randomCallId)

Check warning on line 529 in StreamVideoTests/IntegrationTests/CallCRUDTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
try await firstUserCall.create(memberIds: [user1, user2])
let secondUserClient = try await makeClient(for: user2)
func test_setAndDeleteVoipDevices() async throws {
throw XCTSkip("https://github.com/GetStream/ios-issues-tracking/issues/541")
let deviceId = UUID().uuidString

Check warning on line 615 in StreamVideoTests/IntegrationTests/CallCRUDTests.swift

GitHub Actions / Test LLC (Debug) 

code after 'throw' will never be executed
let call = client.call(callType: defaultCallType, callId: randomCallId)
try await call.create(memberIds: [user1, user2])