<commit_before>package rescala.reactivestreams


import org.reactivestreams.{Publisher, Subscriber, Subscription}
import rescala.core.{Base, Scheduler, Pulse, REName, ReSource, ReSourciV, ReevaluationResultWithValue, Struct, Turn, ValuePersistency}

import scala.util.{Failure, Success}


object REPublisher {

  def apply[T, S <: Struct](dependency: ReSourciV[Pulse[T], S])(implicit fac: Scheduler[S]): REPublisher[T, S] =
    new REPublisher[T, S](dependency, fac)


  class REPublisher[T, S <: Struct](dependency: ReSourciV[Pulse[T], S], fac: Scheduler[S]) extends Publisher[T] {

    override def subscribe(s: Subscriber[_ >: T]): Unit = {
      val sub = REPublisher.subscription(dependency, s, fac)
      s.onSubscribe(sub)
    }

  }

  class SubscriptionReactive[T, S <: Struct](
    bud: S#State[Pulse[T], S],
    dependency: ReSourciV[Pulse[T], S],
    subscriber: Subscriber[_ >: T],
    fac: Scheduler[S],
    name: REName
  ) extends Base[Pulse[T], S](bud, name) with Subscription {

    var requested: Long = 0
    var cancelled = false

    override protected[rescala] def reevaluate(ticket: Turn[S], before: Pulse[T], indeps: Set[ReSource[S]]): ReevaluationResultWithValue[Value, S] = {
      ticket.makeStaticReevaluationTicket().staticDependPulse(dependency).toOptionTry match {
        case None => ReevaluationResultWithValue.StaticPulse[T, S](Pulse.NoChange, indeps)
        case Some(tryValue) =>
          synchronized {
            while (requested <= 0 && !cancelled) wait(100)
            if (cancelled) ReevaluationResultWithValue.DynamicPulse[T, S](Pulse.NoChange, indepsAfter = Set.empty, indepsAdded = Set.empty, indepsRemoved = Set(dependency))
            else {
              requested -= 1
              tryValue match {
                case Success(v) =>
                  subscriber.onNext(v)
                  ReevaluationResultWithValue.StaticPulse[T, S](Pulse.NoChange, indeps)
                case Failure(t) =>
                  subscriber.onError(t)
                  cancelled = true
                  ReevaluationResultWithValue.DynamicPulse[T, S](Pulse.NoChange, indepsAfter = Set.empty, indepsAdded = Set.empty, indepsRemoved = Set(dependency))
              }
            }
          }
      }
    }
<commit_msg>REPublisher compiles<commit_after>package rescala.reactivestreams


import org.reactivestreams.{Publisher, Subscriber, Subscription}
import rescala.core.{Base, DynamicTicket, Pulse, REName, ReSourciV, ReevaluationResult, ReevaluationResultWithoutValue, Scheduler, Struct, ValuePersistency}

import scala.util.{Failure, Success}


object REPublisher {

  def apply[T, S <: Struct](dependency: ReSourciV[Pulse[T], S])(implicit fac: Scheduler[S]): REPublisher[T, S] =
    new REPublisher[T, S](dependency, fac)


  class REPublisher[T, S <: Struct](dependency: ReSourciV[Pulse[T], S], fac: Scheduler[S]) extends Publisher[T] {

    override def subscribe(s: Subscriber[_ >: T]): Unit = {
      val sub = REPublisher.subscription(dependency, s, fac)
      s.onSubscribe(sub)
    }

  }

  class SubscriptionReactive[T, S <: Struct](
    bud: S#State[Pulse[T], S],
    dependency: ReSourciV[Pulse[T], S],
    subscriber: Subscriber[_ >: T],
    fac: Scheduler[S],
    name: REName
  ) extends Base[Pulse[T], S](bud, name) with Subscription {

    var requested: Long = 0
    var cancelled = false

    override protected[rescala] def reevaluate(ticket: DynamicTicket[S], before: Pulse[T]): ReevaluationResult[Value, S] = {
      ticket.staticDependPulse(dependency).toOptionTry match {
        case None => ReevaluationResultWithoutValue[S](propagate = false)
        case Some(tryValue) =>
          synchronized {
            while (requested <= 0 && !cancelled) wait(100)
            if (cancelled) {
              ticket.enableDynamic = true
              ReevaluationResultWithoutValue[S](propagate = false)
            }
            else {
              requested -= 1
              tryValue match {
                case Success(v) =>
                  subscriber.onNext(v)
                  ReevaluationResultWithoutValue[S](propagate = false)
                case Failure(t) =>
                  subscriber.onError(t)
                  cancelled = true
                  ticket.enableDynamic = true
                  ReevaluationResultWithoutValue[S](propagate = false)
              }
            }
          }
      }
    }

{'fraction_non_alphanumeric': 0.1141750227894257, 'fraction_numerical': 0.0027347310847766638, 'mean_word_length': 2.7868852459016393, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 15, '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': '20922', 'n_tokens_mistral': 1421, 'n_tokens_neox': 1324, 'n_words': 285}