mirror of
https://github.com/Novaloop-AG/element-x-ios.git
synced 2026-11-03 06:55:59 +00:00
Fix CurrentValuePublisher.map dropping updates when the mapped publisher is chained inline (#5870)
* Fix CurrentValuePublisher.map dropping updates when the mapped publisher is chained inline
The eager implementation bridged through a second CurrentValueSubject fed by
a sink whose AnyCancellable was stored on the returned struct. Combine
consumes publisher structs when building a subscription, so chaining the
mapped publisher inline (e.g. `publisher.map { … }.removeDuplicates(…)
.sink { … }`, as RoomScreenViewModel and RoomDetailsScreenViewModel do for
knock requests) dropped the only copy holding that cancellable: the feeding
sink was cancelled and subscribers only ever received the initial value.
In practice this meant a room admin viewing a knockable room never saw the
"wants to join" banner (or a live "Requests to join" badge) for a knock
arriving while the room was open — it only appeared when the knock predated
opening the room. Caught by a live cross-client knocking test; the
double-execution of the transform at subscription time (eager transform +
current-value replay) was the giveaway.
Make map lazy instead: keep the transformed AnyPublisher and a value
provider, so subscribers attach straight through to the source subject (which
replays its current value itself) and no keep-alive is needed. Adds a
regression test chaining a mapped publisher inline.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* swiftformat
* better docs
---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Mauro Romito <mauro.romito@element.io>
This commit is contained in:
@@ -13,33 +13,41 @@ import Combine
|
|||||||
///
|
///
|
||||||
/// `CurrentValueSubject` is documented as thread-safe but is not formally `Sendable`, hence `@unchecked`.
|
/// `CurrentValueSubject` is documented as thread-safe but is not formally `Sendable`, hence `@unchecked`.
|
||||||
nonisolated struct CurrentValuePublisher<Output, Failure: Error>: Publisher, @unchecked Sendable {
|
nonisolated struct CurrentValuePublisher<Output, Failure: Error>: Publisher, @unchecked Sendable {
|
||||||
private let subject: CurrentValueSubject<Output, Failure>
|
private let upstream: AnyPublisher<Output, Failure>
|
||||||
/// Retains the upstream subscription feeding ``subject`` when this publisher is derived via ``map(_:)``.
|
private let valueProvider: () -> Output
|
||||||
private let cancellable: AnyCancellable?
|
|
||||||
|
|
||||||
init(_ subject: CurrentValueSubject<Output, Failure>, cancellable: AnyCancellable? = nil) {
|
init(_ subject: CurrentValueSubject<Output, Failure>) {
|
||||||
self.subject = subject
|
upstream = subject.eraseToAnyPublisher()
|
||||||
self.cancellable = cancellable
|
valueProvider = { subject.value }
|
||||||
}
|
}
|
||||||
|
|
||||||
init(_ value: Output) {
|
init(_ value: Output) {
|
||||||
self.init(CurrentValueSubject(value))
|
self.init(CurrentValueSubject(value))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private init(upstream: AnyPublisher<Output, Failure>, valueProvider: @escaping () -> Output) {
|
||||||
|
self.upstream = upstream
|
||||||
|
self.valueProvider = valueProvider
|
||||||
|
}
|
||||||
|
|
||||||
func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
|
func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
|
||||||
subject.receive(subscriber: subscriber)
|
upstream.receive(subscriber: subscriber)
|
||||||
}
|
}
|
||||||
|
|
||||||
var value: Output {
|
var value: Output {
|
||||||
subject.value
|
valueProvider()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
nonisolated extension CurrentValuePublisher where Failure == Never {
|
nonisolated extension CurrentValuePublisher where Failure == Never {
|
||||||
|
/// Transforms the published values and ``value``.
|
||||||
|
///
|
||||||
|
/// The transform runs lazily, once per subscriber and per ``value`` read. It must stay lazy:
|
||||||
|
/// this struct is discarded when the mapped publisher is chained inline, so anything eager
|
||||||
|
/// (a live subscription feeding a second subject) would be cancelled and stop the updates.
|
||||||
func map<T>(_ transform: @escaping (Output) -> T) -> CurrentValuePublisher<T, Never> {
|
func map<T>(_ transform: @escaping (Output) -> T) -> CurrentValuePublisher<T, Never> {
|
||||||
let subject = CurrentValueSubject<T, Never>(transform(value))
|
.init(upstream: upstream.map(transform).eraseToAnyPublisher(),
|
||||||
let cancellable = sink { subject.send(transform($0)) }
|
valueProvider: { transform(value) })
|
||||||
return .init(subject, cancellable: cancellable)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,48 @@
|
|||||||
|
//
|
||||||
|
// Copyright 2026 Element Creations Ltd.
|
||||||
|
//
|
||||||
|
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||||
|
// Please see LICENSE files in the repository root for full details.
|
||||||
|
//
|
||||||
|
|
||||||
|
import Combine
|
||||||
|
@testable import ElementX
|
||||||
|
import Testing
|
||||||
|
|
||||||
|
struct CurrentValuePublisherTests {
|
||||||
|
@Test
|
||||||
|
func publisherReplaysCurrentValueAndDeliversSubsequentOnes() {
|
||||||
|
let subject = CurrentValueSubject<Int, Never>(1)
|
||||||
|
var received = [Int]()
|
||||||
|
let cancellable = subject.asCurrentValuePublisher().sink { received.append($0) }
|
||||||
|
subject.send(2)
|
||||||
|
subject.send(3)
|
||||||
|
#expect(received == [1, 2, 3])
|
||||||
|
cancellable.cancel()
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
func mappedPublisherChainedInlineDeliversSubsequentValues() {
|
||||||
|
let subject = CurrentValueSubject<Int, Never>(1)
|
||||||
|
var received = [Int]()
|
||||||
|
// The mapped publisher is intentionally chained without being stored: the struct is
|
||||||
|
// discarded mid-chain, which used to cancel an eager map's subscription and stop updates.
|
||||||
|
let cancellable = subject.asCurrentValuePublisher()
|
||||||
|
.map { $0 * 2 }
|
||||||
|
.removeDuplicates()
|
||||||
|
.sink { received.append($0) }
|
||||||
|
subject.send(2)
|
||||||
|
subject.send(3)
|
||||||
|
#expect(received == [2, 4, 6])
|
||||||
|
cancellable.cancel()
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
func mappedPublisherValueStaysCurrent() {
|
||||||
|
let subject = CurrentValueSubject<Int, Never>(1)
|
||||||
|
let mapped = subject.asCurrentValuePublisher().map { $0 * 2 }
|
||||||
|
#expect(mapped.value == 2)
|
||||||
|
subject.send(5)
|
||||||
|
#expect(mapped.value == 10)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user