Maybe I am missing something but "subscriber is gone" block is not going to solve the problem. AsyncNinja is allows you to have any amount of subscribers to your primitive. Looks like invalidation of streamHandler if any single subscriber is gone is just going to ruin a mechanism you are trying to provide. My point is that you need to invalidate streamHandler if all subscribers are gone. This is what
_asyncNinja_notifyFinalization does.
Also releasing producer within finalisation block could lead to retain cycles.
but last subscriber is gone
this should be more productive
there is 1 bug and 1 feature
func merge<Update,Success>(channels: [Channel<Update,Success>]) -> Channel<Update,Void> {
return channel(executor: Executor.primary) { (producer : Producer<Update,Void>) -> Void in
for ch in channels {
ch.bind(producer)
}
}
}
that merge function can merge updates from arbitrary number of channels
for example:
let selection = merge(channels: myView.checkboxes.map { $0.actionChannel() } )
.debounce(interval: 1)
reaction for any of 10 checkboxes
event.flatMap { doSomethingForTwoSeccons() }.onUpdate { * update executes immediately instead of waiting for two seconds*\ }
I added failing test for flatMap
there are two methods
func download(from: String, to: String) -> Channel<Double,URL>func unzip(from: String, to: String, password: String) -> Channel<Double,URL>
each channel updates its progress and completes with destination url
I'm trying to combine sequencially these methods
func prepareSourceFiles() -> Channel<Double,URL> {
return download(from: webURL, to: dstZip)
.map { $0 / 2.0 } // 100% is 50% <= IGNORED
.flatMapSuccess { unzip(from: $0.path, to: dstUnzip, password: "qwerty") }
.map { 0.5 + $0 / 2.0 } // 0% is 50%
}
so downloading progress will be first 50% and unzipping is next 50%
but
in result I get only progress from unzipping while progress for downloading was skipped
does AsyncNinaja has some easy replacement for Rx CombineLatest? http://reactivex.io/documentation/operators/combinelatest.html
I'm developing the CombineLatest
This message was deleted
import Foundation
typealias ES = EventSource
func combineLatestUpdates<ES1:ES, ES2:ES>(_ es1: ES1, _ es2: ES2, executor: Executor = Executor.main)
-> Channel<(ES1.Update,ES2.Update),(ES1.Success, ES2.Success)> {
return CombineLatest2(es1, es2, executor: executor).producer
}
fileprivate class CombineLatest2<ES1:ES, ES2:ES> : ExecutionContext, ReleasePoolOwner {
typealias ResultUpdate = (ES1.Update, ES2.Update)
typealias ResultSuccess = (ES1.Success, ES2.Success)
public let releasePool = ReleasePool()
public var executor: Executor
var update1 : ES1.Update?
var update2 : ES2.Update?
var success1 : ES1.Success?
var success2 : ES2.Success?
let producer = Producer<ResultUpdate,ResultSuccess>()
init(_ es1: ES1, _ es2: ES2, executor: Executor) {
self.executor = executor
producer._asyncNinja_retainUntilFinalization(self)
es1.onUpdate(context: self) { ctx, upd in ctx.update1 = upd; ctx.trySendUpdate() }
.onSuccess(context: self) { ctx, success in ctx.success1 = success; ctx.tryComplete() }
.onFailure(context: self) { ctx, error in ctx.producer.fail(error) }
es2.onUpdate(context: self) { ctx, upd in ctx.update2 = upd; ctx.trySendUpdate() }
.onSuccess(context: self) { ctx, success in ctx.success2 = success; ctx.tryComplete() }
.onFailure(context: self) { ctx, error in ctx.producer.fail(error) }
}
func trySendUpdate() {
guard let upd1 = update1 else { return }
guard let upd2 = update2 else { return }
producer.update((upd1,upd2))
}
func tryComplete() {
guard let suc1 = success1 else { return }
guard let suc2 = success2 else { return }
producer.succeed((suc1,suc2), from: executor)
}
}
this solution works fine for combining elements
but there is a memory leak
producer._asyncNinja_retainUntilFinalization(self)
feature I requested previously could help
it would be handy to release it when last subscription released
it was exact case when proper question contains answer inside
my solution to this problem is to extend ObjCInjectedRetainer and ReleasePoolOwner
extension Retainer {
func combineLatestUpdates<ES1:ES, ES2:ES>(_ es1: ES1, _ es2: ES2, executor: Executor = Executor.main)
-> Channel<(ES1.Update,ES2.Update),(ES1.Success, ES2.Success)> {
return CombineLatest2(es1, es2, executor: executor)
.retain(with: self)
.producer
}
}