combine_async_publisher.swift

class Observer: Subscription {
    public func unsubscribe() {
        ...
    }

    public func request(_ demand: Subscribers.Demand) {
        // We only inform on changes.
        print("Demand for \(demand) received.")
    }
    
    public func cancel() {
        print("Unsubscribed.")
        unsubscribe()
    }
}

    public class MyPublisher: Publisher {
        public typealias Output = [MyThing]
        public typealias Failure = Never
        
        weak var thingManager: ThingManager?
        var subscribers = [Int: Observer]()
        var subscriberIdSeed: Int = 0
        
        init(thingManager: ThingManager) {
            self.thingManager = thingManager
        }
        
        public func receive<S>(subscriber: S) where S : Subscriber, MyPublisher.Failure == S.Failure, MyPublisher.Output == S.Input {
            guard let thingManager = thingManager else { fatalError("Owning manager went out of scope. Should never happen.") }
            subscriberIdSeed += 1
            let subscriberId = subscriberIdSeed
            
            let observer = thingManager.subscribe(dispatchQueue: queue, resultHandler: { things in
                Swift.print("notified.")
                let demand = subscriber.receive(things)
                switch demand {
                case .unlimited:
                    Swift.print("\tunlimited.")
                    break
                case .none:
                    Swift.print("\tnone -> unsubscribed.")
                    self.subscribers[subscriberId] = nil
                    subscriber.receive(completion: .finished)
                default:
                    break
                }
            })
            subscribers[subscriberId] = observer
            subscriber.receive(subscription: observer)
        }
    }

    class TestSubscriber: Subscriber {
        typealias Input = [MyThing]
        
        typealias Failure = Never
        
        internal var things = [MyThing]()
        internal var completed = false

        func receive(subscription: Subscription) {
            print("Subscription started.")
        }

        func receive(_ input: Input) -> Subscribers.Demand {
            print("Data Received \(input).")
            things = input
            return .unlimited
        }

        func receive(completion: Subscribers.Completion<Failure>) {
            print("Subscription completed.")
            completed = true
        }
    }

Anyone know how to correctly implement async Publisher + Subscriber in Combine (so pushes instead of pulls)? I have a ThingManager I can subscribe to and it’ll call a closure I provide. Is this the right way?

In particular, am I handling the result from Subscriber.receive() correctly?

Yes using a closure you provide is the way that I implement this kind of pattern otherwise you need to keep a reference to the caller