import XCTest import Foundation #if SSK_LEGACY @testable import SwiftSignalKitLegacy #else @testable import SwiftSignalKit2 #endif final class OperatorReduceLeftTests: XCTestCase { func testReduceLeftEmitsAccumulatedValueOnCompletion() { let log = OperatorLog() let a = OperatorSource("a", log) let disposable = operatorRecord(a.signal |> reduceLeft(value: 100, f: { $0 + $1 }), log) a.emit(1) a.emit(2) a.emit(3) XCTAssertEqual(log.events, ["a.subscribe#1"]) a.complete() disposable.dispose() XCTAssertEqual(log.events, ["a.subscribe#1", "next 106", "completed", "a.dispose#1"]) } func testReduceLeftWithoutValuesEmitsInitialValue() { let log = OperatorLog() operatorRecord(Signal.complete() |> reduceLeft(value: 7, f: { $0 + $1 }), log) XCTAssertEqual(log.events, ["next 7", "completed"]) } func testReduceLeftErrorDropsAccumulatedValue() { let log = OperatorLog() let a = OperatorSource("a", log) let b = OperatorSource("b", log) operatorRecord(a.signal |> reduceLeft(value: 0, f: { $0 + $1 }), log) a.emit(1) a.fail("e") let disposable = operatorRecord(b.signal |> reduceLeft(value: 0, f: { $0 + $1 }), log) b.emit(1) disposable.dispose() b.complete() XCTAssertEqual(log.events, ["a.subscribe#1", "error e", "a.dispose#1", "b.subscribe#1", "b.dispose#1"]) } func testReduceLeftAccumulatorIsPerSubscription() { let log = OperatorLog() let s = operatorSyncSource("s", log, values: [1, 2, 3], terminal: OperatorTerminal.complete) let signal = s.signal |> reduceLeft(value: 10, f: { $0 * $1 }) operatorRecord(signal, log) operatorRecord(signal, log) XCTAssertEqual(log.events, ["s.subscribe#1", "next 60", "completed", "s.dispose#1", "s.subscribe#2", "next 60", "completed", "s.dispose#2"]) } func testReduceLeftWithEmitCanEmitIntermediateValues() { let log = OperatorLog() let a = OperatorSource("a", log) let signal = a.signal |> reduceLeft(value: 0, f: { (current: Int, next: Int, emit: (Int) -> Void) -> Int in log.add("reduce \(current) \(next)") if next % 2 == 0 { emit(current) } return current + next }) let disposable = operatorRecord(signal, log) a.emit(1) a.emit(2) a.emit(3) a.emit(4) a.complete() disposable.dispose() XCTAssertEqual(log.events, [ "a.subscribe#1", "reduce 0 1", "reduce 1 2", "next 1", "reduce 3 3", "reduce 6 4", "next 6", "next 10", "completed", "a.dispose#1" ]) } func testReduceLeftWithEmitErrorAndDisposal() { let log = OperatorLog() let a = OperatorSource("a", log) let b = OperatorSource("b", log) let reducer = { (current: Int, next: Int, emit: (Int) -> Void) -> Int in emit(next) return current + next } operatorRecord(a.signal |> reduceLeft(value: 0, f: reducer), log) a.emit(5) a.fail("e") let disposable = operatorRecord(b.signal |> reduceLeft(value: 0, f: reducer), log) b.emit(6) disposable.dispose() b.complete() XCTAssertEqual(log.events, ["a.subscribe#1", "next 5", "error e", "a.dispose#1", "b.subscribe#1", "next 6", "b.dispose#1"]) } func testReduceLeftWithEmitOnSynchronousSource() { let log = OperatorLog() let s = operatorSyncSource("s", log, values: [1, 2], terminal: OperatorTerminal.complete) operatorRecord(s.signal |> reduceLeft(value: 0, f: { (current: Int, next: Int, emit: (Int) -> Void) -> Int in emit(-next) return current + next }), log) XCTAssertEqual(log.events, ["s.subscribe#1", "next -1", "next -2", "next 3", "completed", "s.dispose#1"]) } func testReduceLeftPassthroughRunsGeneratorsSequentiallyWithUpdatedValue() { let log = OperatorLog() let o = OperatorSource("o", log) let generated = OperatorBox), String>>() let signal = o.signal |> reduceLeft(0, generator: { (current: Int, next: Int) -> Signal<(Int, Passthrough), String> in log.add("generator \(current) \(next)") let source = OperatorSource<(Int, Passthrough), String>("g\(next)", log) generated.append(source) return source.signal }) let disposable = operatorRecord(signal, log) o.emit(1) o.emit(2) o.emit(3) generated.values[0].emit((10, .Some(100))) generated.values[0].emit((11, .None)) generated.values[0].complete() generated.values[1].emit((20, .Some(200))) o.complete() generated.values[1].complete() generated.values[2].complete() disposable.dispose() XCTAssertEqual(log.events, [ "o.subscribe#1", "generator 0 1", "g1.subscribe#1", "next 100", "generator 11 2", "g2.subscribe#1", "g1.dispose#1", "next 200", "o.dispose#1", "generator 20 3", "g3.subscribe#1", "g2.dispose#1", "next 20", "completed", "g3.dispose#1" ]) } func testReduceLeftPassthroughWithSynchronousGenerators() { let log = OperatorLog() let s = operatorSyncSource("s", log, values: [1, 2, 3], terminal: OperatorTerminal.complete) let signal = s.signal |> reduceLeft(0, generator: { (current: Int, next: Int) -> Signal<(Int, Passthrough), String> in let sum = current + next return .single((sum, next == 2 ? .None : .Some(sum * 10))) }) operatorRecord(signal, log) XCTAssertEqual(log.events, ["s.subscribe#1", "next 10", "next 60", "next 6", "completed", "s.dispose#1"]) } func testReduceLeftPassthroughCompletionWhileIdleEmitsCurrentValue() { let log = OperatorLog() let o = OperatorSource("o", log) operatorRecord(o.signal |> reduceLeft(5, generator: { (current: Int, next: Int) -> Signal<(Int, Passthrough), String> in return .single((current + next, .None)) }), log) o.complete() let p = OperatorSource("p", log) operatorRecord(p.signal |> reduceLeft(5, generator: { (current: Int, next: Int) -> Signal<(Int, Passthrough), String> in return .single((current + next, .None)) }), log) p.emit(1) p.emit(2) p.complete() XCTAssertEqual(log.events, ["o.subscribe#1", "next 5", "completed", "o.dispose#1", "p.subscribe#1", "next 8", "completed", "p.dispose#1"]) } func testReduceLeftPassthroughGeneratorErrorPropagatesAndSkipsQueued() { let log = OperatorLog() let o = OperatorSource("o", log) let g = OperatorSource<(Int, Passthrough), String>("g", log) let calls = OperatorCounter() operatorRecord(o.signal |> reduceLeft(0, generator: { (_: Int, _: Int) -> Signal<(Int, Passthrough), String> in calls.increment() return g.signal }), log) o.emit(1) o.emit(2) g.fail("gen") XCTAssertEqual(log.events, ["o.subscribe#1", "g.subscribe#1", "error gen", "g.dispose#1", "o.dispose#1"]) XCTAssertEqual(calls.value, 1) } func testReduceLeftPassthroughOuterErrorAndDisposal() { let log = OperatorLog() let o = OperatorSource("o", log) let g = OperatorSource<(Int, Passthrough), String>("g", log) operatorRecord(o.signal |> reduceLeft(0, generator: { (_: Int, _: Int) in g.signal }), log) o.emit(1) o.fail("outer") let p = OperatorSource("p", log) let disposable = operatorRecord(p.signal |> reduceLeft(0, generator: { (_: Int, _: Int) in g.signal }), log) p.emit(1) disposable.dispose() XCTAssertEqual(log.events, ["o.subscribe#1", "g.subscribe#1", "error outer", "g.dispose#1", "o.dispose#1", "p.subscribe#1", "g.subscribe#2", "g.dispose#2", "p.dispose#1"]) } } final class OperatorMaterializeTests: XCTestCase { private func describe(_ event: SignalEvent) -> String { switch event { case let .Next(value): return "Next(\(value))" case let .Error(error): return "Error(\(error))" case .Completion: return "Completion" } } private func recordEvents(_ signal: Signal, NoError>, _ log: OperatorLog) -> Disposable { return signal.start(next: { event in log.add("next \(self.describe(event))") }, completed: { log.add("completed") }) } func testDematerializeConvertsCompletionIntoEventThenCompletes() { let log = OperatorLog() let a = OperatorSource("a", log) let disposable = recordEvents(a.signal |> dematerialize, log) a.emit(1) a.emit(2) a.complete() disposable.dispose() XCTAssertEqual(log.events, ["a.subscribe#1", "next Next(1)", "next Next(2)", "next Completion", "completed", "a.dispose#1"]) } func testDematerializeConvertsErrorIntoEventThenCompletes() { let log = OperatorLog() let a = OperatorSource("a", log) let _ = recordEvents(a.signal |> dematerialize, log) a.emit(1) a.fail("e") XCTAssertEqual(log.events, ["a.subscribe#1", "next Next(1)", "next Error(e)", "completed", "a.dispose#1"]) } func testDematerializeDisposalAndSynchronousSource() { let log = OperatorLog() let a = OperatorSource("a", log) let disposable = recordEvents(a.signal |> dematerialize, log) disposable.dispose() let _ = recordEvents(Signal.single(3) |> dematerialize, log) let _ = recordEvents(Signal.fail("x") |> dematerialize, log) XCTAssertEqual(log.events, ["a.subscribe#1", "a.dispose#1", "next Next(3)", "next Completion", "completed", "next Error(x)", "completed"]) } func testMaterializeConvertsEventsBackIntoSignalTermination() { let log = OperatorLog() let a = OperatorSource, NoError>("a", log) let disposable = operatorRecord(a.signal |> materialize, log) a.emit(.Next(1)) a.emit(.Next(2)) a.emit(.Error("e")) a.emit(.Next(3)) disposable.dispose() let b = OperatorSource, NoError>("b", log) operatorRecord(b.signal |> materialize, log) b.emit(.Next(4)) b.emit(.Completion) let c = OperatorSource, NoError>("c", log) operatorRecord(c.signal |> materialize, log) c.emit(.Next(5)) c.complete() XCTAssertEqual(log.events, [ "a.subscribe#1", "next 1", "next 2", "error e", "a.dispose#1", "b.subscribe#1", "next 4", "completed", "b.dispose#1", "c.subscribe#1", "next 5", "completed", "c.dispose#1" ]) } func testMaterializeDematerializeRoundTrip() { let log = OperatorLog() let s = operatorSyncSource("s", log, values: [1, 2], terminal: OperatorTerminal.fail("boom")) operatorRecord(s.signal |> dematerialize |> materialize, log) XCTAssertEqual(log.events, ["s.subscribe#1", "next 1", "next 2", "error boom", "s.dispose#1"]) } } final class OperatorFeedbackLoopTests: XCTestCase { func testLoopsWithInitialStateUntilOnceReturnsNilWithoutEmittingValues() { let log = OperatorLog() let sources = [OperatorSource("l1", log), OperatorSource("l2", log)] let calls = OperatorCounter() let reduceCalls = OperatorCounter() let signal: Signal = feedbackLoop(once: { (state: SignalFeedbackLoopState) -> Signal? in let call = calls.increment() switch state { case .initial: log.add("once \(call) initial") case let .loop(value): log.add("once \(call) loop \(value)") } if call <= sources.count { return sources[call - 1].signal } return nil }, reduce: { lhs, rhs in reduceCalls.increment() return lhs + rhs }) let disposable = operatorRecord(signal, log) sources[0].emit(1) sources[0].emit(2) sources[0].complete() sources[1].emit(3) sources[1].complete() disposable.dispose() XCTAssertEqual(reduceCalls.value, 0) XCTAssertEqual(log.events, [ "once 1 initial", "l1.subscribe#1", "once 2 initial", "l2.subscribe#1", "l1.dispose#1", "once 3 initial", "completed", "l2.dispose#1" ]) } func testOnceReturningNilImmediatelyCompletes() { let log = OperatorLog() let signal: Signal = feedbackLoop(once: { (_: SignalFeedbackLoopState) -> Signal? in log.add("once") return nil }, reduce: { $0 + $1 }) operatorRecord(signal, log) XCTAssertEqual(log.events, ["once", "completed"]) } func testErrorPropagatesAndStopsLooping() { let log = OperatorLog() let source = OperatorSource("l", log) let calls = OperatorCounter() let signal: Signal = feedbackLoop(once: { (_: SignalFeedbackLoopState) -> Signal? in calls.increment() return source.signal }, reduce: { $0 + $1 }) operatorRecord(signal, log) source.emit(1) source.fail("e") XCTAssertEqual(log.events, ["l.subscribe#1", "error e", "l.dispose#1"]) XCTAssertEqual(calls.value, 1) } func testDisposalDisposesCurrentIterationAndStopsLooping() { let log = OperatorLog() let source = OperatorSource("l", log) let calls = OperatorCounter() let signal: Signal = feedbackLoop(once: { (_: SignalFeedbackLoopState) -> Signal? in calls.increment() return source.signal }, reduce: { $0 + $1 }) let disposable = operatorRecord(signal, log) disposable.dispose() source.complete() XCTAssertEqual(log.events, ["l.subscribe#1", "l.dispose#1"]) XCTAssertEqual(calls.value, 1) } func testSynchronouslyCompletingIterations() { let log = OperatorLog() let calls = OperatorCounter() let signal: Signal = feedbackLoop(once: { (_: SignalFeedbackLoopState) -> Signal? in let call = calls.increment() log.add("once \(call)") if call <= 3 { return .single(call) } return nil }, reduce: { $0 + $1 }) operatorRecord(signal, log) XCTAssertEqual(log.events, ["once 1", "once 2", "once 3", "once 4", "completed"]) } }