compose operator for RxSwift

ObservableType+compose.swift

struct ComposeTransformer<T, R> {
    let transformer: (Observable<T>) -> Observable<R>
    init(transformer: @escaping (Observable<T>) -> Observable<R>) {
        self.transformer = transformer
    }
    
    func call(_ observable: Observable<T>) -> Observable<R> {
        return transformer(observable)
    }
}

extension ObservableType {
    func compose<T>(_ transformer: ComposeTransformer<E, T>) -> Observable<T> {
        return transformer.call(self.asObservable())
    }
}

sample.swift

// ==================
// before
Observable.just([1, 2, 3, 4, 5])
    .map{ $0.reduce(0, +) }
    .subscribeOn(SerialDispatchQueueScheduler(qos: .default))
    .observeOn(MainScheduler.instance)
    .subscribe({ event in
        print(event)
    })


// ==================
// after (1)
Observable.just([1, 2, 3, 4, 5])
    .map{ $0.reduce(0, +) }
    .compose(ComposeTransformer<Int, Int> { observable in
        return observable
            .subscribeOn(SerialDispatchQueueScheduler(qos: .default))
            .observeOn(MainScheduler.instance)
    }).subscribe({ event in
        print(event)
    })

// ===================
// after (2)
let composeTransformer = ComposeTransformer<Int, Int> { observable in
        return observable
            .subscribeOn(SerialDispatchQueueScheduler(qos: .default))
            .observeOn(MainScheduler.instance)
}

Observable.just([1, 2, 3, 4, 5])
    .map{ $0.reduce(0, +) }
    .compose(composeTransformer)
    .subscribe({ event in
        print(event)
    })