Testing non blocking futures

FutureDemo.scala

package myscala.others

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.{Failure, Random, Success}

object FutureDemo {

  def sleep(duration: Long) {
    Thread.sleep(duration)
  }

  def nonBlocking() {
    var r1 = Random.nextInt(1000) + 500
    val intFuture: Future[Float] = Future {
      Thread.sleep(r1) // dorme entre 0.5 e 1.5 segundos
      (r1.toInt / 1000.0).asInstanceOf[Float]
    }

    // use a "callback" which is non blocking
    intFuture onComplete {
      case Success(t) => {
        println(t)
      }
      case Failure(e) => {
        println(s"An error has occured: $e.getMessage")
      }
    }
    // Isto abaixo aguarda um segundo e depois sai.
    // Se o Future termina antes ele é processado senão já era !
    sleep(1000)
    println("•• r1 = " + r1)
  }

  import java.time.{ZoneId, ZonedDateTime}

  def utcDateTime(): ZonedDateTime = {
    // DateTime.now(DateTimeZone.UTC).getMillis()
    val utcZoneId = ZoneId.of("UTC")
    val zonedDateTime = ZonedDateTime.now
    val utcDateTime = zonedDateTime.withZoneSameInstant(utcZoneId)

    utcDateTime
  }

  def crossProduct() {
    println(utcDateTime())
    var finalTime: ZonedDateTime = null;
    var r1 = Random.nextInt(70) + 1400
    var r2 = Random.nextInt(70) + 800
    val initTimeMilis = System.currentTimeMillis()
    var finalTimeMilis = System.currentTimeMillis()
    val intFuture1: Future[Float] = Future {
      Thread.sleep(r1) // dorme entre 1 e 2 segundos
      r1.toInt
    }
    val intFuture2: Future[Float] = Future {
      Thread.sleep(r2) // dorme entre 1 e 2 segundos
      r2.toInt
    }
    // use "for comprehension" which is non blocking
    // import scala.concurrent.ExecutionContext.Implicits.global

    val (f1, f2) = (intFuture1, intFuture2)
    val f = for {
      r1 <- f1
      r2 <- f2
    } yield " r1 = " + r1 + ", and r2 = " + r2
    f onComplete {
      case Success(s) => {
        finalTime = utcDateTime()
        finalTimeMilis = System.currentTimeMillis()
        println(s"result: $s")
      }
      case Failure(e) => {
        throw new RuntimeException("Falha no crossProduct")
      } // Handle failure
    }

    // Isto abaixo aguarda um 1.5 segundos e depois sai.
    // Se os Futures terminam antes eles são processados senão já era !
    // Se pelo menos um demorar mais que 1.5 segundos toda a operação
    // no "for comprehension" é abandonada
    // Overhead dos Futures fica em torno de 200 milisegundos
    sleep(1500)
    println("•• " + finalTime + ", " +
      (finalTimeMilis - initTimeMilis) +
      " -> r1 = " + r1 + ", r2 = " + r2)
  }

  def main(args: Array[String]) = {

    // nonBlocking()

    crossProduct()
  }
}