avgkvsz.scala

import java.util.ArrayList
import scala.collection.JavaConversions._
import com.stumbleupon.async.{Callback, Deferred}
import org.hbase.async.{HBaseClient, KeyValue, Scanner}
import com.twitter.util.{Future, Promise, Return, Throw}

/** Converts a Deferred into a Future.  */                       
implicit def futureFromDeferred[A](d: Deferred[A]): Future[A] = {
  val promise = new Promise[A]                                   
                                                                 
  d.addBoth(new Callback[Unit, A] {                              
    def call(arg: A) = promise() = arg match {                   
      case e: Throwable => Throw(e)                              
      case _ => Return(arg)                                      
    }                                                            
  })                                                             
                                                                 
  promise                                                        
}                                                                

val client = new HBaseClient("localhost")
val scanner = client.newScanner("tsdb")
var numkvs = 0
var totalsize = 0
def cb(rows: ArrayList[ArrayList[KeyValue]]): Future[(Int, Int)] = {
  if (rows != null) {
    numkvs += rows.map(_.size).sum
    totalsize += rows.flatten.map(_.value.length).sum
    scanner.nextRows andThen cb
  } else Future((numkvs, totalsize))
}

println("I am starting in thread " + Thread.currentThread.getName)
scanner.nextRows andThen cb map { result =>
  println("I have results in thread " + Thread.currentThread.getName)
  println("Got " + result._1 + " KVs totaling " + result._2 + " bytes")
  client.shutdown.get()  // Don't forget this or your JVM won't exit because of unterminated threads
}
println("I am done in thread " + Thread.currentThread.getName)