Merge branch 'master' of github.com:DeclarativeHub/ReactiveKit

README.md

@@ -999,7 +999,7 @@ At this point you might have the idea how to achieve the behaviour of the `share
 
 ### Connectable signals
 
-We have see two kinds of signals so far. A `Signal` that produces events only if the observer is registered and a `Subject` that produces events regardless if there are any observers registered. A connectable signals will be third kind of a signal we will implement. This one will start producing events when we call `connect()` on it. Let us define a protocol first.
+We have seen two kinds of signals so far. A `Signal` that produces events only if the observer is registered and a `Subject` that produces events regardless if there are any observers registered. A connectable signal will be the third kind of a signal we will implement. This one will start producing events when we call `connect()` on it. Let us define a protocol first.
 
 ```swift
 /// Represents a signal that is started by calling `connect` on it.

Sources/SignalProtocol.swift

@@ -769,12 +769,13 @@ public extension SignalProtocol {
     }
   }
 
-  /// Emit elements of the reciver until given signal completes and then complete the receiver.
+  /// Emit elements of the receiver until the given signal sends an event (of any kind)
+  /// and then completes the receiver (subsequent events on the receiver are ignored).
   public func take<S: SignalProtocol>(until signal: S) -> Signal<Element, Error> {
     return Signal { observer in
       let disposable = CompositeDisposable()
-
-      disposable += signal.observe { event in
+      
+      disposable += signal.observe { _ in
         observer.completed()
       }
 

Tests/ReactiveKitTests/SignalTests.swift

@@ -217,6 +217,28 @@ class SignalTests: XCTestCase {
     let takenLast2 = operation.take(last: 2)
     takenLast2.expectComplete(after: [2, 3])
   }
+
+  func testTakeUntil() {
+    let bob = Scheduler()
+    let eve = Scheduler()
+
+    let operation = Signal<Int, TestError>.sequence([1, 2, 3, 4]).observeIn(bob.context)
+    let interrupt = Signal<String, TestError>.sequence(["A", "B"]).observeIn(eve.context)
+
+    let takeuntil = operation.take(until: interrupt)
+
+    let exp = expectation(description: "completed")
+    takeuntil.expectAsyncComplete(after: [1, 2], expectation: exp)
+
+    bob.runOne()                // Sends 1.
+    bob.runOne()                // Sends 2.
+    eve.runOne()                // Sends A, effectively stopping the receiver.
+    bob.runOne()                // Ignored.
+    eve.runRemaining()          // Ignored. Sends B, with termination.
+    bob.runRemaining()          // Ignored.
+
+    waitForExpectations(timeout: 1, handler: nil)
+  }
 
 //  func testThrottle() {
 //    let operation = Signal<Int, TestError>.interval(0.4, queue: Queue.global).take(5)