Utility Classes

case class Duration(private val millis: Long) {
  def +(that: Duration): Duration
  def -(that: Duration): Duration  
  def *(times: Int): Duration
  def /(that: Duration): Double
  def <(that: Duration): Boolean
  def >(that: Duration): Boolean
  def milliseconds: Long
  def prettyPrint: String
}

case class Time(private val millis: Long) {
  def +(duration: Duration): Time
  def -(duration: Duration): Time
  def -(that: Time): Duration
  def <(that: Time): Boolean
  def >(that: Time): Boolean
  def milliseconds: Long
}

// Duration helpers
object Milliseconds { def apply(milliseconds: Long): Duration }
object Seconds { def apply(seconds: Long): Duration }
object Minutes { def apply(minutes: Long): Duration }

abstract class State[S] {
  def exists(): Boolean
  def get(): S
  def update(newState: S): Unit
  def remove(): Unit
  def isTimingOut(): Boolean
}

class StateSpec[K, V, S, T] {
  def function(mappingFunction: (K, Option[V], State[S]) => Option[T]): StateSpec[K, V, S, T]
  def initialState(rdd: RDD[(K, S)]): StateSpec[K, V, S, T]
  def numPartitions(numPartitions: Int): StateSpec[K, V, S, T]
  def timeout(duration: Duration): StateSpec[K, V, S, T]
}

val spec = StateSpec.function(mappingFunction)
  .initialState(initialRDD)
  .numPartitions(10)
  .timeout(Seconds(30))