<commit_before>package scalaz.stream.actor

import scala._
import scala.annotation.tailrec
import scalaz._
import scalaz.concurrent.{Strategy, Actor, Task}
import scalaz.stream.Process._
import scalaz.stream.Step
import scalaz.stream.wye.{AwaitBoth, AwaitR, AwaitL}
import scalaz.stream.{Process, wye}
import java.util.concurrent.atomic.AtomicBoolean

object debug {
  def apply(s:String,o:AnyRef*) = {
   //println(s,o.mkString(","))
  }
}

object WyeActor {

  trait WyeSide[A, L, R, O] {
    /** returns next wye after processing the result of the step **/
    def receive(r: \/[Throwable, Step[Task, A]])(y2: Wye[L, R, O]): Wye[L, R, O]
  }

 val Interrupted = new java.lang.Exception("Interrupted to clean") {
    override def fillInStackTrace(): Throwable = this
  }

  sealed trait Msg
  case class Ready[A, L, R, O](from: WyeSide[A, L, R, O], s: Throwable \/ Step[Task, A]) extends Msg
  case class Get[A](cb: (Throwable \/ Seq[A]) => Unit) extends Msg
  case class Done(rsn: Throwable, cb: (Throwable \/ Unit) => Unit) extends Msg

  trait WyeSideOps[A, L, R, O] extends WyeSide[A, L, R, O] {

    // Next step of process that feds into wye.
    // - left contains cleanup in case the step is running to perform any cleanup needed
    // - right contains next step of the process
    var step: (Process[Task,A] \/ Step[Task, A])

    // when this is set to`true`
    // it indicates the running task to be interrupted and cleanup process to start
    private val cleanup: AtomicBoolean = new AtomicBoolean(false)

    def feedA(as: Seq[A])(y2: Wye[L, R, O]): Wye[L, R, O]
    def haltA(e: Throwable)(y2: Wye[L, R, O]): Wye[L, R, O]

    //feeds the wye by element or signals halt to wye, producing next state of process
    def receive(r: Throwable \/ Step[Task, A])(y2: Wye[L, R, O]): Wye[L, R, O] = {
      r match {
        case \/-(s) =>
          step = \/-(s)
          s match {
            case Step(\/-(h),Halt(e),c) =>
              debug("HLTS", this, h, e, c,  y2)
              haltA(e)(feedA(h)(y2))

            case Step(\/-(h),t,c) =>
              debug("RDYS", this, h, t, s, y2)
              val fed = feedA(h)(y2)
              debug("RDYS-FED", fed)
              fed

            case Step(-\/(e),t,c) =>
              debug("HLTE", this, e, t,c, y2)
              haltA(e)(y2)
          }

        case -\/(e)    =>
          step = \/-(Step.failed(e))
          haltA(e)(y2)
      }
    }

    def isClean: Boolean = step.toOption.exists { case s if s.isCleaned => true ; case _ => false }
    def isHalt: Boolean = step.toOption.exists { case s if s.isHalted => true ; case _ => false }
    def haltedBy: Option[Throwable] = step.toOption.collect {
      case Step(_,Halt(e),Halt(_)) => e
    }

    //returns true when the process is cleaned, or runs the cleanup and returns false
    //if process is running is no-op and returns false
    def runCleanup(a: Actor[Msg], e: Throwable): Boolean = {
      debug("CLNP", this, e, step)
      step match {
        case \/-(s) if s.isCleaned => true

        case \/-(s) =>
          debug("CLNP!!!", this, s.tail , "|*|", s.cleanup)
          runClean(s.cleanup,e,a)
          false

        case -\/(c) if cleanup.get == false =>
          cleanup.set(true) //interrupt
          debug("CLNP!!!", this, cleanup.get.toString,  c)
          a ! Ready(this,\/-(Step.failed(Interrupted))) //this will have to be removed once Task will return error once interrupted in scalaz.task
          false

        case -\/(c) => false

      }
    }

    def pull(a: Actor[Msg]): Boolean = {
      debug("PULLED",this, "|*|", step)
      step match {
        case \/-(s) if s.isCleaned => false
        case \/-(s) if s.isHalted => runClean(s.cleanup,End,a) ; true
        case \/-(s) => run(s.tail,a) ; true
        case -\/(c) => false // in process
      }

    }

    private def runClean(c:Process[Task,A], e: Throwable, actor: Actor[Msg]) : Unit = {
      step = -\/(halt)
      debug("RUNC",this,c)
      c.causedBy(e).run.runAsync { cb => actor ! Ready(this, cb.map(_ => Step.failed(e)))}
    }


    private def run(s: Process[Task, A], actor: Actor[Msg]): Unit = {
      step = -\/(s.cleanup)
      debug("RUN",this,s)
      s.runStep.runAsync { cb => actor ! Ready(this, cb) }
    }


  }

  /**
   * Actor that backs the `wye`. Actor is reading non-deterministically from both sides
   * and interprets wye to produce output stream.
   *
   * @param pl left process
   * @param pr right process
   * @param y  wye to control queueing and merging
   * @param S  strategy, preferably executor service
   * @tparam L Type of left process element
   * @tparam R Type of right process elements
   * @tparam O Output type of resulting process
   * @return Process with merged elements.
   */
  def wyeActor[L, R, O](pl: Process[Task, L], pr: Process[Task, R])(y: Wye[L, R, O])(S: Strategy): Process[Task, O] = {

    //current state of the wye
    var yy: Wye[L, R, O] = y<commit_msg>Added runAsyncInterruptibly to wye step run<commit_after>package scalaz.stream.actor

import scala._
import scala.annotation.tailrec
import scalaz._
import scalaz.concurrent.{Strategy, Actor, Task}
import scalaz.stream.Process._
import scalaz.stream.Step
import scalaz.stream.wye.{AwaitBoth, AwaitR, AwaitL}
import scalaz.stream.{Process, wye}
import java.util.concurrent.atomic.AtomicBoolean

object debug {
  def apply(s:String,o:AnyRef*) = {
   //println(s,o.mkString(","))
  }
}

object WyeActor {

  trait WyeSide[A, L, R, O] {
    /** returns next wye after processing the result of the step **/
    def receive(r: \/[Throwable, Step[Task, A]])(y2: Wye[L, R, O]): Wye[L, R, O]
  }

 val Interrupted = new java.lang.Exception("Interrupted to clean") {
    override def fillInStackTrace(): Throwable = this
  }

  sealed trait Msg
  case class Ready[A, L, R, O](from: WyeSide[A, L, R, O], s: Throwable \/ Step[Task, A]) extends Msg
  case class Get[A](cb: (Throwable \/ Seq[A]) => Unit) extends Msg
  case class Done(rsn: Throwable, cb: (Throwable \/ Unit) => Unit) extends Msg

  trait WyeSideOps[A, L, R, O] extends WyeSide[A, L, R, O] {

    // Next step of process that feds into wye.
    // - left contains cleanup in case the step is running to perform any cleanup needed
    // - right contains next step of the process
    var step: (Process[Task,A] \/ Step[Task, A])

    // when this is set to`true`
    // it indicates the running task to be interrupted and cleanup process to start
    private val cleanup: AtomicBoolean = new AtomicBoolean(false)

    def feedA(as: Seq[A])(y2: Wye[L, R, O]): Wye[L, R, O]
    def haltA(e: Throwable)(y2: Wye[L, R, O]): Wye[L, R, O]

    //feeds the wye by element or signals halt to wye, producing next state of process
    def receive(r: Throwable \/ Step[Task, A])(y2: Wye[L, R, O]): Wye[L, R, O] = {
      r match {
        case \/-(s) =>
          step = \/-(s)
          s match {
            case Step(\/-(h),Halt(e),c) =>
              debug("HLTS", this, h, e, c,  y2)
              haltA(e)(feedA(h)(y2))

            case Step(\/-(h),t,c) =>
              debug("RDYS", this, h, t, s, y2)
              val fed = feedA(h)(y2)
              debug("RDYS-FED", fed)
              fed

            case Step(-\/(e),t,c) =>
              debug("HLTE", this, e, t,c, y2)
              haltA(e)(y2)
          }

        case -\/(e)    =>
          step = \/-(Step.failed(e))
          haltA(e)(y2)
      }
    }

    def isClean: Boolean = step.toOption.exists { case s if s.isCleaned => true ; case _ => false }
    def isHalt: Boolean = step.toOption.exists { case s if s.isHalted => true ; case _ => false }
    def haltedBy: Option[Throwable] = step.toOption.collect {
      case Step(_,Halt(e),Halt(_)) => e
    }

    //returns true when the process is cleaned, or runs the cleanup and returns false
    //if process is running is no-op and returns false
    def runCleanup(a: Actor[Msg], e: Throwable): Boolean = {
      debug("CLNP", this, e, step)
      step match {
        case \/-(s) if s.isCleaned => true

        case \/-(s) =>
          debug("CLNP!!!", this, s.tail , "|*|", s.cleanup)
          runClean(s.cleanup,e,a)
          false

        case -\/(c) if cleanup.get == false =>
          cleanup.set(true) //interrupt
          debug("CLNP!!!", this, cleanup.get.toString,  c)
          a ! Ready(this,\/-(Step.failed(Interrupted))) //this will have to be removed once Task will return error once interrupted in scalaz.task
          false

        case -\/(c) => false

      }
    }

    def pull(a: Actor[Msg]): Boolean = {
      debug("PULLED",this, "|*|", step)
      step match {
        case \/-(s) if s.isCleaned => false
        case \/-(s) if s.isHalted => runClean(s.cleanup,End,a) ; true
        case \/-(s) => run(s.tail,a) ; true
        case -\/(c) => false // in process
      }

    }

    private def runClean(c:Process[Task,A], e: Throwable, actor: Actor[Msg]) : Unit = {
      step = -\/(halt)
      debug("RUNC",this,c)
      c.causedBy(e).run.runAsync { cb => actor ! Ready(this, cb.map(_ => Step.failed(e)))}
    }


    private def run(s: Process[Task, A], actor: Actor[Msg]): Unit = {
      step = -\/(s.cleanup)
      debug("RUN",this,s)
      s.runStep.runAsyncInterruptibly ({ cb => actor ! Ready(this, cb) },cleanup)
    }


  }

  /**
   * Actor that backs the `wye`. Actor is reading non-deterministically from both sides
   * and interprets wye to produce output stream.
   *
   * @param pl left process
   * @param pr right process
   * @param y  wye to control queueing and merging
   * @param S  strategy, preferably executor service
   * @tparam L Type of left process element
   * @tparam R Type of right process elements
   * @tparam O Output type of resulting process
   * @return Process with merged elements.
   */
  def wyeActor[L, R, O](pl: Process[Task, L], pr: Process[Task, R])(y: Wye[L, R, O])(S: Strategy): Process[Task, O] = {

    //current state of the wye
    var yy: Wye[L, R, O] = y
{'fraction_non_alphanumeric': 0.15090745011531134, 'fraction_numerical': 0.0022059560814198334, 'mean_word_length': 2.3810169491525426, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 49, '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': '5210686', 'n_tokens_mistral': 3492, 'n_tokens_neox': 3387, 'n_words': 1055}