 	func buffer(capacity: Int? = nil) -> (Producer<T>, Disposable) {
 		let buffer = EventBuffer<T>(capacity: capacity)
 
+		// TODO: How does `self` get retained properly?
 		let observationDisposable = self.observe { value in
 			buffer.put(.Next(Box(value)))
 		}

			box.value.put(self._current!)
		}

		return ActionDisposable {
			dispatch_barrier_async(self._queue) {
				self._observers = removeObjectIdenticalTo(box, fromArray: self._observers)
			}
		}
	}

	func combineLatestWith<U>(stream: Signal<U>) -> Signal<(T, U)> {
		return Signal<(T, U)>(initialValue: (self.current, stream.current)) { sink in
			// FIXME: This implementation is probably racey.
			self.observe { [unowned stream] value in sink.put(value, stream.current) }
			stream.observe { [unowned self] value in sink.put(self.current, value) }
		}
	}

