 	return Signal { observer in
 		var taken = 0
 
-		return signal.observe(next: { value in
-			if taken < count {
-				taken++
-				sendNext(observer, value)
-			} else {
-				sendCompleted(observer)
-			}
-		}, error: { error in
-			sendError(observer, error)
-		}, completed: {
-			sendCompleted(observer)
-		})
-	}
-}
-
-/// Forwards all events onto the given scheduler, instead of whichever
-/// scheduler they originally arrived upon.
+		return signal.observe(Signal.Observer { event in
+			switch event {
+			case let .Next(value):
+				if taken < count {
+					taken++
+					fallthrough
+				} else {
+					sendCompleted(observer)
+				}
+
+			default:
+				observer.put(event)
+			}
+		})
+	}
+}
+
+/// Forwards all events onto the given scheduler, instead of whichever scheduler
+/// they originally arrived upon.
 public func observeOn<T, E>(scheduler: SchedulerType)(signal: Signal<T, E>) -> Signal<T, E> {
 	return Signal { observer in
 		return signal.observe(SinkOf { event in

			if oldState.signalCompleted {
				sendCompleted(observer)
			}
		})

		return CompositeDisposable([ signalDisposable, samplerDisposable ])

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