Last active
December 16, 2017 07:52
-
-
Save jverkoey/1b17b4ee9b55431341a198a8b6690c3d to your computer and use it in GitHub Desktop.
Stream prototype in swift
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| /** | |
| 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) } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment