executor.execute( from: nil) { [weak producer] (originalExecutor) in
do {
let onRelease = try block({ producer?.update($0, from: originalExecutor) }, { producer?.complete($0, from: originalExecutor) })
producer?.onSubsriptionRelease = onRelease // <<- LOOK AT ME
} catch{
producer?.complete(Fallible(failure: error), from: originalExecutor)
}
}
It is something like Rx's OnDispose.
I'm really hope everything I wrote was clear :)
func realmChangset(cancellationToken: CancellationToken) -> Channel<RealmCollectionChange<Results<RepoEntity>>, Void> {
return channel() { producer in
producer.bufferSize = 666
cancellationToken?.add(cancellable: producer) // plan A
let notificationToken = allObjects(ofType: RepoEntity.self).observe { changeset in
producer.update(changeset)
}
producer.onRelease = { notificationToken.invalidate }
// producer.complete(fallible: _)
// producer.succeded(...)
// producer.failed(...)
}.cancelableBy(cancellationToken) // plan B
}
- Please make a pull request with unit test that exposes this issue. I will try to it figure out.
- Please try using internal methods
_asyncNinja_retainUntilFinalization,_asyncNinja_notifyFinalization. They could help. Their names could change if their functionality would be proven useful on larger scale - That should be rather easy to introduce. Please make a pull request for that. See similar functions here: https://github.com/AsyncNinja/AsyncNinja/blob/master/Sources/Channel_makers.swift
examples of what I'm aiming for:
// Realm Changeset
func itemsStream() -> Channel<RealmCollectionChange<Results<RepoEntity>>, Void> {
return channel(executor: Executor.main) { producer in
let notificationToken = self.allObjects(ofType: RepoEntity.self).observe { changeset in
producer.update(changeset)
print("changeset update")
}
producer._asyncNinja_retainUntilFinalization(notificationToken)
}
}
// Input stream
func modifierFlagsStream() -> Channel<NSEvent.ModifierFlags, Void> {
return channel(executor: Executor.main) { producer in
let monitorFlags = NSEvent.addLocalMonitorForEvents(matching: .flagsChanged) {
producer.update($0.modifierFlags)
return $0
}
producer._asyncNinja_notifyFinalization {
NSEvent.removeMonitor(monitorFlags!)
}
}
}public func channel<Update, Success>(executor: Executor = .primary, block: @escaping (_ strongProducer: Producer<Update, Success>) throws -> Void) -> Channel<Update, Success> {
let producer = Producer<Update, Success>(bufferSize: AsyncNinjaConstants.defaultChannelBufferSize)
executor.execute(from: nil) { [weak producer] _ in
guard let producer = producer else { return }
do {
try block(producer)
} catch {
producer.fail(error)
}
}
return producer
}
public func channel<C: ExecutionContext, Update, Success>(context: C, executor: Executor? = nil,
block: @escaping (_ strongProducer: Producer<Update, Success>) throws -> Void) -> Channel<Update, Success> {
let producer = Producer<Update, Success>(bufferSize: AsyncNinjaConstants.defaultChannelBufferSize)
context.addDependent(completable: producer)
(executor ?? context.executor).execute(from: nil) { [weak context, weak producer] (originalExecutor) in
guard let producer = producer else { return }
guard let _ = context else {
producer.cancelBecauseOfDeallocatedContext(from: originalExecutor)
return
}
do {
try block(producer)
} catch {
producer.fail(error)
}
}
return producer
}
But there is a problem with infinite streams
In current implementation of AsyncNinja releasing of producer invokes the Finalization
producer.retainWhileSubscribed(notificationToken) producer.notifySubscriptionRelease { NSEvent.removeMonitor(monitorFlags!) }
It should be able to:
- Update and Complete
- Set up Cancelation
- Set Buffer Size
- Handle Unsubscription
Another use case:
public extension ReactiveProperties where Object: NSView {
func trackingAreaWith(options: NSTrackingArea.Options) -> Channel<NSEvent, Void> {
return channel() { producer in
let receiver = TrackingMessageReceiver(producer: producer)
let area = NSTrackingArea.init(rect: self.object.bounds, options: options, owner: /* weak */ receiver, userInfo: nil)
self.object.addTrackingArea(area)
producer._asyncNinja_retainUntilFinalization(receiver)
producer._asyncNinja_notifyFinalization {
self.base.removeTrackingArea(area)
}
}
}
var trackingArea : Channel<NSEvent, Void> {
return trackingAreaWith(options: [ .mouseEnteredAndExited, .activeAlways ])
}
}
fileprivate class TrackingMessageReceiver : NSResponder {
var producer: Producer<NSEvent, Void>
init(producer: Producer<NSEvent, Void>) {
self.producer = producer
super.init()
}
required init?(coder: NSCoder) {
fatalError("init(coder:) has not been implemented")
}
override func mouseEntered(with theEvent: NSEvent) {
producer.update(theEvent)
}
override func mouseExited(with theEvent: NSEvent) {
producer.update(theEvent)
}
}
BTW. this code will not compile:
'object' is inaccessible due to 'internal' protection level
So I've pushed channel factory co-authored with @serg-vinnie . Also I've pushed publishing of ReactiveProperties methods and initializers, so the issue you've been seeing should be gone
I do not quite understand what unsubscription means. Does it mean that subscriber is gone but the channel is still up and running?
Also I see with your point on
Update and Complete are potentially doable.
However cancellation and buffer creation are not. That would create a chain of underlying channels that would pass events to one another.
Maybe something as simple as builder could work.
ChannelProducerProtocol.Update and Complete are potentially doable.
However cancellation and buffer creation are not. That would create a chain of underlying channels that would pass events to one another.
Maybe something as simple as builder could work.
like onDispose in Rx
generic description of what I need looks like this:
func subscribeToDataSource() -> Channel<DataItem, Void> {
return channel() { producer in
let streamHandler = DataStreamSubscription { nextDataItem in
/* STRONG reference to Producer */
producer.update(nextDataItem)
}
producer.onRelease { streamHandler.invalidate() } // Invalidate streamHandler to release strong reference to the Producer
}
}
Do you understand why it is impossible in current implementation of AsyncNinja?
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)
}
}
}