CombineExt provides a collection of operators, publishers and utilities for Combine, that are not provided by Apple themselves, but are common in other Reactive Frameworks and standards.
Home Page: https://combine.community
License: MIT License
Hello! I'd just like to preface this issue by saying that I'm quite new to both Combine and reactive programming in general. So, it is possible that this isn't actually an issue.
I've put together a test that reliably crashes in ReplaySubject. Peeking in the code, it looks like there isn't any kind of synchronization, so this doesn't surprise me too much. Here's the example, which simulates an expensive, on-time initialization process, which is then shared by many potential downstream consumers.
let initPublisher = Deferred {
Future<Int, Error> { (promise) in
print("init starting")
sleep(1)
print("init completed")
promise(.success(42))
}
}
.share(replay: 1)
.subscribe(on: queue) // <- this appears necessary
.eraseToAnyPublisher()
let group = DispatchGroup()
var localCancellables = Set<AnyCancellable>()
for i in 0..<20 {
group.enter()
DispatchQueue.global().async {
print("running \(i)")
initPublisher
.map { (value) -> Int in
print("beginning sleep \(i)")
usleep(500)
print("ending sleep \(i)")
return value
}
.sink { (completion) in
print("completion \(i): \(completion)")
group.leave()
} receiveValue: { (result) in
print("result \(i): \(result)")
}
.store(in: &localCancellables)
}
}
group.wait()
print("completed all")As indicated in the code, it looks like the addition of a subscribe(on:) added the necessary synchronization. My question is, is that expected, or should ReplaySubject be thread-safe?
After #105 were merged.
The code working before from my project not anymore.
Here is a gist:
final class ViewModel: ObservableObject {
@Published var singal1 = true
@Published var singal2 = true
}
func testWithPublished() {
var results = [Bool]()
subscription = model.$singal1.withLatestFrom(model.$singal2) {
$0 && $1
}
.sink(receiveValue: { value in
results.append(value)
})
model.singal1 = false
XCTAssertEqual(results, [true, false])
}As my expect, results should be [true, false], instead of just [true]. Am I misunderstand the behavior of withLatestFrom?
If you have an array of unknown size, you expect it to be published once all elements are there.
If you don't have any elements, I'd argue you expect it to publish immediately with an empty array - since that's the natural value it would assume. Otherwise, other subscribers might wait forever on its output.
Just([]).mapError { $0 as! Element.Failure }.eraseToAnyPublisher()
I find both Passthrough and CurrentValue relays pretty useful however they require that the entire stack adopts that convention because there is no way to currently convert an existing Subject to a Relay.
This is useful when you are for example adding a new scene/feature which takes a subject from a previous screen as a way to communicate some user action.
For simplicity here's an example:
class MainVC: UIViewController {
func showNext(doneSubject: PassthroughSubject<Void, Never>) {
let nextVC = NextVC(doneRelay: ????) // <- there is no way to convert a subject to a relay
....
}
}Since the backing storage for both Relay types are their corresponding (private) subjects, it would be useful to add initializers to the Relay types that take an existing subject as the backing storage.
Is this something you'd be willing to accept as a contribution? Let me know and I'd be happy to get a PR ready. AFAICT We should be able to overload the existing initializer so no breaking changes involved either.
Let me know what you think!
Rog
Thanks for CombineExt, I've been using in my apps for a long time.
I recently ran into a performance issue with "Combine" .collect(.byTime). It starts a repeating timer that runs for the entire duration of the subscription. Depending on the time stride and how many of these you have it can really effect performance even when no items are getting published.
The problem that I was trying to solve with .collect(.byTime) is:
If there's a good Combine or CombineExt way to do this I would love to know. Otherwise I have a publisher that does what I need here. The timing logic is quite simple, though I don't really understand the details of the surrounding "Combine" logic. It's mostly copied/pasted from CombineX.
https://forums.swift.org/t/combines-collect-bytime-schedules-a-repeating-timer/57456
Assuming this type of scheduler doesn't already existing I think the basic logic would be a good addition to CombineExt, though probably someone more in tune with combine internals should check and adapt the code to CombineExt.
CFBundleVersion is a required field according to documentation for the AppStore release.
This key is required by the App Store and is used throughout the system to identify the version of the build. For macOS apps, increment the build version before you distribute a build.
As of now release to App Store with an CombineExt failing with a following error:
ERROR [2020-05-15 17:02:14.73]: [31mERROR ITMS-90056: "This bundle Payload/Confluence.app/Frameworks/CombineExt.framework is invalid. The Info.plist file is missing the required key: CFBundleVersion. Please find more information about CFBundleVersion at https://developer.apple.com/documentation/bundleresources/information_property_list/cfbundleversion"
Hi! I find a bag with a relay publishes.
My code:
func test() {
let finished = AnyPublisher<String, Never>.create { subscriber in
subscriber.send(completion: .finished)
return AnyCancellable { }
}
let stream = PassthroughRelay<String>()
finished.subscribe(stream)
.store(in: &cancellables)
stream.sink { str in
print(str)
}.store(in: &cancellables)
stream.accept("sdfsdf")
}
Any try to send info to "stream" is fail.
Can you help plz?
I'm having issues constructing an invocation to Collection.combineLatest when my Publishers have different types. An example:
@Published var s: String
@Published var i: Int
[s, i].combineLatest -> ERROR: Heterogeneous collection literal could only be inferred to '[Any]'; add explicit type annotation if this is intentionalSince the array can't be constrained to [Publisher] - Protocol 'Publisher' can only be used as a generic constraint because it has Self or associated type requirements, none of the Collection publisher functions are available. Is there a workaround for issues like this? Or is it only expected to use the Collection publishers on publisher of the same output type?
Do you think it would be better to rename it to send(_ input:) to follow PassthroughSubject.send(_ input:) and CurrentValueSubject.send(_ input:) naming pattern?
Hi Folks, we recently started using CombineExt in a project and it's already proven itself invaluable for fixing plenty of common scenarios we wanted to solve, so kudos for that.
We had a need to use share(replay:) recently and noticed after implementing it that we had some retain cycles appearing in our application. After spending some time looking into them, it looks to be caused by ReplaySubject not releasing its subscriptions when they receive a completion.
I've got an idea for a fix, but I'm not well versed enough on Combine yet to appreciate what the other implications of it may be, so any help would be appreciated, thanks.
When using combineLatest on a collection with around a thousand elements, a stack overflow happens.
minimal example:
var bag: Set<AnyCancellable> = []
(0 ..< 10_000).map {_ in Just(42)}
.combineLatest()
.sink { _ in
bag.removeAll()
}
.store(in: &bag)Currently, when calling combineLatest() on an empty collection of publishers, an empty publisher is emitted that completes immediately.
Instead, wouldn't it make sense to publish just an empty array instead (Just([]))?
combineLatest() emits the first output when all source publishers have emitted at least one output.
A for-all operator on an empty sequence would return true, which would mean that the combineLatest operator would have to immediately return [] when given an empty collection.
For me, one of the most useful operators from RxSwift has been debug() to quickly log the stream of items passing through an Observable. A while back, I tried my hand at implementing the operator myself in Combine but got a bit lost in terms of writing the Sink and working with the Combine utility classes. I'd be happy to help assist in adding it to this library with a bit of guidance.
The main purpose of relay is to protect sending failure event, while setting the failure type to Never could achieve the same result. Is there something more Relay could do beyond the subject-counterpart, or something I misunderstood?
I've faced an issue with Relays and subscribe() method when values aren't getting sent to the subscribed publisher.
I've created a simple VC to see what's going on.
class ViewController: UIViewController {
let input = PassthroughSubject<Int, Never>()
let output = PassthroughSubject<Int, Never>()
var subsc = Set<AnyCancellable>()
override func viewDidLoad() {
super.viewDidLoad()
input
.subscribe(output)
.store(in: &subsc)
output
.sink { print("Output", $0) }
.store(in: &subsc)
let fiveSecondsFromNow = DispatchTime.now() + 5
DispatchQueue.main.asyncAfter(deadline: fiveSecondsFromNow) { [weak self] in
self?.input.send(10)
}
}
}I can see Output 10 in the console output 5 seconds later, as expected.
However, if I replace PassthroughSubject with PassthroughRelay
class ViewController: UIViewController {
let input = PassthroughRelay<Int>()
let output = PassthroughRelay<Int>()
var subsc = Set<AnyCancellable>()
override func viewDidLoad() {
super.viewDidLoad()
input
.subscribe(output)
.store(in: &subsc)
output
.sink { print("Output", $0) }
.store(in: &subsc)
let fiveSecondsFromNow = DispatchTime.now() + 5
DispatchQueue.main.asyncAfter(deadline: fiveSecondsFromNow) { [weak self] in
self?.input.accept(10)
}
}
}the output will be empty unless I replace input.subscribe(output) with input.sink { [weak self] in self?.output.accept($0) }.
Is it a bug or relays subscribe method doesn't suppose to work the same way as the one in regular subjects?
I was wondering if there was a reason to not include in this (awesome) library some utility methods to simplify usage of Just, Fail and Empty, which requires an eraseToAnyPublisher() in so many scenarios.
I was thinking about something like this
public extension AnyPublisher {
static func just(_ value: Output) -> AnyPublisher<Output, Failure> {
Just(value)
.setFailureType(to: Failure.self)
.eraseToAnyPublisher()
}
static func empty() -> AnyPublisher<Output, Failure> {
Empty()
.setFailureType(to: Failure.self)
.eraseToAnyPublisher()
}
static func fail(_ error: Failure) -> AnyPublisher<Output, Failure> {
Fail(error: error).eraseToAnyPublisher()
}
}this could allow something like
func somePublisherValue() -> AnyPublisher<String, Never> {
.just("hey")
// instead of Just("hey").eraseToAnyPublisher()
}This should also be super-useful in complex flatMaps with AnyPublishers as return values and some guard/let that always returns a Just/Fail
Is it something that could be useful or it's pointless? Am I missing something? Is it already there and I can't see it? :D
Cheers and thanks (as usual) for the awesome work.
Hello,
I have used share(reply:) operator in the past and just now I have encountered a corner case scenario, where it seems not to be working for some reason. I cannot say it is a problem of the share(replay) operator per se since I do not completely understand what and where the problem is.
Here is a simple failing test:
func testShareOperator() {
let expectation = self.expectation(description: "")
let publisher = Just("random-test-string")
let sharedPublisher = publisher.share(replay: 1)
let downstream1 = sharedPublisher
let downstream2 = sharedPublisher
Publishers.Zip(downstream1, downstream2)
.sink { _ in
expectation.fulfill()
}
.store(in: &self.subscriptions)
waitForExpectations(timeout: 1)
}
For some reason the Zip operator "freezes" and does not deliver the expected value.
If the publisher is given a tiny delay, the test passes. I guess a thread hop is needed for the value to get through. However I do not understand why.
let publisher = Just("random-test-string")
.delay(for: 0.00000000001, scheduler: DispatchQueue.main)
Zip is not the only operator that does not work for me. I am facing the same problem if using CombineLatest instead. However the Concatenate operator, for example, does not have the same problem. I am guessing there is a problem with how (concurrently?) the downstream publishers are being fired in the operator. That is really my only clue.
Maybe I am missing something basic and if so, I am sorry for opening the issue. But I have been really struggling to understand what is going on and any help is appreciated!
Best,
Martin
I have a PR that fixes several issues with CombineEXT.xcodeproj. I forked the project to investigate a bug with CurrentValueRelay and could not run tests until I made these changes. Tests do run fine from the command line.
Fixes to get the project and tests to build:
MapTo and MapToTests files.Project cleanup
Carthage and CHANGELOG.MD files.Missing library and test files added to the project:
MapToValue.swiftMapToValueTests.swiftMapToResult.swiftMapToResultTests.swiftEnumerated.swiftEnumeratedTests.swiftThis project is setup with ViewController -> ChildViewController -> GrandChildViewController
The leak occurs only when ChildViewController sinks to the postsSubject
It seems that if a publisher is bound somewhere else, and then that same publisher is used with a withLatestFrom a retain cycle is introduced and Combine internally crashes
Seems like the DemandBuffer doesn't correctly forward completion information upstream if the upstream Subscriber is being used somewhere else which is why we are crashing in WithLatestFrom and share(replay: 1)
I'd love to use things that have been added since 1.2.0, for example ignoreFailure, without relying on main - I'd prefer to stay versioned. A new tag would be super welcome, thanks!
Our project recently updated to Xcode 14, however when we try building the documentation, we are getting the following errors:
We're targeting iOS 15, and I have tried with versions 1.0.0 and 1.8.0 of CombineExt. Running Monterey 12.6. Also just tried the usual stuff of clearing the package cache, Derived Data etc.
I saw you guys posted an issue to the swift repo a little while ago about this issue, but I'm wondering if a fix for this is still in progress or if there is something else I'm missing? apple/swift#60771
Hi, I am not sure if this is a bug with Combine or CombineExt but I have run a cross a super weird case.
the following code:
Timer.publish(every: 1, on: .main, in: .default)
.autoconnect()
.map { _ -> AnyPublisher<Bool, Never> in
return AnyPublisher<Bool, Error> { subscriber in
subscriber.send(true)
subscriber.send(completion: .finished)
return AnyCancellable {}
}
.replaceError(with: false)
// .catch { _ in Just(false) }
// .assertNoFailure()
.eraseToAnyPublisher()
}
.switchToLatest()
.sink { _ in }
.store(in: &subscriptions)
}
Will produce a memory leak, where every second another retain will occour, however if you replace replaceError with either catch or assertNoFailure everything works fine.
I am using the latest version of xcode and iOS, and I could reproduce both in simulator and on a real device
Just like MapMany; it's pretty common for me to have a publisher of an array, and then I want to filter the items inside the array. So now I have to combine a filter inside a map, and a filterMany would be a pretty nice convenience.
If .combineLatest() is used with Publishers which are recieved on multiple threads then the values are not always received.
This is probably a Combine issue and not an issue of this Library.
func testCollectionCombineLatestWithFinishedEvent_threads() {
for x in 0 ... 1000 {
print("--- iteration \(x)")
let first = PassthroughSubject<Int, Never>()
let second = PassthroughSubject<Int, Never>()
let third = PassthroughSubject<Int, Never>()
let fourth = PassthroughSubject<Int, Never>()
let completedExp = expectation(description: "completed")
let values1234 = expectation(description: "1, 2, 3, 4")
let values12345234 = expectation(description: "1, 2, 3, 4, 5, 2, 3, 6")
var results = [[Int]]()
subscription = [first.receive(on: DispatchQueue.global(qos: .userInitiated)),
second.receive(on: DispatchQueue.global(qos: .userInteractive)),
third.receive(on: DispatchQueue.global(qos: .background)),
fourth.receive(on: DispatchQueue.global(qos: .utility))]
.combineLatest()
.sink(receiveCompletion: { _ in completedExp.fulfill() },
receiveValue: {
results.append($0)
if results == [[1, 2, 3, 4]] {
values1234.fulfill()
}
if results == [[1, 2, 3, 4], [5, 2, 3, 4]] {
values12345234.fulfill()
}
})
first.send(1)
second.send(2)
third.send(3)
fourth.send(4)
wait(for: [values1234], timeout: 5)
XCTAssertEqual(results, [[1, 2, 3, 4]])
first.send(5)
wait(for: [values12345234], timeout: 5)
first.send(completion: .finished)
[second, third, fourth].forEach {
$0.send(completion: .finished)
}
wait(for: [completedExp], timeout: 5)
}
}
or
func testCollectionCombineLatestWithFinishedEvent_threads_2() {
for x in 0 ... 1000 {
print("--- iteration \(x)")
let first = PassthroughSubject<Int, Never>()
let second = PassthroughSubject<Int, Never>()
let third = PassthroughSubject<Int, Never>()
let fourth = PassthroughSubject<Int, Never>()
let completedExp = expectation(description: "completed")
let values1234 = expectation(description: "1, 2, 3, 4")
let values12345234 = expectation(description: "1, 2, 3, 4, 5, 2, 3, 6")
var results = [[Int]]()
subscription = [first, second, third, fourth]
.combineLatest()
.sink(receiveCompletion: { _ in completedExp.fulfill() },
receiveValue: {
results.append($0)
if results == [[1, 2, 3, 4]] {
values1234.fulfill()
}
if results == [[1, 2, 3, 4], [5, 2, 3, 4]] {
values12345234.fulfill()
}
})
DispatchQueue.global(qos: .background).async {
first.send(1)
}
DispatchQueue.global(qos: .userInteractive).async {
second.send(2)
}
DispatchQueue.global(qos: .utility).async {
third.send(3)
}
DispatchQueue.global(qos: .userInitiated).async {
fourth.send(4)
}
wait(for: [values1234], timeout: 5)
XCTAssertEqual(results, [[1, 2, 3, 4]])
DispatchQueue.global(qos: .userInitiated).async {
first.send(5)
}
wait(for: [values12345234], timeout: 5)
first.send(completion: .finished)
[second, third, fourth].forEach {
$0.send(completion: .finished)
}
wait(for: [completedExp], timeout: 5)
}
}
when i use withLatestFrom method it will crash app.
could some one help me?
Fatal error: Unexpected state: received completion but do not have subscription: file Combine/Catch.swift, line 279
2021-07-08 18:11:19.869633+0800 Lit Live[31201:239977] Fatal error: Unexpected state: received completion but do not have subscription: file Combine/Catch.swift, line 279
Will there be a Swift 5.6 compatible release?
error: Build input file cannot be found: 'Carthage/Checkouts/CombineExt/Sources/Operators/MapTo.swift' (in target 'CombineExt' from project 'CombineExt')
👋🏽 I’m working on a post about implementing Publisher.withLatestFrom as a composed operator and — in making sure it matched Rx’s implementation — I noticed that Ext’s doesn’t finish when upstream completes (before the operator’s argument emits). Below are snippets with Rx’s and Ext’s current handling of this scenario:
ObservableType.withLatestFrom on a lone upstream completion.import RxSwift
let disposeBag = DisposeBag()
let first = PublishSubject<Int>()
let second = PublishSubject<Int>()
first
.withLatestFrom(second)
.debug()
.subscribe()
.disposed(by: disposeBag)
first.onCompleted()
// Outputs (trimmed):
//
// ```
// -> subscribed
// -> Event completed
// -> isDisposed
// ```Publisher.withLatestFrom’s handling of the same scenario.final class WithLatestFromTests: XCTestCase {
// …
func testWithLatestFromCompletion() {
let subject1 = PassthroughSubject<Int, Never>()
let subject2 = PassthroughSubject<Int, Never>()
var results = [Int]()
var completed = false
subscription = subject1
.withLatestFrom(subject2)
.sink(
receiveCompletion: { _ in completed = true },
receiveValue: { results.append($0) }
)
subject1.send(completion: .finished)
XCTAssertTrue(completed) // ❌ “`XCTAssertTrue` failed.”
XCTAssertTrue(results.isEmpty)
}
// Similarly on analog tests for other arities.
}Should we update the package to match Rx’s behavior? If so, I’ve got a sketch that passes an updated test suite with these scenarios over on a fork that I’d be happy to PR, but since the WithLatestFrom conformance is already in place, I wanted to check first to see if updating it is easier. 🍻
Xcode 12.5 errors on loading a package with nested 'include' directories.
As CombineExt already implements flatMapLatest, I think it might be a good idea to add concatMap as well.
What do you think? I might look into it on a weekend if we agree on adding it. 😊
Hi everyone,
lately I faced a use case where a reducer function that I pass to a “scan” operator could return an optional.
the expected behaviour was: if the function returns nil then there is no change made in the accumulator and the scan operator continues with the next values.
the signature would be something like:
compactScan(initial: A, _ nextPartialResult: (A, Output) -> A?) -> AnyPublisher<A, Failure>
What do you think?
Hi,
I'm still using Cocoapods, and the latest version is 1.2.0. I see there has been some operators added since (especialy the toggle operator, that I was looking for)
I don't know how SPM handles version updates, but Cocoapods needs tags for tracking the versions, so my question is:
Is there any plan to create an update that includes all new operators or are you waiting to have enough or something?
var cancellable = publiser. handleEvents(receiveCancel: {
Swift.print("cancel before materialise")
})
.materialize()
.handleEvents(receiveCancel: {
Swift.print("cancel before compact")
}.sink{ _ in }
cancellable.cancel()
prints
cancel before compact
expected
cancel before compact
cancel before materialise
let one = PassthroughRelay<String>()
let two = CurrentValueRelay<String>("0")
one.subscribe(two).store(in: &cancellables)
two.sink { data in
print(data)
}.store(in: &cancellables)
one.accept("1")
one.accept("2")
one.accept("3")
if I replace Relay on Subject It is working fine
Now RetryWhen subscribes to upstream twice. This is because the Sink itself subscribes to the upstream in .init.
Test testSuccessfulRetry() checks the number of subscriptions, but not correctly. It should check that the error publisher produces at least one item.
The correct test should look like this:
func testSuccessfulRetry() {
var times = 0
var retriesCount = 0
var expectedOutput: Int?
var completion: Subscribers.Completion<RetryWhenTests.MyError>?
subscription = Deferred(createPublisher: { () -> AnyPublisher<Int, MyError> in
defer { times += 1 }
if times == 0 {
return Fail<Int, MyError>(error: MyError.someError).eraseToAnyPublisher()
}
else {
return Just(5).setFailureType(to: MyError.self).eraseToAnyPublisher()
}
})
.retryWhen { error in
error
.handleEvents(receiveOutput: { _ in retriesCount += 1})
.map { _ in }
}
.sink(
receiveCompletion: { completion = $0 },
receiveValue: { expectedOutput = $0 }
)
XCTAssertEqual(
expectedOutput,
5
)
XCTAssertEqual(completion, .finished)
XCTAssertEqual(times, 2)
XCTAssertEqual(retriesCount, 1)
}
A possible solution is to remove upstream subscription from Sink.init and setting upstreamIsCancelled cancelUpstream function.
private let lock = NSRecursiveLock()
/// ...
func cancelUpstream() {
lock.lock()
guard !upstreamIsCancelled else {
lock.unlock()
return
}
upstreamIsCancelled = true
lock.unlock()
upstreamSubscription.kill()
}
Just came across this library. Many thanks for all the contributions it’s super useful!
I’m looking for info on what I’ve always called 'replace'. I have some version of the following in a bunch of projects:
extension Publisher {
public func replace<T>(_ transform: @autoclosure @escaping () -> T) -> Publishers.Map<Self, T> {
map { _ in transform() }
}
}All it is is a map where you don’t have to manually ignore the transform argument. But it cleans things up just a little in a bunch of places, especially with the @autoclosure allowing a constant value to be passed in. Eg:
// Before
Navigation.Show
.subscribe()
.filter(\.destination.isModal)
.map { _ in Sidebar.Hide(delay: 0.25) }
// After
Navigation.Show
.subscribe()
.filter(\.destination.isModal)
.replace(Sidebar.Hide(delay: 0.25))Is there any precedent for this in other reactive libraries? Or is it pointless sugar?
Thanks for your efforts on this set of useful features.
There's one thing that IMO should work slightly different. Although it would be a breaking change, I still wanted to bring attention to it.
The Collection.zip() operator returns Empty publisher in case the collection is empty (source). It means that the downstream methods like map, flatMap, etc. will not be called. And the receiveValue closure on sink() will not be called as well. My expectation is that those methods should still be called with an empty array passed. See the following example:
var cancelBag = Set<AnyCancellable>()
let publishers: [AnyPublisher<Int, Never>] = [
// Just(1).eraseToAnyPublisher(),
// Just(2).eraseToAnyPublisher()
]
publishers.zip()
.map { results -> Int in
// return the number of results. It could be a number of loaded items, etc.
print("Map results")
return results.count // will not be called :(
}
.sink { results in
print("Got \(results)") // will not be called :(
}
.store(in: &cancelBag)So, I'd changed
case 0:
return Empty().eraseToAnyPublisher()to
case 0:
return Just([]).setFailureType(to: Element.Failure.self).eraseToAnyPublisher()Please let me know your thoughts.
Hi folks 👋
Since Swift Stdlib supports reduce(_:_:) and reduce(into:_:), how about introducing the Combine version of reduce(into:_:) into this library?
reduce(_:_:), like the following:public extension Publisher {
func reduce<Result>(into initialResult: Result, _ nextPartialResult: @escaping (inout Result, Output) -> Void) -> Publishers.Reduce<Self, Result> {
reduce(initialResult) { initial, nextPartial -> Result in
var result = initial
nextPartialResult(&result, nextPartial)
return result
}
}
}Publishers.ReduceInto<Upstream, Output>.Which one do you think makes more sense? 😊
Personally, since the wrapper does not mimic reduce(into:_:) implementation from Stdlib, because it is just a wrapper around reduce(_:_:), I'm leaning towards creating a new operator.
Hi everyone,
It would be great if a Combine implementation of the DelaySubscription operator could be added to this repo in order to further reduce the number of missing operators when compared with other reactive libraries (eg. RxSwift as per the cheat sheet).
As we'd expect from existing implementations found in other libraries, this DelaySubscription operator would allow a subscription to be time-shifted. This means that rather than shifting all emissions of a publisher forward in time (as per the current Delay implementation), only the subscription would be delayed.
Using a real-world example of a live video streaming app to explain the desired difference, if the streaming source was emitting frames (a "hot" publisher), then using .delay(for: .seconds(3), scheduler: scheduler) would mean that when sinked on, frames would be received 3 seconds behind the actual emission. By using .delaySubscription(for: .seconds(3), scheduler: scheduler) this would mean that after the initial 3 second delay (where frames would be dropped), frames would then be received in real time.
Example:
func foo(publishers: [AnyPublisher<String, Never>]) {
publishers
.combineLatest()
.sink { strings in print(strings) }
.store(in: &cancellables)
}Nothing will be printed if publishers is an empty array.
Feels like the correct behavior is to emit an empty array of String in this case. The caller doesn't need a special handling of an empty input case then.
Hello,
I'm currently using a SinglePublisher subprotocol of Publisher, which comes with the guarantee that a publisher publishes exactly one element, or an error.
https://gist.github.com/groue/6226a6a2cde190f819b37aa58ff9dd36
Quoting the doc:
/// The protocol for "single publishers", which publish exactly one element, or
/// an error.
///
/// `Just`, `Future` and `URLSession.DataTaskPublisher` are examples of
/// publishers that conform to `SinglePublisher`.
///
/// Conversely, `Publishers.Sequence` is not a single publisher, because not all
/// sequences contain a single element.
///
/// # SinglePublisher Benefits
///
/// Once you have a publisher that conforms to `SinglePublisher`, you have
/// access to two desirable tools:
///
/// - A `AnySinglePublisher` type that hides details you don’t want to expose
/// across API boundaries. For example, the user of the publisher below knows
/// that it publishes exactly one `String`, no more, no less:
///
/// func namePublisher() -> AnySinglePublisher<String, Error>
///
/// You build an `AnySinglePublisher` with the
/// `eraseToAnySinglePublisher()` method:
///
/// mySinglePublisher.eraseToAnySinglePublisher()
///
/// - A `sinkSingle(receiveResult:)` method that simplifies handling of single
/// publisher results:
///
/// namePublisher().sinkSingle { result in
/// switch result {
/// case let .success(name): print(name)
/// case let .failure(error): print(error)
/// }
/// }
I was wondering if you would be interested in including such a feature in CombineExt.
While there's no switchToFirst operator in Combine, I find it very challenging to implement flatMapFirst like operator from scratch, I'd suggest to start building this operator which's really useful for many use cases
In many places in my codebase, I find myself transforming an error value into a specific kind of output value, like this:
somePublisher.catch({ error in Just(someValueFromError(error)) })I'd love to be able to have some operator that makes this a little nicer to use that omits needing to worry about combining catch and Just in this way, something like this perhaps:
somePublisher.catchToOutput({ someValueFromError($0) })
// also can be used without a closure:
somePublisher.catchToOutput(someValueFromError)The name could be different, I can see perhaps naming such an operator as mapErrorToOutput or catchAsOutput or something as well. Is this something CombineExt would find as a worthwhile addition?
I was looking at for something like RxSwift's ifEmpty(switchTo:), and there appeared to be a solution in
rxswift-to-combine-cheatsheet, but current versions of Combine do not offer replaceEmpty(with publisher: Publisher). I'm thinking something like the below would work, but what do you think? If it's okay, I'll go ahead and make a pull request.
public extension Publisher where Output: Collection {
func replaceEmpty(with publisher: AnyPublisher<Output, Failure>) -> AnyPublisher<Output, Failure> {
flatMap { (output: Output) -> AnyPublisher<Output, Failure> in
if output.isEmpty { return publisher }
return Just(output)
.setFailureType(to: Failure.self)
.eraseToAnyPublisher()
}
.eraseToAnyPublisher()
}
}Hello folks,
We have found a memory leak issue in withLatestFrom operator which is very similar to the memory leak issue found here in ShareReplay.
For better understanding, here is the code snippet
weak var weakSubject1: PassthroughSubject<Int, Never>?
weak var weakSubject2: PassthroughSubject<[Int], Never>?
autoreleasepool {
let subject1 = PassthroughSubject<Int, Never>()
let subject2 = PassthroughSubject<[Int], Never>()
weakSubject1 = subject1
weakSubject2 = subject2
let subscritpion = subject1
.withLatestFrom(subject2)
.map { (a) -> AnyPublisher<Void, Never> in
CurrentValueSubject<Void, Never>(()).eraseToAnyPublisher()
}
.sink { _ in
}
subject2.send([])
subject1.send(1)
subject2.send([])
}
weakSubject2 == nil // destroyed as expected.
weakSubject1 == nil // should have been destroyed but not
Expected:
subject1 and subject2 must be destroyed after the autorelease pool got drained
Actual:
subject1 is held in memory and becomes a zombie even after autorelease pool it belonged to was drained.
Solution:
It is very similar to how the share replay memory leak is taken care in this PR
I am happy to put up a PR to fix the issue. Please let me know if there is a better way to handle this.
I can't show it in debug but based on the logs in analytics. The crash is just in this place
Crashed: com.apple.main-thread
0 libsystem_platform.dylib 0x1823f2598 _os_unfair_lock_recursive_abort + 36
1 libsystem_platform.dylib 0x1823eef40 _os_unfair_lock_lock_slow + 272
2 Combine 0x1ac9c1038 Publishers.ReceiveOn.Inner.receive(completion:) + 468
3 Combine 0x1ac9c13f0 protocol witness for Subscriber.receive(completion:) in conformance Publishers.ReceiveOn<A, B>.Inner<A1> + 20
4 Combine 0x1ac99125c AnySubscriberBox.receive(completion:) + 36
5 Combine 0x1ac9923a4 protocol witness for Subscriber.receive(completion:) in conformance AnySubscriber<A, B> + 28
6 - 0x100d2b3e0 DemandBuffer.flush(adding:) + 109 (DemandBuffer.swift:109)
7 - 0x100d2ae34 DemandBuffer.complete(completion:) + 69 (DemandBuffer.swift:69)
8 - 0x100d334b0 Sink.receive(completion:) + 75 (Sink.swift:75)
9 - 0x100d29ee4 CurrentValueRelay.Subscription.forceFinish() + 73 (CurrentValueRelay.swift:73)
10 - 0x100d2a478 partial apply for closure #1 in CurrentValueRelay.deinit + 4344882296 (<compiler-generated>:4344882296)
11 - 0x100d2a49c partial apply for thunk for @callee_guaranteed (@guaranteed CurrentValueRelay<A>.Subscription<CurrentValueSubject<A, Never>, AnySubscriber<A, Never>>) -> (@error @owned Error) + 4344882332 (<compiler-generated>:4344882332)
12 libswiftCore.dylib 0x18fcad574 Sequence.forEach(_:) + 424
13 - 0x100d29d3c CurrentValueRelay.deinit + 4344880444 (<compiler-generated>:4344880444)
14 - 0x100d29d7c CurrentValueRelay.__deallocating_deinit + 49 (CurrentValueRelay.swift:49)
15 libswiftCore.dylib 0x18fe5fc40 _swift_release_dealloc + 28
16 Combine 0x1aca15534 PublisherBox.__deallocating_deinit + 36
17 libswiftCore.dylib 0x18fe5fc40 _swift_release_dealloc + 28
18 Combine 0x1ac9c2ce0 outlined destroy of Publishers.ReceiveOn<A, B>.Inner<A1>.State + 216
19 Combine 0x1ac9c37e0 outlined assign with take of Publishers.ReceiveOn<A, B>.Inner<A1>.State + 116
20 Combine 0x1ac9c0108 Publishers.ReceiveOn.Inner.cancel() + 616
This object is used in several closures within the viewModel. But everywhere I use a weak link and unwrapping self
//ViewModel
private let stateSubject = CurrentValueRelay<State>(.idle)
private(set) lazy var output: Output = {
return Output(
state: stateSubject.eraseToAnyPublisher()
)
}()//ViewController
output.state
.receive(on: DispatchQueue.main)
.sink(receiveValue: { [weak self] state in
self?.handle(state)
})
.store(in: &disposables)1.Relay only sends a finishing event on deallocation.
but using subscribe<R: Relay>(_ relay: R), the storage subject in the relay received a finishing event.
func testSubscribePublisher() {
var completed = false
relay?
.sink(receiveCompletion: { _ in completed = true },
receiveValue: { self.values.append($0) })
.store(in: &subscriptions)
["1", "2", "3"]
.publisher
.subscribe(relay!)
.store(in: &subscriptions)
XCTAssertFalse(completed)
XCTAssertEqual(values, ["initial", "1", "2", "3"])
// not working, storage subject in the relay is finished
relay!.accept("4")
XCTAssertFalse(completed)
XCTAssertEqual(values, ["initial", "1", "2", "3", "4"])
}
testSubscribeRelay_CurrentValueToPassthrough test case does not. func testSubscribeRelay_CurrentValueToPassthrough() {
var completed = false
let input = CurrentValueRelay<String>("initial")
let output = PassthroughRelay<String>()
input
.subscribe(output)
.store(in: &subscriptions)
output
.sink(receiveCompletion: { _ in completed = true },
receiveValue: { self.values.append($0) })
.store(in: &subscriptions)
input.accept("1")
input.accept("2")
input.accept("3")
XCTAssertFalse(completed)
// need to replace with "XCTAssertEqual(values, ["1", "2", "3"])"
XCTAssertEqual(values, ["initial", "1", "2", "3"])
}
I've just received a huge amount of crash reports caused by an exception during theDemandBuffer deallocation on iOS 13.
CrashReporter Key: fdb68c604e035dc1461eb2f1d3466cd19894c442
Hardware Model: iPhone8,4
Role: Foreground
OS Version: iOS 13.5.1
Exception Type: EXC_BREAKPOINT
Exception Subtype: KERN_INVALID_ADDRESS
EXC_BREAKPOINT: BUG IN CLIENT OF LIBPLATFORM: Trying to recursively lock an os_unfair_lock
0 libsystem_platform.dylib __os_unfair_lock_recursive_abort
1 libsystem_platform.dylib __os_unfair_lock_lock_slow
2 Combine Publishers.ReceiveOn.Inner.receive(completion:)
3 Combine protocol witness for Subscriber.receive(completion:) in conformance Publishers.ReceiveOn<A, B>.Inner<A1>
4 Combine AnySubscriberBox.receive(completion:)
5 Combine protocol witness for Subscriber.receive(completion:) in conformance AnySubscriber<A, B>
6 HomeCodes flush (DemandBuffer.swift:111:24)
7 HomeCodes complete (DemandBuffer.swift:70:13)
8 HomeCodes receive (Sink.swift:75:20)
9 HomeCodes receive (CurrentValueRelay.swift:96:19)
10 HomeCodes partial apply for closure #1 in CurrentValueRelay.deinit (CurrentValueRelay.swift:55:36)
11 HomeCodes partial apply for thunk for @callee_guaranteed (@guaranteed CurrentValueRelay<A>.Subscription<CurrentValueSubject<A, Never>, AnySubscriber<A, Never>>) -> (@error @owned Error) (<compiler-generated>)
12 libswiftCore.dylib Sequence.forEach(_:)
13 HomeCodes deinit (CurrentValueRelay.swift:55:23)
14 HomeCodes deinit (CurrentValueRelay.swift)
15 libswiftCore.dylib __swift_release_dealloc
16 Combine PublisherBox.__deallocating_deinit
17 libswiftCore.dylib __swift_release_dealloc
18 Combine outlined destroy of Publishers.ReceiveOn<A, B>.Inner<A1>.State
19 Combine protocol witness for Cancellable.cancel() in conformance Publishers.ReceiveOn<A, B>.Inner<A1>
20 Combine Subscribers.Sink.cancel()
21 Combine protocol witness for Cancellable.cancel() in conformance Subscribers.Sink<A, B>
22 Combine partial apply
23 Combine AnyCancellable.__deallocating_deinit
24 libswiftCore.dylib __swift_release_dealloc
25 libswiftCore.dylib _swift_arrayDestroy
26 libswiftCore.dylib _SetStorage.deinit
27 libswiftCore.dylib _SetStorage.__deallocating_deinit
28 libswiftCore.dylib __swift_release_dealloc
29 libobjc.A.dylib object_cxxDestructFromClass(objc_object*, objc_class*)
30 libobjc.A.dylib _objc_destructInstance
31 libobjc.A.dylib __objc_rootDealloc
32 UIKitCore -[UIResponder dealloc]
33 UIKitCore -[UIViewController dealloc]
34 HomeCodes deinit (<compiler-generated>)
35 CoreFoundation ___RELEASE_OBJECTS_IN_THE_ARRAY__
36 CoreFoundation -[__NSArrayM dealloc]
37 UIKitCore -[UIViewController dealloc]
38 UIKitCore -[UINavigationController dealloc]
39 HomeCodes deinit (BaseCoordinator.swift)
40 HomeCodes deinit (NavigationCoordinator.swift)
41 HomeCodes deinit (BaseNavigationCoordinator.swift:32:5)
42 HomeCodes deinit (BaseNavigationCoordinator.swift)
43 libswiftCore.dylib __swift_release_dealloc
44 libswiftCore.dylib bool swift::RefCounts<swift::SideTableRefCountBits>::doDecrement<(swift::PerformDeinit)1>(unsigned int)
45 libswiftCore.dylib _swift_arrayDestroy
46 HomeCodes specialized Array.replaceSubrange<A>(_:with:) (HomeCodes)
47 HomeCodes BaseCoordinator.removeChildrenIfNeeded() (<compiler-generated>)
48 HomeCodes removeChildrenIfNeeded (<compiler-generated>)
49 HomeCodes childTransitionCompleted (Coordinator.swift:110:9)
50 HomeCodes childTransitionCompleted (<compiler-generated>)
51 HomeCodes partial apply for closure #1 in BaseCoordinator.registerParent(_:) (BaseCoordinator.swift:106:26)
52 HomeCodes removeChildrenIfNeeded (BaseCoordinator.swift:75:9)
53 HomeCodes removeChildrenIfNeeded (<compiler-generated>)
54 HomeCodes childTransitionCompleted (Coordinator.swift:110:9)
55 HomeCodes childTransitionCompleted (<compiler-generated>)
56 HomeCodes navigationController (NavigationAnimationDelegate.swift:136:22)
57 HomeCodes navigationController (<compiler-generated>)
58 UIKitCore -[UINavigationController navigationTransitionView:didEndTransition:fromView:toView:]
59 UIKitCore -[UINavigationTransitionView _notifyDelegateTransitionDidStopWithContext:]
60 UIKitCore -[UINavigationTransitionView _cleanupTransition]
61 UIKitCore +[UIView(UIViewAnimationWithBlocks) conditionallyAnimate:withAnimation:layout:completion:]
62 UIKitCore -[UINavigationTransitionView transition:fromView:toView:]
63 UIKitCore -[UINavigationController _startTransition:fromViewController:toViewController:]
64 UIKitCore -[UINavigationController _startDeferredTransitionIfNeeded:]
65 UIKitCore -[UINavigationController __viewWillLayoutSubviews]
66 UIKitCore -[UILayoutContainerView layoutSubviews]
67 UIKitCore -[UIView(CALayerDelegate) layoutSublayersOfLayer:]
68 QuartzCore -[CALayer layoutSublayers]
69 QuartzCore CA::Layer::layout_if_needed(CA::Transaction*)
70 UIKitCore -[UIView(Hierarchy) layoutBelowIfNeeded]
71 UIKitCore -[_UISheetPresentationController _sheetLayoutInfoLayout:]
72 UIKitCore -[_UISheetLayoutInfo _layout]
73 UIKitCore ___54-[_UISheetPresentationController transitionWillBegin:]_block_invoke_2
74 UIKitCore +[UIView(Animation) performWithoutAnimation:]
75 UIKitCore ___54-[_UISheetPresentationController transitionWillBegin:]_block_invoke.341
76 UIKitCore -[_UIViewControllerTransitionCoordinator _applyBlocks:releaseBlocks:]
77 UIKitCore -[_UIViewControllerTransitionContext __runAlongsideAnimations]
78 UIKitCore ___63+[UIView(Animation) _setAlongsideAnimations:toRunByEndOfBlock:]_block_invoke
79 UIKitCore -[UIViewAnimationState _runAlongsideAnimations]
80 UIKitCore -[UIViewAnimationState pop]
81 UIKitCore +[UIViewAnimationState popAnimationState]
82 UIKitCore +[UIView(UIViewAnimationWithBlocks) _setupAnimationWithDuration:delay:view:options:factory:animations:start:animationStateGenerator:completion:]
83 UIKitCore +[UIView(UIViewAnimationWithBlocks) animateWithDuration:delay:options:animations:completion:]
84 UIKitCore ___50-[UITransitionView _startTransition:withDuration:]_block_invoke.169
85 UIKitCore +[UIView(UIViewAnimationWithBlocks) conditionallyAnimate:withAnimation:layout:completion:]
86 UIKitCore -[UITransitionView _startTransition:withDuration:]
87 UIKitCore -[UITransitionView transition:fromView:toView:removeFromView:]
88 UIKitCore -[UIViewControllerBuiltinTransitionViewAnimator animateTransition:]
89 UIKitCore ____UIViewControllerTransitioningRunCustomTransition_block_invoke_2
90 UIKitCore +[UIInputResponderController _pinInputViewsForInputResponderController:onBehalfOfResponder:duringBlock:]
91 UIKitCore ____UIViewControllerTransitioningRunCustomTransition_block_invoke.645
92 UIKitCore +[UIView(Animation) _setAlongsideAnimations:toRunByEndOfBlock:]
93 UIKitCore __UIViewControllerTransitioningRunCustomTransition
94 UIKitCore ___56-[UIPresentationController runTransitionForCurrentState]_block_invoke.465
95 UIKitCore __runAfterCACommitDeferredBlocks
96 UIKitCore __cleanUpAfterCAFlushAndRunDeferredBlocks
97 UIKitCore __afterCACommitHandler
98 CoreFoundation ___CFRUNLOOP_IS_CALLING_OUT_TO_AN_OBSERVER_CALLBACK_FUNCTION__
99 CoreFoundation ___CFRunLoopDoObservers
100 CoreFoundation ___CFRunLoopRun
101 CoreFoundation _CFRunLoopRunSpecific
102 GraphicsServices _GSEventRunModal
103 UIKitCore _UIApplicationMain
104 HomeCodes main (main.swift:8:1)
105 libdyld.dylib _start
I've tried to repro it and it looks like it just enough to create a publisher that uses DemandBuffer (e.g. CurrentValueRelay), initialise it and the just deinit it. And it looks like it only reproducible on iOS 13.5 and older.
Did anyone else had the same issue?
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.