How to wrap SKPaymentQueue in Rx Observable

Question

Does anyone know how you'd go about wrapping SKPaymentQueue in an Rx Observable? SKPaymentQueue requires that you register an observer that conforms to the SKPaymentTransactionObserver via the SKPaymentQueue.addTransactionObserver method.

So far I've tried the following but it doesn't work. I suspect because the observer proxy is getting deallocated before it's called. It feels like I'm missing some kind of idiomatic Rx trick here:

class StoreService {
    let paymentQueue = SKPaymentQueue.defaultQueue()

    func purchase(product: SKProduct) -> Observable<SKProduct> {
        return Observable.create { (observer) in
            let transactionObserver = PaymentTransactionProxyObserver(observer)

            self.paymentQueue.addTransactionObserver(transactionObserver)

            let payment = SKPayment(product: product)
            self.paymentQueue.addPayment(payment)

            return AnonymousDisposable {
                self.paymentQueue.removeTransactionObserver(transactionObserver)
            }
        }
    }
}

class PaymentTransactionProxyObserver: NSObject, SKPaymentTransactionObserver {

    let observer: AnyObserver<SKPaymentTransaction>

    init(_ observer: AnyObserver<SKPaymentTransaction>) {
        self.observer = observer
    }

    func paymentQueue(queue: SKPaymentQueue, updatedTransactions transactions: [SKPaymentTransaction]) {
        for transaction in transactions {
            observer.onNext(transaction)
        }
        observer.onCompleted()
    }
}

I've also hunted around in the RxCocoa source looking for inspiration however the closest example, the NSNotificationCenter extension, uses a closure as opposed to an observer with multiple callback methods an a register and unregister step. It also doesn't seem to fit the DelegateProxy pattern used by other extensions.

Answer 1

You are right, your proxy is getting deleted too quickly. PaymentTransactionProxyObserver needs to hold a reference to itself to keep itself alive and then nul that reference after it calls onCompleted.

Here's a gist for inspiration. It uses Promises instead of Rx, but it should help you get the idea. When looking at this code think of PromiseKit's fulfill method as observer.onNext; observer.onCompleted and PromiseKit's reject as observer.onError.

https://gist.github.com/dtartaglia/b92163d6055e1a580d493676e5b1e448