<commit_before>package scala.reactive
package bench



import org.scalameter.api._
import rx.lang.scala._



class RxReactBench extends PerformanceTest.OfflineReport {

  val sumSizes = Gen.range("size")(100000, 500000, 100000)
  val aggregateSizes = Gen.range("size")(10, 100, 10)
  val setSizes = Gen.range("size")(200, 4000, 500)

  performance of "sum" config(
    exec.minWarmupRuns -> 50,
    exec.maxWarmupRuns -> 100,
    exec.benchRuns -> 30,
    exec.independentSamples -> 1
  ) in {
    using(sumSizes) curve("Reactive Collections") in { sz =>
      val e = new Reactive.Emitter[Int]
      val sum = e.scanPast(0)(_ + _)

      var i = 0
      while (i < sz) {
        e += i
        i += 1
      }
      sum()
    }

    using(sumSizes) curve("Rx") in { sz =>
      val s = subjects.BehaviorSubject[Int](0)
      val sum = s.reduce(_ + _)

      var i = 0
      while (i < sz) {
        s.onNext(i)
        i += 1
      }
    }

  }

}

<commit_msg>Adding the optimized Rx version.<commit_after>package scala.reactive
package bench



import org.scalameter.api._
import rx.lang.scala._



class RxReactBench extends PerformanceTest.OfflineReport {

  val sumSizes = Gen.range("size")(100000, 500000, 100000)
  val stringArrays = for (sz <- sumSizes) yield {
    val ss = for (x <- 0 until sz) yield scala.util.Random.nextString(10)
    ss.toArray
  }

  performance of "sum" config(
    exec.minWarmupRuns -> 50,
    exec.maxWarmupRuns -> 100,
    exec.benchRuns -> 30,
    exec.independentSamples -> 1
  ) in {
    using(sumSizes) curve("Reactive Collections") in { sz =>
      val e = new Reactive.Emitter[Int]
      val sum = e.scanPast(0)(_ + _)

      var i = 0
      while (i < sz) {
        e += i
        i += 1
      }
      sum()
    }

    using(sumSizes) curve("Rx") in { sz =>
      val xs = subjects.BehaviorSubject[Int](0)
      val sum = xs.reduce(_ + _)

      var i = 0
      while (i < sz) {
        xs.onNext(i)
        i += 1
      }
    }

    using(sumSizes) curve("Rx-optimized") in { sz =>
      val xs = Subject[Int]()
      val sum = Observable[Int](subscriber => {
        var z = 0
        subscriber.add(xs.subscribe(
          x => { z += x },
          e => subscriber.onError(e),
          () => { subscriber.onNext(z); subscriber.onCompleted() }
        ))
      })

      var i = 0
      while (i < sz) {
        xs.onNext(i)
        i += 1
      }
    }

  }

}


{'fraction_non_alphanumeric': 0.13521360431356283, 'fraction_numerical': 0.03608461219411033, 'mean_word_length': 2.00373599003736, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 20, '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': '6198581', 'n_tokens_mistral': 939, 'n_tokens_neox': 827, 'n_words': 204}