Observable.swift

//: Playground - noun: a place where people can play

import Foundation

protocol ObserverType {
    associatedtype Element
    
    func onNext(_ element: Element)
    func onComplete()
    func onError(_ error: Error)
}


final class AnyObserver<E>: ObserverType {
    typealias Element = E
    
    let _onNext: (Element) -> Void
    let _onComplete: () -> Void
    let _onError: (Error) -> Void
    
    init<O: ObserverType>(_ observer: O) where O.Element == Element {
        _onNext = observer.onNext
        _onComplete = observer.onComplete
        _onError = observer.onError
    }
    
    func onNext(_ element: Element) {
        return _onNext(element)
    }
    
    func onComplete() {
        return _onComplete()
    }
    
    func onError(_ error: Error) {
        return _onError(error)
    }
}

final class Observer<E>: ObserverType {
    typealias Element = E
    
    let _onNext: ((Element) -> Void)?
    let _onComplete: (() -> Void)?
    let _onError: ((Error) -> Void)?
    
    init(onNext: ((Element) -> Void)? = nil, onComplete: (() -> Void)? = nil, onError: ((Error) -> Void)? = nil) {
        _onNext = onNext
        _onComplete = onComplete
        _onError = onError
    }
    
    func onNext(_ element: Element) {
        _onNext?(element)
    }
    
    func onComplete() {
        _onComplete?()
    }
    
    func onError(_ error: Error) {
        _onError?(error)
    }
}

protocol CancellableType {
    func cancel()
}

struct Cancellable: CancellableType {
    let _cancel: (() -> Void)?
    
    init(_ cancel: (() -> Void)? = nil) {
        _cancel = cancel
    }
    
    func cancel() {
        _cancel?()
    }
}

protocol ObservableType {
    associatedtype Element
    
    func subscribe<O: ObserverType>(_ observer: O) -> CancellableType where O.Element == Element
}

class Observable<E> : ObservableType {
    typealias Element = E
    
    func subscribe<O: ObserverType>(_ observer: O) -> CancellableType where O.Element == Element {
        fatalError("not implemented")
    }
    
    static func create(_ subscribe: @escaping (AnyObserver<Element>) throws -> CancellableType) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }
}

class AnonymousObservable<E>: Observable<E> {
    private let _subscribe: (AnyObserver<Element>) throws -> CancellableType
    
    init(_ subscribe: @escaping (AnyObserver<Element>) throws -> CancellableType) {
        _subscribe = subscribe
    }
    
    override
    func subscribe<O: ObserverType>(_ observer: O) -> CancellableType where O.Element == Element {
        do {
            return try _subscribe(AnyObserver(observer))
        }
        catch let error {
            observer.onError(error)
            return Cancellable()
        }
    }
}

class RetryObservable<E>: Observable<E> {
    private var retryCount: Int
    private let wrappedObservable: Observable<E>

    init(retryCount: Int, wrappedObservable: Observable<E>) {
        self.retryCount = retryCount
        self.wrappedObservable = wrappedObservable
    }

    override
    func subscribe<O: ObserverType>(_ observer: O) -> CancellableType where O.Element == Element {
        var subscription: CancellableType!
        var handler: Observer<E>!
        
        handler = Observer<E>(
            onNext: observer.onNext,
            onComplete: observer.onComplete,
            onError: { error in
                guard self.retryCount > 0 else {
                    observer.onError(error)
                    return
                }
                
                self.retryCount -= 1
                subscription?.cancel()
                subscription = self.wrappedObservable.subscribe(handler)
            }
        )
        
        subscription = wrappedObservable.subscribe(handler)
        
        return Cancellable {
            subscription?.cancel()
        }
    }
}

// Usage example

enum MyError: Error {
    case SomethingWentWrong
    case Oops
}

var count = 0

let observable = Observable<Int>.create { observer in
    count += 1
    if count < 3 {
        throw MyError.Oops
    }
    
    observer.onNext(count)

    return Cancellable {
        print("cancel")
    }
}

let retryObservable = RetryObservable(retryCount: 5, wrappedObservable: observable)

let observer = Observer<Int>(
    onNext: { print("next:", $0) },
    onComplete: { print("complete") },
    onError: { print("error:", $0 )}
)

let subscription = retryObservable.subscribe(observer)
subscription.cancel()