Rx Custom transformations

CustomTransform.swift

protocol CustomTransform {
    associatedtype Result

    func apply(_ sink: (Result) -> Void)
}

enum FilterMap<Result>: CustomTransform {
    case filter
    case map(Result)

    func apply(_ sink: (Result) -> Void) {
        switch self {
        case .filter: return
        case .map(let value): sink(value)
        }
    }
}

extension ObservableType {
    func customTransform<T: CustomTransform>(_ transform: @escaping (E) -> T) -> Observable<T.Result> {
        return Observable.create { observer in
            return self.subscribe { event in
                switch event {
                case .next(let element): transform(element).apply { observer.onNext($0) }
                case .completed: observer.onCompleted()
                case .error(let error): observer.onError(error)
                }
            }
        }
    }

    func filterMap<T>(_ transform: @escaping (E) -> FilterMap<T>) -> Observable<T> {
        return customTransform(transform)
    }
}

// Usage: 
Observable.of(1,2,3).filterMap {
   if $0 % 2 == 0 { return .filter }
   return .map(2*$0)
}