Concurrency in scala

distributed_http.sacla

import scala.actors.Actor._
import java.util.concurrent._
import scala.collection.mutable.ListBuffer

val terms = (1900 to 2012) map (i => "best party in "+i)

val pool = Executors.newFixedThreadPool(10)

var result = new ListBuffer[String]()

val sink = actor {
  loop {
    react {
      case s:String => {
        println("received "+s)
        result += s
      }
    } 
  }  
}

terms foreach (s => {
  val r = new Runnable {
    def run {
      Thread.sleep(500)
      sink ! s
    }
  }
  pool.execute(r)
})
pool.shutdown()

while(result.length < terms.length) Thread.sleep(50)

println("finished")
println(result)

distributed_http_with_stop_condition_in_actor.scala

import scala.actors.Actor._
import java.util.concurrent._
import scala.collection.mutable.ListBuffer
import java.net.{URLConnection, URL}
import scala.io.Source
import scala.collection.mutable

val terms = (1950 to 2012) map (i => "best party in "+i)

val pool = Executors.newFixedThreadPool(10)

var result = mutable.Map.empty[String, String]

val sink = actor {
  loop {
    if(result.keys.size < terms.length) {
      react {
        case s:(String,String) => {
         result += s
        }
      }
    } else {
      println("finished")
      result foreach (p => {
        println(p._1+": "+p._2.length)
      })
      pool.shutdown()
      exit()
    } 
  }  
}

terms foreach (s => {
  val r = new Runnable {
    def run {
      val u = "https://www.google.com/#q="+s
      println("getting "+u)
      val conn = new URL(u).openConnection()
    
      conn.connect

      val out = Source.fromInputStream(conn.getInputStream).mkString
      sink ! s ->out
    }
  }
  pool.execute(r)
})

without_exposing_threads.scala

import scala.actors.Actor._
import java.util.concurrent._
import scala.collection.mutable.ListBuffer
import java.net.{URLConnection, URL}
import scala.io.Source
import scala.collection.mutable

val terms = (1950 to 2012) map (i => "best party in "+i)

var result = mutable.Map.empty[String, String]

val sink = actor {
  loop {
    if(result.keys.size < terms.length) {
      react {
        case s:(String,String) => {
         result += s
        }
      }
    } else {
      println(Thread.currentThread.getName+"\tfinished")
      result foreach (p => {
        println(Thread.currentThread.getName+"\t"+p._1+": "+p._2.length)
      })
      exit()
    } 
  }  
}

terms foreach (s => {
  val a = actor {
    val u = "https://www.google.com/#q="+s
    println(Thread.currentThread.getName+"\tgetting "+u)
    val conn = new URL(u).openConnection()
    conn.connect
    val out = Source.fromInputStream(conn.getInputStream).mkString
    sink ! s ->out
  }
})