-
Notifications
You must be signed in to change notification settings - Fork 4.2k
/
Observable+Multiple.swift
308 lines (226 loc) · 12.8 KB
/
Observable+Multiple.swift
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
//
// Observable+Multiple.swift
// RxSwift
//
// Created by Krunoslav Zaher on 3/12/15.
// Copyright © 2015 Krunoslav Zaher. All rights reserved.
//
import Foundation
// MARK: combineLatest
extension Observable {
/**
Merges the specified observable sequences into one observable sequence by using the selector function whenever any of the observable sequences produces an element.
- seealso: [combinelatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html)
- parameter resultSelector: Function to invoke whenever any of the sources produces an element.
- returns: An observable sequence containing the result of combining elements of the sources using the specified result selector function.
*/
public static func combineLatest<C: Collection>(_ collection: C, _ resultSelector: @escaping ([C.Iterator.Element.E]) throws -> Element) -> Observable<Element>
where C.Iterator.Element: ObservableType {
return CombineLatestCollectionType(sources: collection, resultSelector: resultSelector)
}
}
// MARK: zip
extension Observable {
/**
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.
- seealso: [zip operator on reactivex.io](http://reactivex.io/documentation/operators/zip.html)
- parameter resultSelector: Function to invoke for each series of elements at corresponding indexes in the sources.
- returns: An observable sequence containing the result of combining elements of the sources using the specified result selector function.
*/
public static func zip<C: Collection>(_ collection: C, _ resultSelector: @escaping ([C.Iterator.Element.E]) throws -> Element) -> Observable<Element>
where C.Iterator.Element: ObservableType {
return ZipCollectionType(sources: collection, resultSelector: resultSelector)
}
}
// MARK: switch
extension ObservableType where E : ObservableConvertibleType {
/**
Transforms an observable sequence of observable sequences into an observable sequence
producing values only from the most recent observable sequence.
Each time a new inner observable sequence is received, unsubscribe from the
previous inner observable sequence.
- seealso: [switch operator on reactivex.io](http://reactivex.io/documentation/operators/switch.html)
- returns: The observable sequence that at any point in time produces the elements of the most recent inner observable sequence that has been received.
*/
public func switchLatest() -> Observable<E.E> {
return Switch(source: asObservable())
}
}
// MARK: concat
extension ObservableType {
/**
Concatenates the second observable sequence to `self` upon successful termination of `self`.
- seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html)
- parameter second: Second observable sequence.
- returns: An observable sequence that contains the elements of `self`, followed by those of the second sequence.
*/
public func concat<O: ObservableConvertibleType>(_ second: O) -> Observable<E> where O.E == E {
return Observable.concat([self.asObservable(), second.asObservable()])
}
}
extension Observable {
/**
Concatenates all observable sequences in the given sequence, as long as the previous observable sequence terminated successfully.
This operator has tail recursive optimizations that will prevent stack overflow.
Optimizations will be performed in cases equivalent to following:
[1, [2, [3, .....].concat()].concat].concat()
- seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html)
- returns: An observable sequence that contains the elements of each given sequence, in sequential order.
*/
public static func concat<S: Sequence >(_ sequence: S) -> Observable<Element>
where S.Iterator.Element == Observable<Element> {
return Concat(sources: sequence, count: nil)
}
/**
Concatenates all observable sequences in the given collection, as long as the previous observable sequence terminated successfully.
This operator has tail recursive optimizations that will prevent stack overflow.
Optimizations will be performed in cases equivalent to following:
[1, [2, [3, .....].concat()].concat].concat()
- seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html)
- returns: An observable sequence that contains the elements of each given sequence, in sequential order.
*/
public static func concat<S: Collection >(_ collection: S) -> Observable<Element>
where S.Iterator.Element == Observable<Element> {
return Concat(sources: collection, count: collection.count.toIntMax())
}
}
extension ObservableType where E : ObservableConvertibleType {
/**
Concatenates all inner observable sequences, as long as the previous observable sequence terminated successfully.
- seealso: [concat operator on reactivex.io](http://reactivex.io/documentation/operators/concat.html)
- returns: An observable sequence that contains the elements of each observed inner sequence, in sequential order.
*/
public func concat() -> Observable<E.E> {
return merge(maxConcurrent: 1)
}
}
// MARK: merge
extension ObservableType where E : ObservableConvertibleType {
/**
Merges elements from all observable sequences in the given enumerable sequence into a single observable sequence.
- seealso: [merge operator on reactivex.io](http://reactivex.io/documentation/operators/merge.html)
- returns: The observable sequence that merges the elements of the observable sequences.
*/
public func merge() -> Observable<E.E> {
return Merge(source: asObservable())
}
/**
Merges elements from all inner observable sequences into a single observable sequence, limiting the number of concurrent subscriptions to inner sequences.
- seealso: [merge operator on reactivex.io](http://reactivex.io/documentation/operators/merge.html)
- parameter maxConcurrent: Maximum number of inner observable sequences being subscribed to concurrently.
- returns: The observable sequence that merges the elements of the inner sequences.
*/
public func merge(maxConcurrent: Int)
-> Observable<E.E> {
return MergeLimited(source: asObservable(), maxConcurrent: maxConcurrent)
}
}
// MARK: catch
extension ObservableType {
/**
Continues an observable sequence that is terminated by an error with the observable sequence produced by the handler.
- seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html)
- parameter handler: Error handler function, producing another observable sequence.
- 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.
*/
public func catchError(_ handler: @escaping (Swift.Error) throws -> Observable<E>)
-> Observable<E> {
return Catch(source: asObservable(), handler: handler)
}
/**
Continues an observable sequence that is terminated by an error with a single element.
- seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html)
- parameter element: Last element in an observable sequence in case error occurs.
- returns: An observable sequence containing the source sequence's elements, followed by the `element` in case an error occurred.
*/
public func catchErrorJustReturn(_ element: E)
-> Observable<E> {
return Catch(source: asObservable(), handler: { _ in Observable.just(element) })
}
}
extension Observable {
/**
Continues an observable sequence that is terminated by an error with the next observable sequence.
- seealso: [catch operator on reactivex.io](http://reactivex.io/documentation/operators/catch.html)
- returns: An observable sequence containing elements from consecutive source sequences until a source sequence terminates successfully.
*/
public static func catchError<S: Sequence>(_ sequence: S) -> Observable<Element>
where S.Iterator.Element == Observable<Element> {
return CatchSequence(sources: sequence)
}
}
// MARK: takeUntil
extension ObservableType {
/**
Returns the elements from the source observable sequence until the other observable sequence produces an element.
- seealso: [takeUntil operator on reactivex.io](http://reactivex.io/documentation/operators/takeuntil.html)
- parameter other: Observable sequence that terminates propagation of elements of the source sequence.
- returns: An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation.
*/
public func takeUntil<O: ObservableType>(_ other: O)
-> Observable<E> {
return TakeUntil(source: asObservable(), other: other.asObservable())
}
}
// MARK: skipUntil
extension ObservableType {
/**
Returns the elements from the source observable sequence that are emitted after the other observable sequence produces an element.
- seealso: [skipUntil operator on reactivex.io](http://reactivex.io/documentation/operators/skipuntil.html)
- parameter other: Observable sequence that starts propagation of elements of the source sequence.
- returns: An observable sequence containing the elements of the source sequence that are emitted after the other sequence emits an item.
*/
public func skipUntil<O: ObservableType>(_ other: O)
-> Observable<E> {
return SkipUntil(source: asObservable(), other: other.asObservable())
}
}
// MARK: amb
extension ObservableType {
/**
Propagates the observable sequence that reacts first.
- seealso: [amb operator on reactivex.io](http://reactivex.io/documentation/operators/amb.html)
- parameter right: Second observable sequence.
- returns: An observable sequence that surfaces either of the given sequences, whichever reacted first.
*/
public func amb<O2: ObservableType>
(_ right: O2)
-> Observable<E> where O2.E == E {
return Amb(left: asObservable(), right: right.asObservable())
}
}
extension Observable {
/**
Propagates the observable sequence that reacts first.
- seealso: [amb operator on reactivex.io](http://reactivex.io/documentation/operators/amb.html)
- returns: An observable sequence that surfaces any of the given sequences, whichever reacted first.
*/
public static func amb<S: Sequence>(_ sequence: S) -> Observable<Element>
where S.Iterator.Element == Observable<Element> {
return sequence.reduce(Observable<S.Iterator.Element.E>.never()) { a, o in
return a.amb(o.asObservable())
}
}
}
// withLatestFrom
extension ObservableType {
/**
Merges two observable sequences into one observable sequence by combining each element from self with the latest element from the second source, if any.
- seealso: [combineLatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html)
- parameter second: Second observable source.
- parameter resultSelector: Function to invoke for each element from the self combined with the latest element from the second source, if any.
- 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.
*/
public func withLatestFrom<SecondO: ObservableConvertibleType, ResultType>(_ second: SecondO, resultSelector: @escaping (E, SecondO.E) throws -> ResultType) -> Observable<ResultType> {
return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: resultSelector)
}
/**
Merges two observable sequences into one observable sequence by using latest element from the second sequence every time when `self` emitts an element.
- seealso: [combineLatest operator on reactivex.io](http://reactivex.io/documentation/operators/combinelatest.html)
- parameter second: Second observable source.
- 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.
*/
public func withLatestFrom<SecondO: ObservableConvertibleType>(_ second: SecondO) -> Observable<SecondO.E> {
return WithLatestFrom(first: asObservable(), second: second.asObservable(), resultSelector: { $1 })
}
}