Added take(x) operator & unit tests

DeclarativeHub · Mar 5, 2016 · b5df08c · b5df08c
1 parent 6a5a87b
commit b5df08c
Show file tree

Hide file tree

Showing 3 changed files with 101 additions and 1 deletion.
diff --git a/ReactiveKit.xcodeproj/project.pbxproj b/ReactiveKit.xcodeproj/project.pbxproj
@@ -199,7 +199,7 @@
 		ECBCCE011BEB6BBE00723476 /* Queue.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = Queue.swift; sourceTree = "<group>"; };
 		ECBCCE041BEB6BBE00723476 /* ActiveStream.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = ActiveStream.swift; sourceTree = "<group>"; };
 		ECBCCE051BEB6BBE00723476 /* Stream.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = Stream.swift; sourceTree = "<group>"; };
-		ECBCCE061BEB6BBE00723476 /* StreamType.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = StreamType.swift; sourceTree = "<group>"; };
+		ECBCCE061BEB6BBE00723476 /* StreamType.swift */ = {isa = PBXFileReference; fileEncoding = 4; indentWidth = 2; lastKnownFileType = sourcecode.swift; path = StreamType.swift; sourceTree = "<group>"; tabWidth = 2; };
 		ECBCCE2A1BEB6BE100723476 /* NoError.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; name = NoError.swift; path = ../Operation/NoError.swift; sourceTree = "<group>"; };
 		ECBCCE2B1BEB6BE100723476 /* Stream+Operation.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = "Stream+Operation.swift"; sourceTree = "<group>"; };
 		ECBCCE2C1BEB6BE100723476 /* Operation.swift */ = {isa = PBXFileReference; fileEncoding = 4; indentWidth = 2; lastKnownFileType = sourcecode.swift; path = Operation.swift; sourceTree = "<group>"; tabWidth = 2; };

diff --git a/ReactiveKit/Streams/StreamType.swift b/ReactiveKit/Streams/StreamType.swift
@@ -136,6 +136,31 @@ extension StreamType {
     }
   }
 
+  @warn_unused_result
+  public func take(count:Int) -> Stream<Self.Event> {
+    return create { observer in
+
+      var remainder = count
+      let outerDisposable = CompositeDisposable()
+
+      let disposable = self.observe(on: nil, observer: { event in
+        if remainder > 0{
+          remainder--
+
+          observer(event)
+        }
+        if remainder == 0{
+          outerDisposable.dispose()
+        }
+      })
+
+      outerDisposable.addDisposable(disposable)
+      return outerDisposable
+    }
+  }
+
+
+
   @warn_unused_result
   public func skip(var count: Int) -> Stream<Event> {
     return create { observer in

diff --git a/ReactiveKitTests/StreamSpec.swift b/ReactiveKitTests/StreamSpec.swift
@@ -184,6 +184,81 @@ class StreamSpec: QuickSpec {
           }
         }
       }
+
+      context("when taking 0") {
+        var observedEvents: [Int] = []
+        var disposable: DisposableType!
+
+        beforeEach {
+          disposable = stream.take(0).observe(on: ImmediateExecutionContext) {
+            observedEvents.append($0)
+          }
+        }
+
+        it("does takes nothing") {
+          expect(observedEvents) == []
+        }
+
+        describe("can be disposed") {
+          beforeEach {
+            disposable.dispose()
+          }
+
+          it("is disposed") {
+            expect(simpleDisposable.isDisposed).to(beTrue())
+          }
+        }
+      }
+
+      context("when taking 1") {
+        var observedEvents: [Int] = []
+        var disposable: DisposableType!
+
+        beforeEach {
+          disposable = stream.take(1).observe(on: ImmediateExecutionContext) {
+            observedEvents.append($0)
+          }
+        }
+
+        it("does take only 1") {
+          expect(observedEvents) == [1]
+        }
+
+        describe("can be disposed") {
+          beforeEach {
+            disposable.dispose()
+          }
+
+          it("is disposed") {
+            expect(simpleDisposable.isDisposed).to(beTrue())
+          }
+        }
+      }
+
+      context("when taking 2") {
+        var observedEvents: [Int] = []
+        var disposable: DisposableType!
+
+        beforeEach {
+          disposable = stream.take(2).observe(on: ImmediateExecutionContext) {
+            observedEvents.append($0)
+          }
+        }
+
+        it("does takes 1 and 2") {
+          expect(observedEvents) == [1,2]
+        }
+
+        describe("can be disposed") {
+          beforeEach {
+            disposable.dispose()
+          }
+
+          it("is disposed") {
+            expect(simpleDisposable.isDisposed).to(beTrue())
+          }
+        }
+      }
 
       context("when skip by 1") {
         var observedEvents: [Int] = []