def each(timeout: nil)
      @deadline = Time.now + timeout if timeout

      while true
        # prevent any intervening messages from being processed and discarded before we're back in Celluloid.receive()
        Celluloid.exclusive {
          if error?
            debug { "raise: #{self.describe_observables}" }

            raise self.error
          elsif ready?
            yield *self.values

            @deadline = Time.now + timeout if timeout
          end
        }

        # must be atomic!
        debug { "wait: #{self.describe_observables}" }

        update(receive())
      end
    end
def each(timeout: nil)
      @deadline = Time.now + timeout if timeout

      while true
        # prevent any intervening messages from being processed and discarded before we're back in Celluloid.receive()
        Celluloid.exclusive {
          if error?
            debug { "raise: #{self.describe_observables}" }

            raise self.error
          elsif ready?
            yield *self.values

            @deadline = Time.now + timeout if timeout
          end
        }

        # must be atomic!
        debug { "wait: #{self.describe_observables}" }

        update(receive())
      end
    end
Yield each set of ready? observed values while alive, or raise on error?

 The yield is exclusive, because suspending the observing task would mean that
 any observable messages would get discarded.

 @param timeout [Float] timeout between each yield