 ///
 /// `Interrupted` events on inner producers will be treated like `Completed`
 /// events on inner producers.
-public func flatten<T, E>(strategy: FlattenStrategy)(producer: Signal<SignalProducer<T, E>, E>) -> Signal<T, E> {
-	switch strategy {
-	case .Merge:
-		return producer |> merge
-
-	case .Concat:
-		return producer |> concat
-
-	case .Latest:
-		return producer |> switchToLatest
+public func flatten<T, E>(strategy: FlattenStrategy) -> Signal<SignalProducer<T, E>, E> -> Signal<T, E> {
+	return { signal in
+		switch strategy {
+		case .Merge:
+			return signal |> merge
+
+		case .Concat:
+			return signal |> concat
+
+		case .Latest:
+			return signal |> switchToLatest
+		}
 	}
 }
 

/// Returns a signal that will send values from `signal`, sampled (possibly
/// multiple times) by `sampler`, then complete once both input signals have
/// completed, or interrupt if either input signal is interrupted.
public func sampleOn<T, E>(sampler: Signal<(), NoError>)(signal: Signal<T, E>) -> Signal<T, E> {
	return Signal { observer in
		let state = Atomic(SampleState<T>())

		let signalDisposable = signal.observe(next: { value in
			state.modify { (var st) in
				st.latestValue = value
				return st
			}

			return
		}, error: { error in
			sendError(observer, error)
		}, completed: {
			let oldState = state.modify { (var st) in
				st.signalCompleted = true
				return st
			}

			if oldState.samplerCompleted {
				sendCompleted(observer)
			}
		}, interrupted: {
			sendInterrupted(observer)
		})

		let samplerDisposable = sampler.observe(next: { _ in
			if let value = state.value.latestValue {
				sendNext(observer, value)
			}
		}, completed: {
			let oldState = state.modify { (var st) in
				st.samplerCompleted = true
				return st
			}

			if oldState.signalCompleted {
				sendCompleted(observer)
			}
		}, interrupted: {
			sendInterrupted(observer)
		})

		return CompositeDisposable(ignoreNil([ signalDisposable, samplerDisposable ]))
	}
}

/// Forwards events from `signal` until `trigger` sends a Next or Completed
/// event, at which point the returned signal will complete.
public func takeUntil<T, E>(trigger: Signal<(), NoError>)(signal: Signal<T, E>) -> Signal<T, E> {
	return Signal { observer in
		let signalDisposable = signal.observe(observer)
		let triggerDisposable = trigger.observe(Signal.Observer { event in
			switch event {
			case .Next, .Completed:
				sendCompleted(observer)

			case .Error, .Interrupted:
				break
			}
		})

		return CompositeDisposable(ignoreNil([ signalDisposable, triggerDisposable ]))
	}
}



/// Forwards only those values from `signal` which do not pass `isRepeat` with
/// respect to the previous value. The first value is always forwarded.
public func skipRepeats<T, E>(isRepeat: (T, T) -> Bool) -> Signal<T, E> -> Signal<T, E> {
	return { signal in
		return signal
			|> map { Optional($0) }
			|> combinePrevious(nil)
			|> filter { (a, b) in
				if let a = a, b = b where isRepeat(a, b) {
					return false
				} else {
					return true
				}
			}
			|> map { $0.1! }
	}
}

/// Does not forward any values from `signal` until `predicate` returns false,
/// at which point the returned signal behaves exactly like `signal`.
public func skipWhile<T, E>(predicate: T -> Bool) -> Signal<T, E> -> Signal<T, E> {
	return { signal in
		return Signal { observer in
			var shouldSkip = true

			return signal.observe(Signal.Observer { event in
				switch event {
				case let .Next(value):
					shouldSkip = shouldSkip && predicate(value.value)
					if !shouldSkip {
						fallthrough
					}

				default:
					observer.put(event)
				}
			})
		}
	}
}

