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())
    }
}

// ==================

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

// ===================

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

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