0001 // 0002 // Observable+Multiple.swift 0003 // Rx 0004 // 0005 // Created by Krunoslav Zaher on 3/12/15. 0006 // Copyright © 2015 Krunoslav Zaher. All rights reserved. 0007 // 0008 0009 import Foundation 0010 0011 // MARK: combineLatest 0012 0013 extension CollectionType where Generator.Element : ObservableType { 0014 0015 /** 0016 Merges the specified observable sequences into one observable sequence by using the selector function whenever any of the observable sequences produces an element. 0017 0018 - seealso: [combinelatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html) 0019 0020 - parameter resultSelector: Function to invoke whenever any of the sources produces an element. 0021 - returns: An observable sequence containing the result of combining elements of the sources using the specified result selector function. 0022 */ 0023 @warn_unused_result(message="http://git.io/rxs.uo") 0024 public func combineLatest<R>(resultSelector: [Generator.Element.E] throws -> R) -> Observable<R> { 0025 return CombineLatestCollectionType(sources: self, resultSelector: resultSelector) 0026 } 0027 } 0028 0029 // MARK: zip 0030 0031 extension CollectionType where Generator.Element : ObservableType { 0032 0033 /** 0034 Merges the specified observable sequences into one observable sequence by using the selector function whenever all of the observable sequences have produced an element at a corresponding index. 0035 0036 - seealso: [zip operator on reactivex.io](http://reactivex.io/documentation/operators/zip.html) 0037 0038 - parameter resultSelector: Function to invoke for each series of elements at corresponding indexes in the sources. 0039 - returns: An observable sequence containing the result of combining elements of the sources using the specified result selector function. 0040 */ 0041 @warn_unused_result(message="http://git.io/rxs.uo") 0042 public func zip<R>(resultSelector: [Generator.Element.E] throws -> R) -> Observable<R> { 0043 return ZipCollectionType(sources: self, resultSelector: resultSelector) 0044 } 0045 } 0046 0047 // MARK: switch 0048 0049 extension ObservableType where E : ObservableConvertibleType { 0050 0051 /** 0052 Transforms an observable sequence of observable sequences into an observable sequence 0053 producing values only from the most recent observable sequence. 0054 0055 Each time a new inner observable sequence is received, unsubscribe from the 0056 previous inner observable sequence. 0057 0058 - seealso: [switch operator on reactivex.io](http://reactivex.io/documentation/operators/switch.html) 0059 0060 - returns: The observable sequence that at any point in time produces the elements of the most recent inner observable sequence that has been received. 0061 */ 0062 @warn_unused_result(message="http://git.io/rxs.uo") 0063 public func switchLatest() -> Observable<E.E> { 0064 return Switch(source: asObservable()) 0065 } 0066 } 0067 0068 // MARK: concat 0069 0070 extension ObservableType { 0071 0072 /** 0073 Concatenates the second observable sequence to `self` upon successful termination of `self`. 0074 0075 - seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html) 0076 0077 - parameter second: Second observable sequence. 0078 - returns: An observable sequence that contains the elements of `self`, followed by those of the second sequence. 0079 */ 0080 @warn_unused_result(message="http://git.io/rxs.uo") 0081 public func concat<O: ObservableConvertibleType where O.E == E>(second: O) -> Observable<E> { 0082 return [asObservable(), second.asObservable()].concat() 0083 } 0084 } 0085 0086 extension SequenceType where Generator.Element : ObservableType { 0087 0088 /** 0089 Concatenates all observable sequences in the given sequence, as long as the previous observable sequence terminated successfully. 0090 0091 This operator has tail recursive optimizations that will prevent stack overflow. 0092 0093 Optimizations will be performed in cases equivalent to following: 0094 0095 [1, [2, [3, .....].concat()].concat].concat() 0096 0097 - seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html) 0098 0099 - returns: An observable sequence that contains the elements of each given sequence, in sequential order. 0100 */ 0101 @warn_unused_result(message="http://git.io/rxs.uo") 0102 public func concat() 0103 -> Observable<Generator.Element.E> { 0104 return Concat(sources: self, count: nil) 0105 } 0106 } 0107 0108 extension CollectionType where Generator.Element : ObservableType { 0109 0110 /** 0111 Concatenates all observable sequences in the given sequence, as long as the previous observable sequence terminated successfully. 0112 0113 This operator has tail recursive optimizations that will prevent stack overflow and enable generating 0114 infinite observable sequences while using limited amount of memory during generation. 0115 0116 Optimizations will be performed in cases equivalent to following: 0117 0118 [1, [2, [3, .....].concat()].concat].concat() 0119 0120 - seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html) 0121 0122 - returns: An observable sequence that contains the elements of each given sequence, in sequential order. 0123 */ 0124 @warn_unused_result(message="http://git.io/rxs.uo") 0125 public func concat() 0126 -> Observable<Generator.Element.E> { 0127 return Concat(sources: self, count: self.count.toIntMax()) 0128 } 0129 } 0130 0131 extension ObservableType where E : ObservableConvertibleType { 0132 0133 /** 0134 Concatenates all inner observable sequences, as long as the previous observable sequence terminated successfully. 0135 0136 - seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html) 0137 0138 - returns: An observable sequence that contains the elements of each observed inner sequence, in sequential order. 0139 */ 0140 @warn_unused_result(message="http://git.io/rxs.uo") 0141 public func concat() -> Observable<E.E> { 0142 return merge(maxConcurrent: 1) 0143 } 0144 } 0145 0146 // MARK: merge 0147 0148 extension ObservableType where E : ObservableConvertibleType { 0149 0150 /** 0151 Merges elements from all observable sequences in the given enumerable sequence into a single observable sequence. 0152 0153 - seealso: [merge operator on reactivex.io](http://reactivex.io/documentation/operators/merge.html) 0154 0155 - returns: The observable sequence that merges the elements of the observable sequences. 0156 */ 0157 @warn_unused_result(message="http://git.io/rxs.uo") 0158 public func merge() -> Observable<E.E> { 0159 return Merge(source: asObservable()) 0160 } 0161 0162 /** 0163 Merges elements from all inner observable sequences into a single observable sequence, limiting the number of concurrent subscriptions to inner sequences. 0164 0165 - seealso: [merge operator on reactivex.io](http://reactivex.io/documentation/operators/merge.html) 0166 0167 - parameter maxConcurrent: Maximum number of inner observable sequences being subscribed to concurrently. 0168 - returns: The observable sequence that merges the elements of the inner sequences. 0169 */ 0170 @warn_unused_result(message="http://git.io/rxs.uo") 0171 public func merge
Observable+Multiple.swift:82 (maxConcurrent maxConcurrent: Int) 0172 -> Observable<E.E> { 0173 return MergeLimited(source: asObservable(), maxConcurrent: maxConcurrent) 0174 } 0175 } 0176 0177 // MARK: catch 0178 0179 extension ObservableType { 0180 0181 /** 0182 Continues an observable sequence that is terminated by an error with the observable sequence produced by the handler. 0183 0184 - seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html) 0185 0186 - parameter handler: Error handler function, producing another observable sequence. 0187 - returns: An observable sequence containing the source sequence's elements, followed by the elements produced by the handler's resulting observable sequence in case an error occurred. 0188 */ 0189 @warn_unused_result(message="http://git.io/rxs.uo") 0190 public func catchError(handler: (ErrorType) throws -> Observable<E>) 0191 -> Observable<E> { 0192 return Catch(source: asObservable(), handler: handler) 0193 } 0194 0195 /** 0196 Continues an observable sequence that is terminated by an error with a single element. 0197 0198 - seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html) 0199 0200 - parameter element: Last element in an observable sequence in case error occurs. 0201 - returns: An observable sequence containing the source sequence's elements, followed by the `element` in case an error occurred. 0202 */ 0203 @warn_unused_result(message="http://git.io/rxs.uo") 0204 public func catchErrorJustReturn(element: E) 0205 -> Observable<E> { 0206 return Catch(source: asObservable(), handler: { _ in Observable.just(element) }) 0207 } 0208 0209 } 0210 0211 extension SequenceType where Generator.Element : ObservableType { 0212 /** 0213 Continues an observable sequence that is terminated by an error with the next observable sequence. 0214 0215 - seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html) 0216 0217 - returns: An observable sequence containing elements from consecutive source sequences until a source sequence terminates successfully. 0218 */ 0219 @warn_unused_result(message="http://git.io/rxs.uo") 0220 public func catchError() 0221 -> Observable<Generator.Element.E> { 0222 return CatchSequence(sources: self) 0223 } 0224 } 0225 0226 // MARK: takeUntil 0227 0228 extension ObservableType { 0229 0230 /** 0231 Returns the elements from the source observable sequence until the other observable sequence produces an element. 0232 0233 - seealso: [takeUntil operator on reactivex.io](http://reactivex.io/documentation/operators/takeuntil.html) 0234 0235 - parameter other: Observable sequence that terminates propagation of elements of the source sequence. 0236 - returns: An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation. 0237 */ 0238 @warn_unused_result(message="http://git.io/rxs.uo") 0239 public func takeUntil<O: ObservableType>(other: O) 0240 -> Observable<E> { 0241 return TakeUntil(source: asObservable(), other: other.asObservable()) 0242 } 0243 } 0244 0245 // MARK: skipUntil 0246 0247 extension ObservableType { 0248 0249 /** 0250 Returns the elements from the source observable sequence until the other observable sequence produces an element. 0251 0252 - seealso: [skipUntil operator on reactivex.io](http://reactivex.io/documentation/operators/skipuntil.html) 0253 0254 - parameter other: Observable sequence that terminates propagation of elements of the source sequence. 0255 - returns: An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation. 0256 */ 0257 @warn_unused_result(message="http://git.io/rxs.uo") 0258 public func skipUntil<O: ObservableType>(other: O) 0259 -> Observable<E> { 0260 return SkipUntil(source: asObservable(), other: other.asObservable()) 0261 } 0262 } 0263 0264 // MARK: amb 0265 0266 extension ObservableType { 0267 0268 /** 0269 Propagates the observable sequence that reacts first. 0270 0271 - seealso: [amb operator on reactivex.io](http://reactivex.io/documentation/operators/amb.html) 0272 0273 - parameter right: Second observable sequence. 0274 - returns: An observable sequence that surfaces either of the given sequences, whichever reacted first. 0275 */ 0276 @warn_unused_result(message="http://git.io/rxs.uo") 0277 public func amb
Observable+Multiple.swift:142 <O2: ObservableType where O2.E == E> 0278 (right: O2) 0279 -> Observable<E> { 0280 return Amb(left: asObservable(), right: right.asObservable()) 0281 } 0282 } 0283 0284 extension SequenceType where Generator.Element : ObservableType { 0285 0286 /** 0287 Propagates the observable sequence that reacts first. 0288 0289 - seealso: [amb operator on reactivex.io](http://reactivex.io/documentation/operators/amb.html) 0290 0291 - returns: An observable sequence that surfaces any of the given sequences, whichever reacted first. 0292 */ 0293 @warn_unused_result(message="http://git.io/rxs.uo") 0294 public func amb() 0295 -> Observable<Generator.Element.E> { 0296 return self.reduce(Observable.never()) { a, o in 0297 return a.amb(o.asObservable()) 0298 } 0299 } 0300 } 0301 0302 // withLatestFrom 0303 0304 extension ObservableType { 0305 0306 /** 0307 Merges two observable sequences into one observable sequence by combining each element from self with the latest element from the second source, if any. 0308 0309 - seealso: [combineLatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html) 0310 0311 - parameter second: Second observable source. 0312 - parameter resultSelector: Function to invoke for each element from the self combined with the latest element from the second source, if any. 0313 - returns: An observable sequence containing the result of combining each element of the self with the latest element from the second source, if any, using the specified result selector function. 0314 */ 0315 public func withLatestFrom<SecondO: ObservableConvertibleType, ResultType>(second: SecondO, resultSelector: (E, SecondO.E) throws -> ResultType) -> Observable<ResultType> { 0316 return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: resultSelector) 0317 } 0318 0319 /** 0320 Merges two observable sequences into one observable sequence by using latest element from the second sequence every time when `self` emitts an element. 0321 0322 - seealso: [combineLatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html) 0323 0324 - parameter second: Second observable source. 0325 - returns: An observable sequence containing the result of combining each element of the self with the latest element from the second source, if any, using the specified result selector function. 0326 */ 0327 public func withLatestFrom<SecondO: ObservableConvertibleType>(second: SecondO) -> Observable<SecondO.E> { 0328 return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: { $1 }) 0329 } 0330 } 0331
Observable+Multiple.swift:297