<commit_before>package io.reactors
package protocol



import org.scalatest._
import scala.collection._
import scala.concurrent.Await
import scala.concurrent.Promise
import scala.concurrent.duration._



class BackpressureProtocolsSpec extends FunSuite {
  val system = ReactorSystem.default("backpressure-protocols")

  test("open a per client backpressure channel and send events") {
    val done = Promise[Seq[String]]()
    val server = system.backpressurePerClient[String](3) { events =>
      val seen = mutable.Buffer[String]()
      events onEvent { s =>
        seen += s
        if (seen.length == 5) {
          done.success(seen)
          Reactor.self.main.seal()
        }
      }
    }
    system.spawnLocal[Long] { self =>
      server.link onEvent { link =>
        def traverse(i: Int) {
          if (i == 5) {
            self.main.seal()
          } else {
            if (link.trySend(i.toString)) traverse(i + 1)
            else link.available.once.on(traverse(i))
          }
        }
        traverse(0)
      }
    }
    assert(Await.result(done.future, 10.seconds) == Seq("0", "1", "2", "3", "4"))
  }

  test("open backpressure for all channel and send events") {
    val done = Promise[Seq[String]]()
    val server = system.backpressureForAll[String](3) { events =>
      val seen = mutable.Buffer[String]()
      events onEvent { s =>
        seen += s
        if (seen.length == 15) {
          done.success(seen)
          Reactor.self.main.seal()
        }
      }
    }
    for (j <- 0 until 3) system.spawnLocal[Long] { self =>
      server.link onEvent { link =>
        def traverse(i: Int) {
          if (i == 5) {
            self.main.seal()
          } else {
            val x = j * 5 + i
            if (link.trySend(x.toString)) traverse(i + 1)
            else link.available.once.on(traverse(i))
          }
        }
        traverse(0)
      }
    }
    val result = Await.result(done.future, 10.seconds).toSet
    assert(result == (0 until 15).map(_.toString).toSet)
  }
}
<commit_msg>Add check for the "for-all" backpressure.<commit_after>package io.reactors
package protocol



import io.reactors.test._
import org.scalacheck._
import org.scalacheck.Prop.forAllNoShrink
import org.scalacheck.Gen.choose
import org.scalatest._
import scala.collection._
import scala.concurrent.Await
import scala.concurrent.Promise
import scala.concurrent.duration._



class BackpressureProtocolsSpec extends FunSuite {
  val system = ReactorSystem.default("backpressure-protocols")

  test("open a per client backpressure channel and send events") {
    val done = Promise[Seq[String]]()
    val server = system.backpressurePerClient[String](3) { events =>
      val seen = mutable.Buffer[String]()
      events onEvent { s =>
        seen += s
        if (seen.length == 5) {
          done.success(seen)
          Reactor.self.main.seal()
        }
      }
    }
    system.spawnLocal[Long] { self =>
      server.link onEvent { link =>
        def traverse(i: Int) {
          if (i == 5) {
            self.main.seal()
          } else {
            if (link.trySend(i.toString)) traverse(i + 1)
            else link.available.once.on(traverse(i))
          }
        }
        traverse(0)
      }
    }
    assert(Await.result(done.future, 10.seconds) == Seq("0", "1", "2", "3", "4"))
  }

  test("open backpressure for all channel and send events") {
    val done = Promise[Seq[String]]()
    val server = system.backpressureForAll[String](3) { events =>
      val seen = mutable.Buffer[String]()
      events onEvent { s =>
        seen += s
        if (seen.length == 15) {
          done.success(seen)
          Reactor.self.main.seal()
        }
      }
    }
    for (j <- 0 until 3) system.spawnLocal[Long] { self =>
      server.link onEvent { link =>
        def traverse(i: Int) {
          if (i == 5) {
            self.main.seal()
          } else {
            val x = j * 5 + i
            if (link.trySend(x.toString)) traverse(i + 1)
            else link.available.once.on(traverse(i))
          }
        }
        traverse(0)
      }
    }
    val result = Await.result(done.future, 10.seconds).toSet
    assert(result == (0 until 15).map(_.toString).toSet)
  }
}


class BackpressureProtocolsCheck
extends Properties("BackpressureProtocolsCheck") with ExtendedProperties {
  val system = ReactorSystem.default("check-system")

  val sizes = detChoose(0, 50)

  property("backpressure for all") = forAllNoShrink(sizes) {
    num =>
    stackTraced {
      val done = Promise[Seq[String]]()
      val server = system.backpressureForAll[String](3) { events =>
        val seen = mutable.Buffer[String]()
        events onEvent { s =>
          seen += s
          if (seen.length == num * 5) {
            done.success(seen)
            Reactor.self.main.seal()
          }
        }
      }
      for (j <- 0 until num) system.spawnLocal[Long] { self =>
        server.link onEvent { link =>
          def traverse(i: Int) {
            if (i == 5) {
              self.main.seal()
            } else {
              val x = j * 5 + i
              if (link.trySend(x.toString)) traverse(i + 1)
              else link.available.once.on(traverse(i))
            }
          }
          traverse(0)
        }
      }
      val result = Await.result(done.future, 10.seconds).toSet
      result == (0 until (num * 5)).map(_.toString).toSet
    }
  }
}

{'fraction_non_alphanumeric': 0.1300572695363015, 'fraction_numerical': 0.012192869019028265, 'mean_word_length': 1.9829201101928375, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 24, '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': '2658360', 'n_tokens_mistral': 1766, 'n_tokens_neox': 1680, 'n_words': 378}