 
 /// Returns a signal that will yield the first `count` values from the
 /// input signal.
-public func take<T, E>(count: Int)(signal: Signal<T, E>) -> Signal<T, E> {
-	precondition(count >= 0)
-
-	return Signal { observer in
-		if count == 0 {
-			sendCompleted(observer)
-			return nil
-		}
-
-		var taken = 0
-
-		return signal.observe(Signal.Observer { event in
-			switch event {
-			case let .Next(value):
-				if taken < count {
-					taken++
-					sendNext(observer, value.value)
-				}
-
-				if taken == count {
-					sendCompleted(observer)
-				}
-
-			default:
-				observer.put(event)
+public func take<T, E>(count: Int) -> Signal<T, E> -> Signal<T, E> {
+	return { signal in
+		precondition(count >= 0)
+
+		return Signal { observer in
+			if count == 0 {
+				sendCompleted(observer)
+				return nil
 			}
-		})
+
+			var taken = 0
+
+			return signal.observe(Signal.Observer { event in
+				switch event {
+				case let .Next(value):
+					if taken < count {
+						taken++
+						sendNext(observer, value.value)
+					}
+
+					if taken == count {
+						sendCompleted(observer)
+					}
+
+				default:
+					observer.put(event)
+				}
+			})
+		}
 	}
 }
 

}

/// Maps each value in the signal to a new value.
public func map<T, U, E>(transform: T -> U)(signal: Signal<T, E>) -> Signal<U, E> {
	return Signal { observer in
		return signal.observe(Signal.Observer { event in
			observer.put(event.map(transform))
		})
	}
}

/// Maps errors in the signal to a new error.
public func mapError<T, E, F>(transform: E -> F)(signal: Signal<T, E>) -> Signal<T, F> {
	return Signal { observer in
		return signal.observe(Signal.Observer { event in
			switch event {
			case let .Next(value):
				sendNext(observer, value.value)
			case let .Error(error):
				sendError(observer, transform(error.value))
			case .Completed:
				sendCompleted(observer)
			case .Interrupted:
				sendInterrupted(observer)
			}
		})
	}
}

/// Preserves only the values of the signal that pass the given predicate.
public func filter<T, E>(predicate: T -> Bool)(signal: Signal<T, E>) -> Signal<T, E> {
	return Signal { observer in
		return signal.observe(Signal.Observer { event in
			switch event {
			case let .Next(value):
				if predicate(value.value) {
					sendNext(observer, value.value)
				}

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



/// Zips elements of two signals into pairs. The elements of any Nth pair
/// are the Nth elements of the two input signals.
public func zipWith<T, U, E>(otherSignal: Signal<U, E>) -> Signal<T, E> -> Signal<(T, U), E> {
	return { signal in
		return Signal { observer in
			let initialStates: (ZipState<T>, ZipState<U>) = (ZipState(), ZipState())
			let states: Atomic<(ZipState<T>, ZipState<U>)> = Atomic(initialStates)
			
			let flush = { () -> () in
				var originalStates: (ZipState<T>, ZipState<U>)!
				states.modify { states in
					originalStates = states
					
					var updatedStates = states
					let extractCount = min(states.0.values.count, states.1.values.count)
					
					removeRange(&updatedStates.0.values, 0 ..< extractCount)
					removeRange(&updatedStates.1.values, 0 ..< extractCount)
					return updatedStates
				}
				
				while !originalStates.0.values.isEmpty && !originalStates.1.values.isEmpty {
					let left = originalStates.0.values.removeAtIndex(0)
					let right = originalStates.1.values.removeAtIndex(0)
					sendNext(observer, (left, right))
				}
				
				if originalStates.0.isFinished || originalStates.1.isFinished {
					sendCompleted(observer)
				}
			}
			
			let onError = { sendError(observer, $0) }
			let onInterrupted = { sendInterrupted(observer) }
			
			let signalDisposable = signal.observe(next: { value in
				states.modify { (var states) in
					states.0.values.append(value)
					return states
				}
				
				flush()
			}, error: onError, completed: {
				states.modify { (var states) in
					states.0.completed = true
					return states
				}
					
				flush()
			}, interrupted: onInterrupted)
			
			let otherDisposable = otherSignal.observe(next: { value in
				states.modify { (var states) in
					states.1.values.append(value)
					return states
				}
				
				flush()
			}, error: onError, completed: {
				states.modify { (var states) in
					states.1.completed = true
					return states
				}
					
				flush()
			}, interrupted: onInterrupted)
			
			return CompositeDisposable(ignoreNil([ signalDisposable, otherDisposable ]))
		}
	}
}

internal func zipWith<T, E>(otherSignal: Signal<T, E>) -> Signal<[T], E> -> Signal<[T], E> {
	return { signal in
		return signal |> zipWith(otherSignal) |> map { $0.0 + [$0.1] }
	}
}

/// Applies `operation` to values from `signal` with `Success`ful results
/// forwarded on the returned signal and `Failure`s sent as `Error` events.
public func try<T, E>(operation: T -> Result<(), E>) -> Signal<T, E> -> Signal<T, E> {
	return { signal in
		return signal |> tryMap { value in
			return operation(value).map {
				return value
			}
		}
	}
}

/// Applies `operation` to values from `signal` with `Success`ful results mapped
/// on the returned signal and `Failure`s sent as `Error` events.
public func tryMap<T, U, E>(operation: T -> Result<U, E>) -> Signal<T, E> -> Signal<U, E> {
	return { signal in
		return Signal { observer in
			signal.observe(next: { value in
				operation(value).analysis(ifSuccess: { value in
					sendNext(observer, value)
				}, ifFailure: { error in
					sendError(observer, error)
				})
			}, error: { error in
				sendError(observer, error)
			}, completed: {
				sendCompleted(observer)
			}, interrupted: {
				sendInterrupted(observer)
			})
		}
	}
}

