 
 /// Returns a signal that will yield an array of values when `signal` completes.
 public func collect<T, E>(signal: Signal<T, E>) -> Signal<[T], E> {
-	return signal |> reduce(CollectState()) { $0.append($1) } |> map { $0.values }
+	return signal.collect()
+}
+
+extension SignalType {
+	public func collect() -> Signal<[T], E> {
+		return signal
+			.reduce(CollectState()) { $0.append($1) }
+			.map { $0.values }
+	}
 }
 
 /// Forwards all events onto the given scheduler, instead of whichever
 /// scheduler they originally arrived upon.
 public func observeOn<T, E>(scheduler: SchedulerType) -> Signal<T, E> -> Signal<T, E> {
-	return { signal in
-		return Signal { observer in
-			return signal.observe { event in
+	return { $0.observeOn(scheduler) }
+}
+
+extension SignalType {
+	public func observeOn(scheduler: SchedulerType) -> Signal<T, E> {
+		return Signal { observer in
+			return self.signal.observe { event in
 				scheduler.schedule {
 					observer(event)
 				}

/// Waits until `signal` completes and then forwards the final `count` values
/// on the returned signal.
public func takeLast<T, E>(count: Int) -> Signal<T, E> -> Signal<T, E> {
	return { signal in
		return Signal { observer in
			var buffer = [T]()
			buffer.reserveCapacity(count)

			return signal.observe(next: { value in
				// To avoid exceeding the reserved capacity of the buffer, we remove then add.
				// Remove elements until we have room to add one more.
				while (buffer.count + 1) > count {

///
/// `Error` and `Interrupted` events are always scheduled immediately.
public func delay<T, E>(interval: NSTimeInterval, onScheduler scheduler: DateSchedulerType) -> Signal<T, E> -> Signal<T, E> {
	return { $0.delay(interval, onScheduler: scheduler) }
}

extension SignalType {
	public func delay(interval: NSTimeInterval, onScheduler scheduler: DateSchedulerType) -> Signal<T, E> {
		precondition(interval >= 0)

		return Signal { observer in
			return self.signal.observe { event in
				switch event {
				case .Error, .Interrupted:
					scheduler.schedule {
