Update outdated pods.

Pods/RxSwift/RxSwift/Subjects/AsyncSubject.swift

@@ -0,0 +1,146 @@
+//
+//  AsyncSubject.swift
+//  RxSwift
+//
+//  Created by Victor Galán on 07/01/2017.
+//  Copyright © 2017 Krunoslav Zaher. All rights reserved.
+//
+
+/// An AsyncSubject emits the last value (and only the last value) emitted by the source Observable,
+/// and only after that source Observable completes.
+///
+/// (If the source Observable does not emit any values, the AsyncSubject also completes without emitting any values.)
+public final class AsyncSubject<Element>
+    : Observable<Element>
+    , SubjectType
+    , ObserverType
+    , SynchronizedUnsubscribeType {
+    public typealias SubjectObserverType = AsyncSubject<Element>
+
+    typealias Observers = AnyObserver<Element>.s
+    typealias DisposeKey = Observers.KeyType
+
+    /// Indicates whether the subject has any observers
+    public var hasObservers: Bool {
+        _lock.lock(); defer { _lock.unlock() }
+        return _observers.count > 0
+    }
+
+    let _lock = RecursiveLock()
+
+    // state
+    private var _observers = Observers()
+    private var _isStopped = false
+    private var _stoppedEvent = nil as Event<Element>? {
+        didSet {
+            _isStopped = _stoppedEvent != nil
+        }
+    }
+    private var _lastElement: Element?
+
+    /// Creates a subject.
+    public override init() {
+        #if TRACE_RESOURCES
+            _ = Resources.incrementTotal()
+        #endif
+        super.init()
+    }
+
+    /// Notifies all subscribed observers about next event.
+    ///
+    /// - parameter event: Event to send to the observers.
+    public func on(_ event: Event<E>) {
+        let (observers, event) = _synchronized_on(event)
+        switch event {
+        case .next:
+            dispatch(observers, event)
+            dispatch(observers, .completed)
+        case .completed:
+            dispatch(observers, event)
+        case .error:
+            dispatch(observers, event)
+        }
+    }
+
+    func _synchronized_on(_ event: Event<E>) -> (Observers, Event<E>) {
+        _lock.lock(); defer { _lock.unlock() }
+        if _isStopped {
+            return (Observers(), .completed)
+        }
+
+        switch event {
+        case .next(let element):
+            _lastElement = element
+            return (Observers(), .completed)
+        case .error:
+            _stoppedEvent = event
+
+            let observers = _observers
+            _observers.removeAll()
+
+            return (observers, event)
+        case .completed:
+
+            let observers = _observers
+            _observers.removeAll()
+
+            if let lastElement = _lastElement {
+                _stoppedEvent = .next(lastElement)
+                return (observers, .next(lastElement))
+            }
+            else {
+                _stoppedEvent = event
+                return (observers, .completed)
+            }
+        }
+    }
+
+    /// Subscribes an observer to the subject.
+    ///
+    /// - parameter observer: Observer to subscribe to the subject.
+    /// - returns: Disposable object that can be used to unsubscribe the observer from the subject.
+    public override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
+        _lock.lock(); defer { _lock.unlock() }
+        return _synchronized_subscribe(observer)
+    }
+
+    func _synchronized_subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == E {
+        if let stoppedEvent = _stoppedEvent {
+            switch stoppedEvent {
+            case .next:
+                observer.on(stoppedEvent)
+                observer.on(.completed)
+            case .completed:
+                observer.on(stoppedEvent)
+            case .error:
+                observer.on(stoppedEvent)
+            }
+            return Disposables.create()
+        }
+
+        let key = _observers.insert(observer.on)
+
+        return SubscriptionDisposable(owner: self, key: key)
+    }
+
+    func synchronizedUnsubscribe(_ disposeKey: DisposeKey) {
+        _lock.lock(); defer { _lock.unlock() }
+        _synchronized_unsubscribe(disposeKey)
+    }
+    
+    func _synchronized_unsubscribe(_ disposeKey: DisposeKey) {
+        _ = _observers.removeKey(disposeKey)
+    }
+    
+    /// Returns observer interface for subject.
+    public func asObserver() -> AsyncSubject<Element> {
+        return self
+    }
+
+    #if TRACE_RESOURCES
+    deinit {
+        _ = Resources.decrementTotal()
+    }
+    #endif
+}
+

Pods/RxSwift/RxSwift/Subjects/BehaviorSubject.swift

@@ -6,8 +6,6 @@
 //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
 //
 
-import Foundation
-
 /// Represents a value that changes over time.
 ///
 /// Observers can subscribe to the subject to receive the last (or initial) value and all subsequent notifications.
@@ -18,20 +16,24 @@ public final class BehaviorSubject<Element>
     , SynchronizedUnsubscribeType
     , Disposable {
     public typealias SubjectObserverType = BehaviorSubject<Element>
-    typealias DisposeKey = Bag<AnyObserver<Element>>.KeyType
+
+    typealias Observers = AnyObserver<Element>.s
+    typealias DisposeKey = Observers.KeyType
     
     /// Indicates whether the subject has any observers
     public var hasObservers: Bool {
-        _lock.lock(); defer { _lock.unlock() }
-        return _observers.count > 0
+        _lock.lock()
+        let value = _observers.count > 0
+        _lock.unlock()
+        return value
     }
     
-    let _lock = NSRecursiveLock()
+    let _lock = RecursiveLock()
     
     // state
     private var _isDisposed = false
-    private var _value: Element
-    private var _observers = Bag<AnyObserver<Element>>()
+    private var _element: Element
+    private var _observers = Observers()
     private var _stoppedEvent: Event<Element>?
 
     /// Indicates whether the subject has been disposed.
@@ -43,7 +45,11 @@ public final class BehaviorSubject<Element>
     ///
     /// - parameter value: Initial value sent to observers when no other value has been received by the subject yet.
     public init(value: Element) {
-        _value = value
+        _element = value
+
+        #if TRACE_RESOURCES
+            _ = Resources.incrementTotal()
+        #endif
     }
     
     /// Gets the current value or throws an error.
@@ -60,7 +66,7 @@ public final class BehaviorSubject<Element>
                 throw error
             }
             else {
-                return _value
+                return _element
             }
         //}
     }
@@ -69,18 +75,19 @@ public final class BehaviorSubject<Element>
     ///
     /// - parameter event: Event to send to the observers.
     public func on(_ event: Event<E>) {
-        _synchronized_on(event).on(event)
+        _lock.lock()
+        dispatch(_synchronized_on(event), event)
+        _lock.unlock()
     }
 
-    func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {
-        _lock.lock(); defer { _lock.unlock() }
+    func _synchronized_on(_ event: Event<E>) -> Observers {
         if _stoppedEvent != nil || _isDisposed {
-            return Bag()
+            return Observers()
         }
         
         switch event {
-        case .next(let value):
-            _value = value
+        case .next(let element):
+            _element = element
         case .error, .completed:
             _stoppedEvent = event
         }
@@ -93,8 +100,10 @@ public final class BehaviorSubject<Element>
     /// - parameter observer: Observer to subscribe to the subject.
     /// - returns: Disposable object that can be used to unsubscribe the observer from the subject.
     public override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
-        _lock.lock(); defer { _lock.unlock() }
-        return _synchronized_subscribe(observer)
+        _lock.lock()
+        let subscription = _synchronized_subscribe(observer)
+        _lock.unlock()
+        return subscription
     }
 
     func _synchronized_subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == E {
@@ -108,15 +117,16 @@ public final class BehaviorSubject<Element>
             return Disposables.create()
         }
         
-        let key = _observers.insert(observer.asObserver())
-        observer.on(.next(_value))
+        let key = _observers.insert(observer.on)
+        observer.on(.next(_element))
     
         return SubscriptionDisposable(owner: self, key: key)
     }
 
     func synchronizedUnsubscribe(_ disposeKey: DisposeKey) {
-        _lock.lock(); defer { _lock.unlock() }
+        _lock.lock()
         _synchronized_unsubscribe(disposeKey)
+        _lock.unlock()
     }
 
     func _synchronized_unsubscribe(_ disposeKey: DisposeKey) {
@@ -134,10 +144,16 @@ public final class BehaviorSubject<Element>
 
     /// Unsubscribe all observers and release resources.
     public func dispose() {
-        _lock.performLocked {
-            _isDisposed = true
-            _observers.removeAll()
-            _stoppedEvent = nil
-        }
+        _lock.lock()
+        _isDisposed = true
+        _observers.removeAll()
+        _stoppedEvent = nil
+        _lock.unlock()
     }
+
+    #if TRACE_RESOURCES
+        deinit {
+        _ = Resources.decrementTotal()
+        }
+    #endif
 }

Pods/RxSwift/RxSwift/Subjects/PublishSubject.swift

@@ -6,32 +6,33 @@
 //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
 //
 
-import Foundation
-
 /// Represents an object that is both an observable sequence as well as an observer.
 ///
 /// Each notification is broadcasted to all subscribed observers.
-final public class PublishSubject<Element>
+public final class PublishSubject<Element>
     : Observable<Element>
     , SubjectType
     , Cancelable
     , ObserverType
     , SynchronizedUnsubscribeType {
     public typealias SubjectObserverType = PublishSubject<Element>
-    
-    typealias DisposeKey = Bag<AnyObserver<Element>>.KeyType
+
+    typealias Observers = AnyObserver<Element>.s
+    typealias DisposeKey = Observers.KeyType
     
     /// Indicates whether the subject has any observers
     public var hasObservers: Bool {
-        _lock.lock(); defer { _lock.unlock() }
-        return _observers.count > 0
+        _lock.lock()
+        let count = _observers.count > 0
+        _lock.unlock()
+        return count
     }
     
-    private var _lock = NSRecursiveLock()
+    private let _lock = RecursiveLock()
     
     // state
     private var _isDisposed = false
-    private var _observers = Bag<AnyObserver<Element>>()
+    private var _observers = Observers()
     private var _stopped = false
     private var _stoppedEvent = nil as Event<Element>?
     
@@ -43,22 +44,24 @@ final public class PublishSubject<Element>
     /// Creates a subject.
     public override init() {
         super.init()
+        #if TRACE_RESOURCES
+            _ = Resources.incrementTotal()
+        #endif
     }
     
     /// Notifies all subscribed observers about next event.
     ///
     /// - parameter event: Event to send to the observers.
     public func on(_ event: Event<Element>) {
-        _synchronized_on(event).on(event)
+        dispatch(_synchronized_on(event), event)
     }
 
-    func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {
+    func _synchronized_on(_ event: Event<E>) -> Observers {
         _lock.lock(); defer { _lock.unlock() }
-
         switch event {
         case .next(_):
             if _isDisposed || _stopped {
-                return Bag()
+                return Observers()
             }
             
             return _observers
@@ -71,7 +74,7 @@ final public class PublishSubject<Element>
                 return observers
             }
 
-            return Bag()
+            return Observers()
         }
     }
     
@@ -82,8 +85,10 @@ final public class PublishSubject<Element>
     - returns: Disposable object that can be used to unsubscribe the observer from the subject.
     */
     public override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
-        _lock.lock(); defer { _lock.unlock() }
-        return _synchronized_subscribe(observer)
+        _lock.lock()
+        let subscription = _synchronized_subscribe(observer)
+        _lock.unlock()
+        return subscription
     }
 
     func _synchronized_subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == E {
@@ -97,13 +102,14 @@ final public class PublishSubject<Element>
             return Disposables.create()
         }
         
-        let key = _observers.insert(observer.asObserver())
+        let key = _observers.insert(observer.on)
         return SubscriptionDisposable(owner: self, key: key)
     }
 
     func synchronizedUnsubscribe(_ disposeKey: DisposeKey) {
-        _lock.lock(); defer { _lock.unlock() }
+        _lock.lock()
         _synchronized_unsubscribe(disposeKey)
+        _lock.unlock()
     }
 
     func _synchronized_unsubscribe(_ disposeKey: DisposeKey) {
@@ -117,8 +123,9 @@ final public class PublishSubject<Element>
     
     /// Unsubscribe all observers and release resources.
     public func dispose() {
-        _lock.lock(); defer { _lock.unlock() }
+        _lock.lock()
         _synchronized_dispose()
+        _lock.unlock()
     }
 
     final func _synchronized_dispose() {
@@ -126,4 +133,10 @@ final public class PublishSubject<Element>
         _observers.removeAll()
         _stoppedEvent = nil
     }
+
+    #if TRACE_RESOURCES
+        deinit {
+            _ = Resources.decrementTotal()
+        }
+    #endif
 }

Pods/RxSwift/RxSwift/Subjects/ReplaySubject.swift

@@ -6,8 +6,6 @@
 //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
 //
 
-import Foundation
-
 /// Represents an object that is both an observable sequence as well as an observer.
 ///
 /// Each notification is broadcasted to all subscribed and future observers, subject to buffer trimming policies.
@@ -17,31 +15,43 @@ public class ReplaySubject<Element>
     , ObserverType
     , Disposable {
     public typealias SubjectObserverType = ReplaySubject<Element>
-    
+
+    typealias Observers = AnyObserver<Element>.s
+    typealias DisposeKey = Observers.KeyType
+
     /// Indicates whether the subject has any observers
     public var hasObservers: Bool {
-        _lock.lock(); defer { _lock.unlock() }
-        return _observers.count > 0
+        _lock.lock()
+        let value = _observers.count > 0
+        _lock.unlock()
+        return value
     }
     
-    fileprivate var _lock = NSRecursiveLock()
+    fileprivate let _lock = RecursiveLock()
     
     // state
     fileprivate var _isDisposed = false
-    fileprivate var _stoppedEvent = nil as Event<Element>?
-    fileprivate var _observers = Bag<AnyObserver<Element>>()
-    
-    typealias DisposeKey = Bag<AnyObserver<Element>>.KeyType
-    
+    fileprivate var _isStopped = false
+    fileprivate var _stoppedEvent = nil as Event<Element>? {
+        didSet {
+            _isStopped = _stoppedEvent != nil
+        }
+    }
+    fileprivate var _observers = Observers()
+
     func unsubscribe(_ key: DisposeKey) {
-        abstractMethod()
+        rxAbstractMethod()
+    }
+
+    final var isStopped: Bool {
+        return _isStopped
     }
     
     /// Notifies all subscribed observers about next event.
     ///
     /// - parameter event: Event to send to the observers.
     public func on(_ event: Event<E>) {
-        abstractMethod()
+        rxAbstractMethod()
     }
     
     /// Returns observer interface for subject.
@@ -72,41 +82,51 @@ public class ReplaySubject<Element>
     public static func createUnbounded() -> ReplaySubject<Element> {
         return ReplayAll()
     }
+
+    #if TRACE_RESOURCES
+        override init() {
+            _ = Resources.incrementTotal()
+        }
+
+        deinit {
+            _ = Resources.decrementTotal()
+        }
+    #endif
 }
 
-class ReplayBufferBase<Element>
+fileprivate class ReplayBufferBase<Element>
     : ReplaySubject<Element>
     , SynchronizedUnsubscribeType {
     
     func trim() {
-        abstractMethod()
+        rxAbstractMethod()
     }
     
     func addValueToBuffer(_ value: Element) {
-        abstractMethod()
+        rxAbstractMethod()
     }
     
-    func replayBuffer(_ observer: AnyObserver<Element>) {
-        abstractMethod()
+    func replayBuffer<O: ObserverType>(_ observer: O) where O.E == Element {
+        rxAbstractMethod()
     }
     
     override func on(_ event: Event<Element>) {
-        _synchronized_on(event).on(event)
+        dispatch(_synchronized_on(event), event)
     }
 
-    func _synchronized_on(_ event: Event<E>) -> Bag<AnyObserver<Element>> {
+    func _synchronized_on(_ event: Event<E>) -> Observers {
         _lock.lock(); defer { _lock.unlock() }
         if _isDisposed {
-            return Bag()
+            return Observers()
         }
         
-        if _stoppedEvent != nil {
-            return Bag()
+        if _isStopped {
+            return Observers()
         }
         
         switch event {
-        case .next(let value):
-            addValueToBuffer(value)
+        case .next(let element):
+            addValueToBuffer(element)
             trim()
             return _observers
         case .error, .completed:
@@ -119,8 +139,10 @@ class ReplayBufferBase<Element>
     }
     
     override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
-        _lock.lock(); defer { _lock.unlock() }
-        return _synchronized_subscribe(observer)
+        _lock.lock()
+        let subscription = _synchronized_subscribe(observer)
+        _lock.unlock()
+        return subscription
     }
 
     func _synchronized_subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == E {
@@ -129,22 +151,23 @@ class ReplayBufferBase<Element>
             return Disposables.create()
         }
      
-        let AnyObserver = observer.asObserver()
+        let anyObserver = observer.asObserver()
         
-        replayBuffer(AnyObserver)
+        replayBuffer(anyObserver)
         if let stoppedEvent = _stoppedEvent {
             observer.on(stoppedEvent)
             return Disposables.create()
         }
         else {
-            let key = _observers.insert(AnyObserver)
+            let key = _observers.insert(observer.on)
             return SubscriptionDisposable(owner: self, key: key)
         }
     }
 
     func synchronizedUnsubscribe(_ disposeKey: DisposeKey) {
-        _lock.lock(); defer { _lock.unlock() }
+        _lock.lock()
         _synchronized_unsubscribe(disposeKey)
+        _lock.unlock()
     }
 
     func _synchronized_unsubscribe(_ disposeKey: DisposeKey) {
@@ -162,13 +185,13 @@ class ReplayBufferBase<Element>
     }
 
     func synchronizedDispose() {
-        _lock.lock(); defer { _lock.unlock() }
+        _lock.lock()
         _synchronized_dispose()
+        _lock.unlock()
     }
 
     func _synchronized_dispose() {
         _isDisposed = true
-        _stoppedEvent = nil
         _observers.removeAll()
     }
 }
@@ -187,8 +210,8 @@ final class ReplayOne<Element> : ReplayBufferBase<Element> {
     override func addValueToBuffer(_ value: Element) {
         _value = value
     }
-    
-    override func replayBuffer(_ observer: AnyObserver<Element>) {
+
+    override func replayBuffer<O: ObserverType>(_ observer: O) where O.E == Element {
         if let value = _value {
             observer.on(.next(value))
         }
@@ -210,8 +233,8 @@ class ReplayManyBase<Element> : ReplayBufferBase<Element> {
     override func addValueToBuffer(_ value: Element) {
         _queue.enqueue(value)
     }
-    
-    override func replayBuffer(_ observer: AnyObserver<E>) {
+
+    override func replayBuffer<O: ObserverType>(_ observer: O) where O.E == Element {
         for item in _queue {
             observer.on(.next(item))
         }

Pods/RxSwift/RxSwift/Subjects/SubjectType.swift

@@ -6,8 +6,6 @@
 //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
 //
 
-import Foundation
-
 /// Represents an object that is both an observable sequence as well as an observer.
 public protocol SubjectType : ObservableType {
     /// The type of the observer that represents this subject.

Pods/RxSwift/RxSwift/Subjects/Variable.swift

@@ -6,13 +6,11 @@
 //  Copyright © 2015 Krunoslav Zaher. All rights reserved.
 //
 
-import Foundation
-
 /// Variable is a wrapper for `BehaviorSubject`.
 ///
 /// Unlike `BehaviorSubject` it can't terminate with error, and when variable is deallocated
 /// it will complete it's observable sequence (`asObservable`).
-public class Variable<Element> {
+public final class Variable<Element> {
 
     public typealias E = Element
     
@@ -22,7 +20,11 @@ public class Variable<Element> {
  
     // state
     private var _value: E
-    
+
+    #if DEBUG
+        fileprivate var _numberOfConcurrentCalls: AtomicInt = 0
+    #endif
+
     /// Gets or sets current value of variable.
     ///
     /// Whenever a new value is set, all the observers are notified of the change.
@@ -34,6 +36,15 @@ public class Variable<Element> {
             return _value
         }
         set(newValue) {
+            #if DEBUG
+                if AtomicIncrement(&_numberOfConcurrentCalls) > 1 {
+                    rxFatalError("Warning: Recursive call or synchronization error!")
+                }
+
+                defer {
+                    _ = AtomicDecrement(&_numberOfConcurrentCalls)
+                }
+            #endif
             _lock.lock()
             _value = newValue
             _lock.unlock()