Merge pull request #11 from neilpa/uncollect

Uncollect

README.md

@@ -1,5 +1,5 @@
 # Rex [![Carthage compatible](https://img.shields.io/badge/Carthage-compatible-4BC51D.svg?style=flat)](https://github.com/Carthage/Carthage)
-Additional operators for [ReactiveCocoa](https://github.com/ReactiveCocoa/ReactiveCocoa) that may not fit in the core framework.
+Extensions for [ReactiveCocoa](https://github.com/ReactiveCocoa/ReactiveCocoa) that may not fit in the core framework.
 
 ## Signal
 All `Signal` operators can also be lifted to`SignalProducer`.
@@ -19,6 +19,15 @@ func ignoreError<T, E>(signal: Signal<T, E>) -> Signal<T, NoError>
 func ignoreError<T, E>(#replacement: Event<T, NoError>)(signal: Signal<T, E>) -> Signal<T, NoError>
 ```
 
+##### `uncollect`
+
+Flattens batches of elements sent on `signal` into each individual element. The inverse of `collect`.
+
+```swift
+func uncollect<S: SequenceType, E>(signal: Signal<S, E>) -> Signal<S.Generator.Element, E>
+```
+
+
 ## SignalProducer
 Operators specific to `SignalProducer`.
 

Rex.xcodeproj/project.pbxproj

@@ -163,10 +163,10 @@
 			isa = PBXGroup;
 			children = (
 				CCF8C8F41AFFED9E00FF444A /* FoundationExtensions.swift */,
+				7DE995CA1B03A56800CA9420 /* Operators.swift */,
 				D8003EBC1AFED01000D7D3C5 /* Signal.swift */,
 				D8003EB81AFEC7A900D7D3C5 /* SignalProducer.swift */,
 				D8003E911AFEC3D400D7D3C5 /* Supporting Files */,
-				7DE995CA1B03A56800CA9420 /* Operators.swift */,
 			);
 			path = Source;
 			sourceTree = "<group>";

Source/Signal.swift

@@ -50,3 +50,21 @@ public func ignoreError<T, E>(#replacement: Event<T, NoError>)(signal: Signal<T,
         })
     }
 }
+
+/// Returns a signal that flattens sequences of elements. The inverse of `collect`.
+public func uncollect<S: SequenceType, E>(signal: Signal<S, E>) -> Signal<S.Generator.Element, E> {
+    return Signal { observer in
+        return signal.observe(Signal.Observer { event in
+            switch event {
+            case let .Next(sequence):
+                map(sequence.value) { sendNext(observer, $0) }
+            case let .Error(error):
+                sendError(observer, error.value)
+            case .Completed:
+                sendCompleted(observer)
+            case .Interrupted:
+                sendInterrupted(observer)
+            }
+        })
+    }
+}

Tests/SignalTests.swift

@@ -68,5 +68,24 @@ final class SignalTests: XCTestCase {
         NSError() --> sink
         XCTAssertTrue(interrupted)
     }
-
+
+    func testUncollect() {
+        let (signal, sink) = Signal<[Int], NoError>.pipe()
+        var values: [Int] = []
+
+        signal
+            |> uncollect
+            |> observe(next: {
+                values.append($0)
+            })
+
+        [] --> sink
+        XCTAssert(values.isEmpty)
+
+        [1] --> sink
+        XCTAssert(values == [1])
+
+        [2, 3] --> sink
+        XCTAssert(values == [1, 2, 3])
+    }
 }