<commit_before>package fpscala.chapter7


import java.util.concurrent._

import scala.util._


/**
 * Created by sajit on 12/2/15.
 */

object Par {

  type Par[A] = ExecutorService => Future[A]

  def run[A](s: ExecutorService)(a: Par[A]): Future[A] = a(s)

  def map3[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] = es => {
    val af: Future[A] = a(es)
    val bf: Future[B] = b(es)
    val timeout: Long = 1l
    val aValMb = Try(af.get(timeout, TimeUnit.SECONDS))
    val bValMb = Try(bf.get(timeout, TimeUnit.SECONDS))
    (aValMb, bValMb) match {
      case (Success(aVal), Success(bVal)) => UnitFuture(f(aVal, bVal))
      case _ => throw new Exception
    }
  }

  def map[A, B](pa: Par[A])(f: A => B): Par[B] = map2(pa, unit(()))((a, _) => f(a))

  def map2[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] = es => {
    val af = a(es)
    val bf = b(es)
    UnitFuture(f(af.get(), bf.get()))
  }

  def unit[A](a: A): Par[A] = (es: ExecutorService) => UnitFuture(a)

  def asyncF[A, B](f: A => B): A => Par[B] = { a => lazyUnit(f(a)) }

  def lazyUnit[A](a: => A): Par[A] = fork(unit(a))

  def fork[A](a: => Par[A]): Par[A] = es => es.submit(new Callable[A] {
    override def call(): A = a(es).get
  })


  private case class UnitFuture[A](get: A) extends Future[A] {
    def isDone = true

    def isCancelled = false

    def get(timeout: Long, units: TimeUnit) = get

    def cancel(eventIfRunning: Boolean) = false

  }

}<commit_msg>Ex 7.5<commit_after>package fpscala.chapter7


import java.util.concurrent._

import scala.util._


/**
 * Created by sajit on 12/2/15.
 */

object Par {

  type Par[A] = ExecutorService => Future[A]

  def run[A](s: ExecutorService)(a: Par[A]): Future[A] = a(s)

  def map3[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] = es => {
    val af: Future[A] = a(es)
    val bf: Future[B] = b(es)
    val timeout: Long = 1l
    val aValMb = Try(af.get(timeout, TimeUnit.SECONDS))
    val bValMb = Try(bf.get(timeout, TimeUnit.SECONDS))
    (aValMb, bValMb) match {
      case (Success(aVal), Success(bVal)) => UnitFuture(f(aVal, bVal))
      case _ => throw new Exception
    }
  }

  def map[A, B](pa: Par[A])(f: A => B): Par[B] = map2(pa, unit(()))((a, _) => f(a))

  def map2[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] = es => {
    val af = a(es)
    val bf = b(es)
    UnitFuture(f(af.get(), bf.get()))
  }

  def unit[A](a: A): Par[A] = (es: ExecutorService) => UnitFuture(a)

  def asyncF[A, B](f: A => B): A => Par[B] = { a => lazyUnit(f(a)) }

  def lazyUnit[A](a: => A): Par[A] = fork(unit(a))

  def fork[A](a: => Par[A]): Par[A] = es => es.submit(new Callable[A] {
    override def call(): A = a(es).get
  })

  /**
   * folds over a list of Pars. and does something.. meh.. compiles atleast.
   * @param ps
   * @tparam A
   * @return
   */

  def sequence[A](ps: List[Par[A]]): Par[List[A]] = ps.foldRight[Par[List[A]]](unit(List()))((aParElement, parOfAList) =>
    map2(aParElement, parOfAList)((element, acc) => element :: acc))



  private case class UnitFuture[A](get: A) extends Future[A] {
    def isDone = true

    def isCancelled = false

    def get(timeout: Long, units: TimeUnit) = get

    def cancel(eventIfRunning: Boolean) = false

  }

}
{'fraction_non_alphanumeric': 0.2079360196862504, 'fraction_numerical': 0.0070747462319286376, 'mean_word_length': 3.1799485861182517, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2402658', 'n_tokens_mistral': 1407, 'n_tokens_neox': 1344, 'n_words': 335}