 		let producer = self.init { observer, disposable in
 			// Assigned to when replay() is invoked synchronously below.
 			var token: RemovalToken?
-
-			let replay = {
-				let originalState = state.modify { state in
+			
+			let replayBuffer = ReplayBuffer<Value>()
+			var replayValues: [Value] = []
+			var replayToken: RemovalToken?
+			var next = state.modify { state in
+				var state = state
+				replayValues = state.values
+				if replayValues.isEmpty {
+					token = state.observers?.insert(observer)
+				} else {
+					replayToken = state.replayBuffers?.insert(replayBuffer)
+				}
+				return state
+			}
+			
+			while !replayValues.isEmpty {
+				replayValues.forEach(observer.sendNext)
+				
+				if next.terminationEvent != nil { break }
+				
+				next = state.modify { state in
 					var state = state
-					token = state.observers?.insert(observer)
+					replayValues = replayBuffer.values
+					replayBuffer.values = []
+					if replayValues.isEmpty {
+						if let replayToken = replayToken {
+							state.replayBuffers?.removeValueForToken(replayToken)
+						}
+						token = state.observers?.insert(observer)
+					}
 					return state
 				}
-
-				originalState.values.forEach(observer.sendNext)
-
-				if let terminationEvent = originalState.terminationEvent {
-					observer.action(terminationEvent)
-				}
-			}
-
-			// Prevent other threads from sending events while we're replaying,
-			// but don't deadlock if we're replaying in response to a buffer
-			// event observed elsewhere.
-			//
-			// In other words, this permits limited signal recursion for the
-			// specific case of replaying past events.
-			if dispatch_get_specific(&queue) != nil {
-				replay()
-			} else {
-				dispatch_sync(queue, replay)
+			}
+			
+			if let terminationEvent = next.terminationEvent {
+				observer.action(terminationEvent)
 			}
 
 			if let token = token {

	public static func buffer(capacity: Int) -> (SignalProducer, Signal<Value, Error>.Observer) {
		precondition(capacity >= 0, "Invalid capacity: \(capacity)")

		// This is effectively used as a synchronous mutex, but permitting
		// limited recursive locking (see below).
		//
		// The queue is a "variable" just so we can use its address as the key
		// and the value for dispatch_queue_set_specific().
		var queue = dispatch_queue_create("org.reactivecocoa.ReactiveCocoa.SignalProducer.buffer", DISPATCH_QUEUE_SERIAL)
		dispatch_queue_set_specific(queue, &queue, &queue, nil)

		// Used as an atomic variable so we can remove observers without needing
		// to run on the queue.
		let state: Atomic<BufferState<Value, Error>> = Atomic(BufferState())

	}
}

/// A uniquely identifying token for Observers that are replaying values in
/// BufferState.
private final class ReplayBuffer<Value> {
	private var values: [Value] = []
}


private struct BufferState<Value, Error: ErrorType> {
	/// All values in the buffer.
	var values: [Value] = []
