Merge pull request #65 from jechol/fix-scan

Fix scan to emit initial value

Sources/RawStream.swift

@@ -162,6 +162,7 @@ public extension RawStreamType {
   public func scan<U: EventType>(initial: U, _ combine: (U, Event) -> U) -> RawStream<U> {
     return RawStream<U> { observer in
       var accumulator = initial
+      observer.observer(accumulator)
       return self.observe { event in
         accumulator = combine(accumulator, event)
         observer.observer(accumulator)
@@ -775,10 +776,7 @@ public extension RawStreamType {
   /// Reduce stream events to a single event by applying given function on each emission.
   @warn_unused_result
   public func reduce<U: EventType>(initial: U, _ combine: (U, Event) -> U) -> RawStream<U> {
-    return RawStream<U> { observer in
-      observer.observer(initial)
-      return self.scan(initial, combine).observe(observer.observer)
-      }.takeLast()
+    return scan(initial, combine).takeLast()
   }
 }
 

Tests/OperationTests.swift

@@ -77,7 +77,7 @@ class OperatorsTests: XCTestCase {
   func testScan() {
     let operation = Operation<Int, TestError>.sequence([1, 2, 3])
     let scanned = operation.scan(0, +)
-    scanned.expectNext([1, 3, 6])
+    scanned.expectNext([0, 1, 3, 6])
   }
 
   func testToOperation() {

Tests/StreamTests.swift

@@ -68,7 +68,7 @@ class StreamTests: XCTestCase {
   func testScan() {
     let stream = Stream.sequence([1, 2, 3])
     let scanned = stream.scan(0, +)
-    scanned.expectNext([1, 3, 6])
+    scanned.expectNext([0, 1, 3, 6])
   }
 
   func testToOperation() {