max.scala

import com.twitter.util.{Var, Witness}
import org.mockito.Matchers.anyInt
import org.mockito.Mockito.{times, verify}
import org.scalatest.mockito.MockitoSugar
import org.scalatest.prop.PropertyChecks
import org.scalatest.{MustMatchers, WordSpec}

class VarsTest extends WordSpec with MustMatchers with PropertyChecks with MockitoSugar {

  def max[T](values: Var[T])(implicit ord: Ordering[T]): Var[T] = {
    val initial = values.sample()
    Var.async[T](initial) { u =>
      var max = initial
      values.changes.respond { v =>
        if (ord.gt(v, max)) {
          max = v
          u.update(max)
        }
      }
    }
  }

  def max2[T](values: Var[T])(implicit ord: Ordering[T]): Var[T] = {
    val initial = values.sample()
    val maxEvents = values.changes.foldLeft(initial) { (x, y) =>
      if (ord.gt(x, y)) x else y
    }.dedup
    Var(initial, maxEvents)
  }

  "always return max value" in {

    val value = Var(10)

    val maxValue = max2(value)

    val mockWitness = mock[Witness[Int]]
    maxValue.changes.register(mockWitness)

    value.update(1)
    maxValue.sample() must be(10)
    value.update(2)
    maxValue.sample() must be(10)
    value.update(3)
    maxValue.sample() must be(10)

    value.update(11)
    maxValue.sample() must be(11)
    value.update(12)
    maxValue.sample() must be(12)
    value.update(13)
    maxValue.sample() must be(13)

    verify(mockWitness, times(4)).notify(anyInt())

  }

}

  /**
   * Sample the current value of this Var. Note that this may lead to
   * surprising results for lazily defined Vars: the act of observing
   * a Var may be kick off a process to populate it; the value
   * returned from sample may then reflect an intermediate value.
   */
  def sample[T](v: Var[T]): T = {
    var opt: Option[T] = None
    v.observe(0, Observer(v => opt = Some(v))).close()
    opt.get
  }