Concat
Concatenates all inner observable sequences, as long as the previous observable sequence terminated successfully.
1
2
3
4
5
6
7
8
public func concat() -> Observable<Element.Element> {
return self.merge(maxConcurrent: 1)
}
public func merge(maxConcurrent: Int)
-> Observable<Element.Element> {
return MergeLimited(source: self.asObservable(), maxConcurrent: maxConcurrent)
}
private final class MergeLimitedSinkIter<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>: ObserverType, LockOwnerType, SynchronizedOnType where SourceSequence.Element == Observer.Element
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
typealias Element = Observer.Element
typealias DisposeKey = CompositeDisposable.DisposeKey
typealias Parent = MergeLimitedSink<SourceElement, SourceSequence, Observer>
private let _parent: Parent
private let _disposeKey: DisposeKey
var _lock: RecursiveLock {
return self._parent._lock
}
func on(_ event: Event<Element>) {
self.synchronizedOn(event)
}
func _synchronized_on(_ event: Event<Element>) {
switch event {
case .next:
self._parent.forwardOn(event)
case .error:
self._parent.forwardOn(event)
self._parent.dispose()
case .completed:
self._parent._group.remove(for: self._disposeKey)
if let next = self._parent._queue.dequeue() {
// active next observable
self._parent.subscribe(next, group: self._parent._group)
}
else {
// descrease activeCount
self._parent._activeCount -= 1
if self._parent._stopped && self._parent._activeCount == 0 {
self._parent.forwardOn(.completed)
self._parent.dispose()
}
}
}
}
private class MergeLimitedSink<SourceElement, SourceSequence: ObservableConvertibleType, Observer: ObserverType>: Sink, ObserverType where Observer.Element == SourceSequence.Element
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
typealias QueueType = Queue<SourceSequence>
let _maxConcurrent: Int
let _lock = RecursiveLock()
var _stopped = false
var _activeCount = 0
var _queue = QueueType(capacity: 2)
let _sourceSubscription = SingleAssignmentDisposable()
let _group = CompositeDisposable()
func run(_ source: Observable<SourceElement>) -> Disposable {
_ = self._group.insert(self._sourceSubscription)
let disposable = source.subscribe(self)
self._sourceSubscription.setDisposable(disposable)
return self._group
}
func subscribe(_ innerSource: SourceSequence, group: CompositeDisposable) {
let subscription = SingleAssignmentDisposable()
let key = group.insert(subscription)
if let key = key {
let observer = MergeLimitedSinkIter(parent: self, disposeKey: key)
let disposable = innerSource.asObservable().subscribe(observer)
subscription.setDisposable(disposable)
}
}
func performMap(_ element: SourceElement) throws -> SourceSequence {
rxAbstractMethod()
}
@inline(__always)
final private func nextElementArrived(element: SourceElement) -> SourceSequence? {
self._lock.lock(); defer { self._lock.unlock() } // {
let subscribe: Bool
if self._activeCount < self._maxConcurrent {
// subscribe if activeCount < maxConcurrent
self._activeCount += 1
subscribe = true
}
else {
do {
// otherwise enqueue the observable
let value = try self.performMap(element)
self._queue.enqueue(value)
} catch {
self.forwardOn(.error(error))
self.dispose()
}
subscribe = false
}
if subscribe {
do {
return try self.performMap(element)
} catch {
self.forwardOn(.error(error))
self.dispose()
}
}
return nil
// }
}
func on(_ event: Event<SourceElement>) {
switch event {
case .next(let element):
if let sequence = self.nextElementArrived(element: element) {
self.subscribe(sequence, group: self._group)
}
case .error(let error):
self._lock.lock(); defer { self._lock.unlock() }
self.forwardOn(.error(error))
self.dispose()
case .completed:
self._lock.lock(); defer { self._lock.unlock() }
if self._activeCount == 0 {
self.forwardOn(.completed)
self.dispose()
}
else {
self._sourceSubscription.dispose()
}
self._stopped = true
}
}
private final class MergeLimitedBasicSink<SourceSequence: ObservableConvertibleType, Observer: ObserverType>: MergeLimitedSink<SourceSequence, SourceSequence, Observer> where Observer.Element == SourceSequence.Element
1
2
3
override func performMap(_ element: SourceSequence) throws -> SourceSequence {
return element
}
final private class MergeLimited<SourceSequence: ObservableConvertibleType>: Producer
1
2
3
4
5
6
7
8
private let _source: Observable<SourceSequence>
private let _maxConcurrent: Int
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == SourceSequence.Element {
let sink = MergeLimitedBasicSink<SourceSequence, Observer>(maxConcurrent: self._maxConcurrent, observer: observer, cancel: cancel)
let subscription = sink.run(self._source)
return (sink: sink, subscription: subscription)
}