Merge pull request #92 from Econa77/feature/replace-subject

Replace PublishSubject with InputSubject

Author: freak4pc

Action.xcodeproj/project.pbxproj

@@ -58,6 +58,10 @@
 		7FB791E91D7F1BB200789D53 /* RxSwift.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 7F612AAB1D7F106900B93BC5 /* RxSwift.framework */; };
 		7FB791EA1D7F1BB200789D53 /* RxCocoa.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = 7F612AAC1D7F106900B93BC5 /* RxCocoa.framework */; };
 		7FB791EC1D7F1BB600789D53 /* Action.framework in Frameworks */ = {isa = PBXBuildFile; fileRef = BE73AD201CDCD101006F8B98 /* Action.framework */; };
+		CA2861C81ED6979400BB327A /* InputSubject.swift in Sources */ = {isa = PBXBuildFile; fileRef = CA2861C71ED6979400BB327A /* InputSubject.swift */; };
+		CA2861CA1ED6A41700BB327A /* InputSubjectTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = CA2861C91ED6A41700BB327A /* InputSubjectTests.swift */; };
+		CA2861CB1ED6B08300BB327A /* InputSubject.swift in Sources */ = {isa = PBXBuildFile; fileRef = CA2861C71ED6979400BB327A /* InputSubject.swift */; };
+		CA2861CC1ED6B08400BB327A /* InputSubject.swift in Sources */ = {isa = PBXBuildFile; fileRef = CA2861C71ED6979400BB327A /* InputSubject.swift */; };
 /* End PBXBuildFile section */
 
 /* Begin PBXContainerItemProxy section */
@@ -126,6 +130,8 @@
 		7F612AD81D7F13B800B93BC5 /* Base */ = {isa = PBXFileReference; lastKnownFileType = file.storyboard; name = Base; path = Base.lproj/LaunchScreen.storyboard; sourceTree = "<group>"; };
 		7F612ADA1D7F13B800B93BC5 /* Info.plist */ = {isa = PBXFileReference; lastKnownFileType = text.plist.xml; path = Info.plist; sourceTree = "<group>"; };
 		BE73AD201CDCD101006F8B98 /* Action.framework */ = {isa = PBXFileReference; explicitFileType = wrapper.framework; includeInIndex = 0; path = Action.framework; sourceTree = BUILT_PRODUCTS_DIR; };
+		CA2861C71ED6979400BB327A /* InputSubject.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = InputSubject.swift; sourceTree = "<group>"; };
+		CA2861C91ED6A41700BB327A /* InputSubjectTests.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = InputSubjectTests.swift; sourceTree = "<group>"; };
 /* End PBXFileReference section */
 
 /* Begin PBXFrameworksBuildPhase section */
@@ -205,6 +211,7 @@
 			children = (
 				7F0569E21DE28587007E1D0D /* Action.swift */,
 				7F0569E01DE28587007E1D0D /* Action+Internal.swift */,
+				CA2861C71ED6979400BB327A /* InputSubject.swift */,
 				7F0569E41DE28587007E1D0D /* UIKitExtensions */,
 				7F0569EF1DE28598007E1D0D /* Supporting Files */,
 			);
@@ -258,6 +265,7 @@
 				7F0569F11DE288EB007E1D0D /* AlertActionTests.swift */,
 				7F0569F21DE288EB007E1D0D /* BarButtonTests.swift */,
 				7F0569F31DE288EB007E1D0D /* ButtonTests.swift */,
+				CA2861C91ED6A41700BB327A /* InputSubjectTests.swift */,
 				7F0569F41DE288EB007E1D0D /* Info.plist */,
 			);
 			path = Tests;
@@ -563,6 +571,7 @@
 				1FCDDA651EAC31EF006EB95B /* Action+Internal.swift in Sources */,
 				1FCDDA661EAC31EF006EB95B /* AlertAction.swift in Sources */,
 				1FCDDA671EAC31EF006EB95B /* UIBarButtonItem+Action.swift in Sources */,
+				CA2861CB1ED6B08300BB327A /* InputSubject.swift in Sources */,
 				1FCDDA681EAC31EF006EB95B /* UIButton+Rx.swift in Sources */,
 				1FCDDA691EAC31EF006EB95B /* UIControl+Rx.swift in Sources */,
 			);
@@ -572,6 +581,7 @@
 			isa = PBXSourcesBuildPhase;
 			buildActionMask = 2147483647;
 			files = (
+				CA2861CC1ED6B08400BB327A /* InputSubject.swift in Sources */,
 				1FCDDA8A1EAC329E006EB95B /* Action.swift in Sources */,
 				1FCDDA8B1EAC329E006EB95B /* Action+Internal.swift in Sources */,
 			);
@@ -593,6 +603,7 @@
 				7F0569F61DE288EB007E1D0D /* AlertActionTests.swift in Sources */,
 				7F0569F51DE288EB007E1D0D /* ActionTests.swift in Sources */,
 				7F0569F81DE288EB007E1D0D /* ButtonTests.swift in Sources */,
+				CA2861CA1ED6A41700BB327A /* InputSubjectTests.swift in Sources */,
 			);
 			runOnlyForDeploymentPostprocessing = 0;
 		};
@@ -613,6 +624,7 @@
 				7F0569E81DE28587007E1D0D /* Action+Internal.swift in Sources */,
 				7F0569ED1DE28587007E1D0D /* UIBarButtonItem+Action.swift in Sources */,
 				7BD1C7551E1D5562000D82DA /* UIControl+Rx.swift in Sources */,
+				CA2861C81ED6979400BB327A /* InputSubject.swift in Sources */,
 				7F0569EA1DE28587007E1D0D /* Action.swift in Sources */,
 				7F0569EE1DE28587007E1D0D /* UIButton+Rx.swift in Sources */,
 			);

Sources/Action/Action.swift

@@ -27,7 +27,7 @@ public final class Action<Input, Element> {
     /// This subject also includes inputs as aguments of execute().
     /// All inputs are always appear in this subject even if the action is not enabled.
     /// Thus, inputs count equals elements count + errors count.
-    public let inputs = PublishSubject<Input>()
+    public let inputs = InputSubject<Input>()
 
     /// Errors aggrevated from invocations of execute().
     /// Delivered on whatever scheduler they were sent from.

Sources/Action/InputSubject.swift

@@ -0,0 +1,93 @@
+import Foundation
+import RxSwift
+
+/// A special subject for Action.inputs. It only emits `.next` event.
+public class InputSubject<Element>: ObservableType, Cancelable, SubjectType, ObserverType {
+
+    public typealias E = Element
+    typealias Key = UInt
+
+    /// Indicates whether the subject has any observers
+    public var hasObservers: Bool {
+        _lock.lock()
+        let count = _observers.count > 0
+        _lock.unlock()
+        return count
+    }
+    
+    // state
+    private let _lock = NSRecursiveLock()
+    private var _nextKey: Key = 0
+    private var _observers: [Key: (Event<Element>) -> ()] = [:]
+    private var _isDisposed = false
+
+    /// Indicates whether the subject has been isDisposed.
+    public var isDisposed: Bool {
+        _lock.lock()
+        let isDisposed = _isDisposed
+        _lock.unlock()
+        return isDisposed
+    }
+
+    /// Creates a subject.
+    public init() {
+        #if TRACE_RESOURCES
+        _ = Resources.incrementTotal()
+        #endif
+    }
+
+    /// Notifies all subscribed observers abount only `.next` event.
+    ///
+    /// - parameter event: Event to send to the observers.
+    public func on(_ event: Event<Element>) {
+        _lock.lock()
+        switch event {
+        case .next(_) where !_isDisposed:
+            _observers.values.forEach { $0(event) }
+        default:
+            break
+        }
+        _lock.unlock()
+    }
+
+    /**
+     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 func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
+        _lock.lock()
+
+        if _isDisposed {
+            observer.on(.error(RxError.disposed(object: self)))
+            return Disposables.create()
+        }
+        
+        let key = _nextKey
+        _nextKey += 1
+        _observers[key] = observer.on
+        _lock.unlock()
+
+        return Disposables.create { [weak self] in
+            self?._lock.lock()
+            self?._observers.removeValue(forKey: key)
+            self?._lock.unlock()
+        }
+    }
+    
+    /// Unsubscribe all observers and release resources.
+    public func dispose() {
+        _lock.lock()
+        _isDisposed = true
+        _observers.removeAll()
+        _lock.unlock()
+    }
+
+    #if TRACE_RESOURCES
+    deinit {
+        _ = Resources.decrementTotal()
+    }
+    #endif
+
+}

Tests/InputSubjectTests.swift

@@ -0,0 +1,113 @@
+import Quick
+import Nimble
+import RxSwift
+import RxTest
+import Action
+
+class InputSubjectTests: QuickSpec {
+    override func spec() {
+        var scheduler: TestScheduler!
+        var disposeBag: DisposeBag!
+
+        beforeEach {
+            scheduler = TestScheduler(initialClock: 0)
+            disposeBag = DisposeBag()
+        }
+
+        describe("Disposable observable") {
+            it("observables can be dispose") {
+                let subject = InputSubject<Int>()
+                let disposable1 = subject.subscribe()
+                let disposable2 = subject.subscribe()
+                expect(subject.hasObservers).to(beTrue())
+                disposable2.dispose()
+                expect(subject.hasObservers).to(beTrue())
+                disposable1.dispose()
+                expect(subject.hasObservers).to(beFalse())
+            }
+
+            it("dispose all observables") {
+                let subject = InputSubject<Int>()
+                _ = subject.subscribe()
+                _ = subject.subscribe()
+                expect(subject.hasObservers).to(beTrue())
+                subject.dispose()
+                expect(subject.hasObservers).to(beFalse())
+                expect(subject.isDisposed).to(beTrue())
+            }
+        }
+
+        describe("emit events") {
+            it("emit .next events") {
+                let subject = InputSubject<Int>()
+                let observer = scheduler.createObserver(Int.self)
+                subject.asObservable()
+                    .bind(to: observer)
+                    .disposed(by: disposeBag)
+                scheduler.scheduleAt(10) { subject.onNext(1) }
+                scheduler.scheduleAt(20) { subject.onNext(2) }
+                scheduler.scheduleAt(30) { subject.onNext(3) }
+                scheduler.start()
+
+                XCTAssertEqual(observer.events, [
+                    next(10, 1),
+                    next(20, 2),
+                    next(30, 3)
+                ])
+            }
+
+            it("ignore .error events") {
+                let subject = InputSubject<Int>()
+                let observer = scheduler.createObserver(Int.self)
+                subject.asObservable()
+                    .bind(to: observer)
+                    .disposed(by: disposeBag)
+                scheduler.scheduleAt(10) { subject.onNext(1) }
+                scheduler.scheduleAt(20) { subject.onError(TestError) }
+                scheduler.scheduleAt(30) { subject.onNext(3) }
+                scheduler.start()
+
+                XCTAssertEqual(observer.events, [
+                    next(10, 1),
+                    next(30, 3)
+                ])
+            }
+
+            it("ignore .completed events") {
+                let subject = InputSubject<Int>()
+                let observer = scheduler.createObserver(Int.self)
+                subject.asObservable()
+                    .bind(to: observer)
+                    .disposed(by: disposeBag)
+                scheduler.scheduleAt(10) { subject.onNext(1) }
+                scheduler.scheduleAt(20) { subject.onCompleted() }
+                scheduler.scheduleAt(30) { subject.onNext(3) }
+                scheduler.start()
+
+                XCTAssertEqual(observer.events, [
+                    next(10, 1),
+                    next(30, 3)
+                ])
+            }
+
+            it("event does not fire on disposed subject") {
+                let subject = InputSubject<Int>()
+                let observer = scheduler.createObserver(Int.self)
+                subject.asObservable()
+                    .bind(to: observer)
+                    .disposed(by: disposeBag)
+                scheduler.scheduleAt(10) { subject.onNext(1) }
+                scheduler.scheduleAt(20) { subject.onNext(2) }
+                scheduler.scheduleAt(30) { subject.dispose() }
+                scheduler.scheduleAt(40) { subject.onNext(4) }
+                scheduler.start()
+
+                XCTAssertEqual(observer.events, [
+                    next(10, 1),
+                    next(20, 2),
+                ])
+            }
+        }
+
+    }
+}