Jason Klassen Graduation Code - Specs thread-safe multi-threaded tests with partially applied functions

gistfile1.txt

/* User: Jason Klassen
*/

package extensions {

import actors.threadpool.LinkedBlockingQueue

object LinkedBlockingQueueExtensions {
  implicit def linkedBlockingQueueExt[A](source: LinkedBlockingQueue[A]) = new LinkedBlockingQueueExt[A](source)

  class LinkedBlockingQueueExt[A](source: LinkedBlockingQueue[A]) {

    def containsValue(query: A): Option[A] = {
      source.contains(query) match {
        case true => Some(query)
        case _ => None
      }
    }
  }

}

object OptionExtensions {
  implicit def optionExt[A](source: Option[A]) = new OptionExt(source)

  class OptionExt[A](source: Option[A]) {

    def value = {
      source.getOrElse("")
    }
  }

}

}

package test {

import org.specs._
import org.specs.util._
import scala.concurrent.ops._

import actors.threadpool.LinkedBlockingQueue
import extensions.LinkedBlockingQueueExtensions._
import extensions.OptionExtensions._

case class KeyValue(key: Int, value: String)


object GraduationSpec extends Specification {

  def formattedValue(key: Int) = {
    "node " + key
  }

  //Simple
  "Many concurrent threads" should {
    "simply build a key value reliably" in {
      var q = new LinkedBlockingQueue[KeyValue](10000)
      for (key <- 1 to 10000) {
        spawn {
          val value = formattedValue(key)
          println(value)
          //uncomment to fail with an instructive error message
          //if (i != 1000)
          q.add(new KeyValue(key, value))
        }
      }

      for (key <- 1 to 10000) {
        val value = formattedValue(key)
        val timeout = new Duration(500)

        val target = new KeyValue(key, value)
        q.containsValue(target).value must be(target).eventually(10, timeout)
      }
    }
  }

  //Complex
  def loopy(max: Int, handler: (KeyValue) => Unit) = {
    for (key <- 1 to max) {
      val value = formattedValue(key)
      handler(new KeyValue(key, value))
    }
  }

  def handleCreation(q: LinkedBlockingQueue[KeyValue], keyValue: KeyValue): Unit = {
    spawn {
      println(keyValue.value)
      //uncomment to fail with an instructive error message
      //if (keyValue.key != 1000)
      q.add(keyValue)
    }
  }

  def handleAssert(q: LinkedBlockingQueue[KeyValue], target: KeyValue): Unit = {
    val timeout = new Duration(500)

    q.containsValue(target).value must be(target).eventually(10, timeout)
  }

  "Many concurrent threads" should {
    "complexly build a key value reliably" in {
      var q = new LinkedBlockingQueue[KeyValue](10000)
      val max = 10000
      loopy(max, handleCreation(q, _: KeyValue))
      loopy(max, handleAssert(q, _: KeyValue))
    }
  }

}

}