Custom Rx Type

NiceSignal.swift

import Foundation
import RxSwift
import RxCocoa

public protocol NiceSignalEventProtocol {
    associatedtype Element

    func asNiceSignalEvent() -> NiceSignalEvent<Element>
}

// TODO: use Either
public enum NiceSignalEvent<Element>: NiceSignalEventProtocol {
    case next(Element)
    case completed(Element)
    case completedEmpty

    var element: Element? {
        switch self {
        case .next(let element), .completed(let element):
            return element
        case .completedEmpty:
            return nil
        }
    }

    var isCompleted: Bool {
        switch self {
        case .next:
            return false
        case .completed, .completedEmpty:
            return true
        }
    }

    func map<Result>(_ f: (Element) -> Result) -> NiceSignalEvent<Result> {
        switch self {
        case .next(let element):
            return .next(f(element))
        case .completed(let element):
            return .completed(f(element))
        case .completedEmpty:
            return .completedEmpty
        }
    }

    public func asNiceSignalEvent() -> NiceSignalEvent<Element> {
        return self
    }
}

public struct NiceSignal<Element> {
    let source: Signal<NiceSignalEvent<Element>>
}

extension NiceSignal {
    public typealias NiceSignalObserver = (NiceSignalEvent<Element>) -> Void

    public static func create(subscribe: @escaping (@escaping NiceSignalObserver) -> Disposable) -> NiceSignal<Element> {
        return NiceSignal<Element>(source: Observable<NiceSignalEvent<Element>>.create { observer in
            return subscribe { event in
                switch event {
                case .next(let element):
                    observer.onNext(.next(element))
                case .completed(let element):
                    observer.onNext(.completed(element))
                    observer.onCompleted()
                case .completedEmpty:
                    observer.onCompleted()
                }
            }
        }.asSignal(onErrorSignalWith: .empty()))
    }

    public func subscribe(_ observer: @escaping NiceSignalObserver) -> Disposable {
        return source.emit(
            onNext: { event in
                switch event {
                case .next(let element):
                    observer(.next(element))
                case .completed(let element):
                    observer(.completed(element))
                case .completedEmpty:
                    break
                }
            }
        )
    }
}

extension NiceSignal: ObservableConvertibleType {
    public func asObservable() -> Observable<Element> {
        return asSharedSequence().asObservable()
    }
}

extension NiceSignal: ObservableType {

    public func subscribe<Observer>(_ observer: Observer) -> Disposable where Observer : ObserverType, Element == Observer.Element {
        return subscribe { event in
            switch event {
            case .next(let element):
                observer.onNext(element)
            case .completed(let element):
                observer.onNext(element)
                observer.onCompleted()
            case .completedEmpty:
                observer.onCompleted()
            }
        }
    }
}

extension PrimitiveSequence {
    func asNiceSignal() -> NiceSignal<Element> {
        return NiceSignal(source:
            self.asSignal(onErrorSignalWith: .empty()).map { .completed($0) }
        )
    }
}

extension ObservableConvertibleType {
    public func asNiceSignal(completeWhen isLast: @escaping (Element) -> Bool) -> NiceSignal<Element> {
        return NiceSignal(source:
            asObservable()
                .map { isLast($0) ? .completed($0): .next($0) }
                .takeUntil(.inclusive, predicate: { $0.isCompleted })
                .asSignal(onErrorSignalWith: .empty())
        )
    }

    public func asNiceSignal(take n: Int) -> NiceSignal<Element> {
        return asObservable().enumerated().asNiceSignal { index, element -> Bool in
            index == n - 1
        }.map { $0.element }
    }
}

extension NiceSignal: SharedSequenceConvertibleType {
    public func asSharedSequence() -> SharedSequence<SignalSharingStrategy, Element> {
        return source.asObservable().compactMap { $0.element }.asSignal(onErrorSignalWith: .empty())
    }
}

// --------------------------- Operations

extension NiceSignal {

    public static func just(_ element: Element) -> NiceSignal<Element> {
        return Single.just(element).asNiceSignal()
    }

    public static func empty() -> NiceSignal<Element> {
        return Observable<Element>.empty().asNiceSignal(completeWhen: { _ in true })
    }

    public func materialize() -> Signal<NiceSignalEvent<Element>> {
        return source
    }

    public func map<Result>(_ selector: @escaping (Element) -> Result) -> NiceSignal<Result> {
        return NiceSignal<Result>(source:
            source.map { $0.map(selector) }
        )
    }

    public func debug(_ identifier: String? = nil, trimOutput: Bool = false, file: String = #file, line: UInt = #line, function: String = #function) -> NiceSignal<Element> {
        return materialize()
            .debug(identifier, trimOutput: trimOutput, file: file, line: line, function: function)
            .dematerialize()
    }
}

extension Signal where Element: NiceSignalEventProtocol {
    public func dematerialize() -> NiceSignal<Element.Element> {
        return NiceSignal(source:
            self.map { $0.asNiceSignalEvent() }.asSignal(onErrorSignalWith: .empty())
        )
    }
}