0001 // 0002 // Observable+Single.swift 0003 // Rx 0004 // 0005 // Created by Krunoslav Zaher on 2/14/15. 0006 // Copyright © 2015 Krunoslav Zaher. All rights reserved. 0007 // 0008 0009 import Foundation 0010 0011 // MARK: distinct until changed 0012 0013 extension ObservableType where E: Equatable { 0014 0015 /** 0016 Returns an observable sequence that contains only distinct contiguous elements according to equality operator. 0017 0018 - seealso: [distinct operator on reactivex.io](http://reactivex.io/documentation/operators/distinct.html) 0019 0020 - returns: An observable sequence only containing the distinct contiguous elements, based on equality operator, from the source sequence. 0021 */ 0022 @warn_unused_result(message="http://git.io/rxs.uo") 0023 public func distinctUntilChanged() 0024 -> Observable<E> { 0025 return self.distinctUntilChanged({ $0 }, comparer: { ($0 == $1) }) 0026 } 0027 } 0028 0029 extension ObservableType { 0030 /** 0031 Returns an observable sequence that contains only distinct contiguous elements according to the `keySelector`. 0032 0033 - seealso: [distinct operator on reactivex.io](http://reactivex.io/documentation/operators/distinct.html) 0034 0035 - parameter keySelector: A function to compute the comparison key for each element. 0036 - returns: An observable sequence only containing the distinct contiguous elements, based on a computed key value, from the source sequence. 0037 */ 0038 @warn_unused_result(message="http://git.io/rxs.uo") 0039 public func distinctUntilChanged<K: Equatable>(keySelector: (E) throws -> K) 0040 -> Observable<E> { 0041 return self.distinctUntilChanged(keySelector, comparer: { $0 == $1 }) 0042 } 0043 0044 /** 0045 Returns an observable sequence that contains only distinct contiguous elements according to the `comparer`. 0046 0047 - seealso: [distinct operator on reactivex.io](http://reactivex.io/documentation/operators/distinct.html) 0048 0049 - parameter comparer: Equality comparer for computed key values. 0050 - returns: An observable sequence only containing the distinct contiguous elements, based on `comparer`, from the source sequence. 0051 */ 0052 @warn_unused_result(message="http://git.io/rxs.uo") 0053 public func distinctUntilChanged(comparer: (lhs: E, rhs: E) throws -> Bool) 0054 -> Observable<E> { 0055 return self.distinctUntilChanged({ $0 }, comparer: comparer) 0056 } 0057 0058 /** 0059 Returns an observable sequence that contains only distinct contiguous elements according to the keySelector and the comparer. 0060 0061 - seealso: [distinct operator on reactivex.io](http://reactivex.io/documentation/operators/distinct.html) 0062 0063 - parameter keySelector: A function to compute the comparison key for each element. 0064 - parameter comparer: Equality comparer for computed key values. 0065 - returns: An observable sequence only containing the distinct contiguous elements, based on a computed key value and the comparer, from the source sequence. 0066 */ 0067 @warn_unused_result(message="http://git.io/rxs.uo") 0068 public func distinctUntilChanged<K>(keySelector: (E) throws -> K, comparer: (lhs: K, rhs: K) throws -> Bool) 0069 -> Observable<E> { 0070 return DistinctUntilChanged(source: self.asObservable(), selector: keySelector, comparer: comparer) 0071 } 0072 } 0073 0074 // MARK: doOn 0075 0076 extension ObservableType { 0077 0078 /** 0079 Invokes an action for each event in the observable sequence, and propagates all observer messages through the result sequence. 0080 0081 - seealso: [do operator on reactivex.io](http://reactivex.io/documentation/operators/do.html) 0082 0083 - parameter eventHandler: Action to invoke for each event in the observable sequence. 0084 - returns: The source sequence with the side-effecting behavior applied. 0085 */ 0086 @warn_unused_result(message="http://git.io/rxs.uo") 0087 public func doOn(eventHandler: (Event<E>) throws -> Void) 0088 -> Observable<E> { 0089 return Do(source: self.asObservable(), eventHandler: eventHandler) 0090 } 0091 0092 /** 0093 Invokes an action for each event in the observable sequence, and propagates all observer messages through the result sequence. 0094 0095 - seealso: [do operator on reactivex.io](http://reactivex.io/documentation/operators/do.html) 0096 0097 - parameter onNext: Action to invoke for each element in the observable sequence. 0098 - parameter onError: Action to invoke upon errored termination of the observable sequence. 0099 - parameter onCompleted: Action to invoke upon graceful termination of the observable sequence. 0100 - returns: The source sequence with the side-effecting behavior applied. 0101 */ 0102 @warn_unused_result(message="http://git.io/rxs.uo") 0103 public func doOn
Observable+Single.swift:25 return self.distinctUntilChanged({ $0 }, comparer: { ($0 == $1) })Observable+Single.swift:41 return self.distinctUntilChanged(keySelector, comparer: { $0 == $1 })Observable+Single.swift:55 return self.distinctUntilChanged({ $0 }, comparer: comparer)(onNext onNext: (E throws -> Void)? = nil, onError: (ErrorType throws -> Void)? = nil, onCompleted: (() throws -> Void)? = nil) 0104 -> Observable<E> { 0105 return Do(source: self.asObservable()) { e in 0106 switch e { 0107 case .Next(let element): 0108 try onNext?(element) 0109 case .Error(let e): 0110 try onError?(e) 0111 case .Completed: 0112 try onCompleted?() 0113 } 0114 } 0115 } 0116 0117 /** 0118 Invokes an action for each Next event in the observable sequence, and propagates all observer messages through the result sequence. 0119 0120 - parameter onNext: Action to invoke for each element in the observable sequence. 0121 - returns: The source sequence with the side-effecting behavior applied. 0122 */ 0123 @warn_unused_result(message="http://git.io/rxs.uo") 0124 public func doOnNext(onNext: (E throws -> Void)) 0125 -> Observable<E> { 0126 return self.doOn(onNext: onNext) 0127 } 0128 0129 /** 0130 Invokes an action for the Error event in the observable sequence, and propagates all observer messages through the result sequence. 0131 0132 - parameter onError: Action to invoke upon errored termination of the observable sequence. 0133 - returns: The source sequence with the side-effecting behavior applied. 0134 */ 0135 @warn_unused_result(message="http://git.io/rxs.uo") 0136 public func doOnError(onError: (ErrorType throws -> Void)) 0137 -> Observable<E> { 0138 return self.doOn(onError: onError) 0139 } 0140 0141 /** 0142 Invokes an action for the Completed event in the observable sequence, and propagates all observer messages through the result sequence. 0143 0144 - parameter onCompleted: Action to invoke upon graceful termination of the observable sequence. 0145 - returns: The source sequence with the side-effecting behavior applied. 0146 */ 0147 @warn_unused_result(message="http://git.io/rxs.uo") 0148 public func doOnCompleted(onCompleted: (() throws -> Void)) 0149 -> Observable<E> { 0150 return self.doOn(onCompleted: onCompleted) 0151 } 0152 } 0153 0154 // MARK: startWith 0155 0156 extension ObservableType { 0157 0158 /** 0159 Prepends a sequence of values to an observable sequence. 0160 0161 - seealso: [startWith operator on reactivex.io](http://reactivex.io/documentation/operators/startwith.html) 0162 0163 - parameter elements: Elements to prepend to the specified sequence. 0164 - returns: The source sequence prepended with the specified values. 0165 */ 0166 @warn_unused_result(message="http://git.io/rxs.uo") 0167 public func startWith(elements: E ...) 0168 -> Observable<E> { 0169 return StartWith(source: self.asObservable(), elements: elements) 0170 } 0171 } 0172 0173 // MARK: retry 0174 0175 extension ObservableType { 0176 0177 /** 0178 Repeats the source observable sequence until it successfully terminates. 0179 0180 **This could potentially create an infinite sequence.** 0181 0182 - seealso: [retry operator on reactivex.io](http://reactivex.io/documentation/operators/retry.html) 0183 0184 - returns: Observable sequence to repeat until it successfully terminates. 0185 */ 0186 @warn_unused_result(message="http://git.io/rxs.uo") 0187 public func retry() -> Observable<E> { 0188 return CatchSequence(sources: InfiniteSequence(repeatedValue: self.asObservable())) 0189 } 0190 0191 /** 0192 Repeats the source observable sequence the specified number of times in case of an error or until it successfully terminates. 0193 0194 If you encounter an error and want it to retry once, then you must use `retry(2)` 0195 0196 - seealso: [retry operator on reactivex.io](http://reactivex.io/documentation/operators/retry.html) 0197 0198 - parameter maxAttemptCount: Maximum number of times to repeat the sequence. 0199 - returns: An observable sequence producing the elements of the given sequence repeatedly until it terminates successfully. 0200 */ 0201 @warn_unused_result(message="http://git.io/rxs.uo") 0202 public func retry(maxAttemptCount: Int) 0203 -> Observable<E> { 0204 return CatchSequence(sources: Repeat(count: maxAttemptCount, repeatedValue: self.asObservable())) 0205 } 0206 0207 /** 0208 Repeats the source observable sequence on error when the notifier emits a next value. 0209 If the source observable errors and the notifier completes, it will complete the source sequence. 0210 0211 - seealso: [retry operator on reactivex.io](http://reactivex.io/documentation/operators/retry.html) 0212 0213 - parameter notificationHandler: A handler that is passed an observable sequence of errors raised by the source observable and returns and observable that either continues, completes or errors. This behavior is then applied to the source observable. 0214 - returns: An observable sequence producing the elements of the given sequence repeatedly until it terminates successfully or is notified to error or complete. 0215 */ 0216 @warn_unused_result(message="http://git.io/rxs.uo") 0217 public func retryWhen<TriggerObservable: ObservableType, Error: ErrorType>(notificationHandler: Observable<Error> -> TriggerObservable) 0218 -> Observable<E> { 0219 return RetryWhenSequence(sources: InfiniteSequence(repeatedValue: self.asObservable()), notificationHandler: notificationHandler) 0220 } 0221 0222 /** 0223 Repeats the source observable sequence on error when the notifier emits a next value. 0224 If the source observable errors and the notifier completes, it will complete the source sequence. 0225 0226 - seealso: [retry operator on reactivex.io](http://reactivex.io/documentation/operators/retry.html) 0227 0228 - parameter notificationHandler: A handler that is passed an observable sequence of errors raised by the source observable and returns and observable that either continues, completes or errors. This behavior is then applied to the source observable. 0229 - returns: An observable sequence producing the elements of the given sequence repeatedly until it terminates successfully or is notified to error or complete. 0230 */ 0231 @warn_unused_result(message="http://git.io/rxs.uo") 0232 public func retryWhen<TriggerObservable: ObservableType>(notificationHandler: Observable<ErrorType> -> TriggerObservable) 0233 -> Observable<E> { 0234 return RetryWhenSequence(sources: InfiniteSequence(repeatedValue: self.asObservable()), notificationHandler: notificationHandler) 0235 } 0236 } 0237 0238 // MARK: scan 0239 0240 extension ObservableType { 0241 0242 /** 0243 Applies an accumulator function over an observable sequence and returns each intermediate result. The specified seed value is used as the initial accumulator value. 0244 0245 For aggregation behavior with no intermediate results, see `reduce`. 0246 0247 - seealso: [scan operator on reactivex.io](http://reactivex.io/documentation/operators/scan.html) 0248 0249 - parameter seed: The initial accumulator value. 0250 - parameter accumulator: An accumulator function to be invoked on each element. 0251 - returns: An observable sequence containing the accumulated values. 0252 */ 0253 @warn_unused_result(message="http://git.io/rxs.uo") 0254 public func scan<A>(seed: A, accumulator: (A, E) throws -> A) 0255 -> Observable<A> { 0256 return Scan(source: self.asObservable(), seed: seed, accumulator: accumulator) 0257 } 0258 }
Observable+Single.swift:126 Observable+Single.swift:138 Observable+Single.swift:150