Stream prototype in swift

stream.swift

/**
 An Observable emits values to its subscribed observers.

 A minimal implementation based upon the reactivex specification:
 http://reactivex.io/documentation/observable.html
 */
public class Observable<Value> {
  /** Add a new observer. The provided instance will receive all values provided to onNext. */
  public func subscribe(_ observer: @escaping (Value) -> Void) -> Observable<Value> {
    observers.append(observer)
    return self
  }

  /** Sends a new value to all observers. */
  public func onNext(value: Value) {
    observers.forEach { observer in
      observer(value)
    }
  }

  private var observers: [(Value) -> Void] = []
}

extension Observable {
  /** Transform the items emitted by an Observable by applying a function to each item. */
  public func map<T>(_ transform: @escaping (Value) -> T) -> Observable<T> {
    let downstream = Observable<T>()

    // Keep a strong reference to the parent (self), but a weak reference to the downstream. This
    // ensures that a reference to a downstream node will keep the entire stream alive.
    subscribe { [weak downstream] in
      let _ = self
      downstream?.onNext(value: transform($0))
    }
    return downstream
  }

  /** Emit only those items from an Observable that pass a test. */
  public func filter(_ isIncluded: @escaping (Value) -> Bool) -> Observable<Value> {
    let downstream = Observable<Value>()

    // Keep a strong reference to the parent (self), but a weak reference to the downstream. This
    // ensures that a reference to a downstream node will keep the entire stream alive.
    subscribe { [weak downstream] in
      let _ = self
      if isIncluded($0) {
        downstream?.onNext(value: $0)
      }
    }
    return downstream
  }
}

class VelocityObservable: Observable<(UIGestureRecognizerState, CGPoint)> {
  init(listeningTo to: UIPanGestureRecognizer) {
    super.init()
    to.addTarget(self, action: #selector(panDidUpdate))
  }

  @objc private func panDidUpdate(gesture: UIPanGestureRecognizer) {
    onNext(value: (gesture.state, gesture.velocity(in: gesture.view)))
  }
}

let pan = UIPanGestureRecognizer()
stream = VelocityObservable(listeningTo: pan)
  .filter     { (state, _) in state == .ended }
  .map        { (_, value) in value }
  .subscribe  { print($0) }