gistfile1.txt

protocol Observer {
  associatedtype Value

  func next(_ value: Value) -> Void
}

class Subscription {
  init(_ unsubscribe: (() -> Void)? = nil) {
    _unsubscribe = unsubscribe
  }
  func unsubscribe() {
    _unsubscribe?()
    _unsubscribe = nil
  }

  private var _unsubscribe: (() -> Void)?
}

class Observable<T> {
  typealias Value = T

  let creator: (AnyObserver<Value>) -> Subscription
  init(_ creator: @escaping (AnyObserver<Value>) -> Subscription) {
    self.creator = creator
  }

  func subscribe<O: Observer>(_ observer: O) -> Subscription where O.Value == T {
    let observer = AnyObserver(observer: observer)
    listeners.add(observer)

    if listeners.count == 1 {
      // Is down-stream reference, so must be weak.
      subscription = creator(AnyObserver { [weak self] in self?.dispatchNext($0) })
    }

    return Subscription({
      self.listeners.remove(observer)

      if self.listeners.count == 0 {
        self.subscription?.unsubscribe()
      }
    })
  }

  func subscribe(next: @escaping (Value) -> Void) -> Subscription {
    let observer = AnyObserver<Value>(next)
    return subscribe(observer)
  }

  private func dispatchNext(_ value: Value) {
    for listener in listeners {
      (listener as! AnyObserver<Value>).next(value)
    }
  }

  private var listeners = NSMutableOrderedSet()
  private var subscription: Subscription?
}

// Swift-specific

class AnyObserver<T>: Observer {
  typealias Value = T

  private let _next: (Value) -> Void
  init<O: Observer>(observer: O) where O.Value == Value {
    _next = { observer.next($0) }
  }

  init(_ next: @escaping (Value) -> Void) {
    _next = next
  }

  func next(_ value: Value) {
    _next(value)
  }
}