<commit_before>package io.reactors
package protocol



import scala.collection._
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration._
import org.scalatest._
import org.scalatest.concurrent.AsyncTimeLimitedTests
import scala.concurrent.Promise



class StreamingLibraryTest extends AsyncFunSuite with AsyncTimeLimitedTests {
  val system = ReactorSystem.default("streaming-lib")

  def timeLimit = 10.seconds

  implicit override def executionContext = ExecutionContext.Implicits.global

  import StreamingLibraryTest._

  test("streaming map") {
    val total = 4096
    val done = Promise[Seq[Int]]()

    system.spawnLocal[Unit] { self =>
      val seen = mutable.Buffer[Int]()
      val source = new Source[Int](system)
      val ready = source.map(_ * 2).foreach { x =>
        seen += x
        if (seen.size == total) done.success(seen)
      }

      (ready ? ()) on {
        var i = 0
        source.valve.available.is(true) on {
          while (source.valve.available() && i < total) {
            source.valve.channel ! i
            i += 1
          }
        }
      }
    }

    done.future.map(t => assert(t == (0 until total)))
  }
}


object StreamingLibraryTest {
  type StreamReq[T] = Channel[Reliable.TwoWay.Req[Int, T]]

  type StreamServer[T] = Channel[StreamReq[T]]

  type StreamMedium[T] = Backpressure.Medium[Reliable.TwoWay.Req[Int, T], T]

  trait Stream[T] {
    def system: ReactorSystem

    def streamServer: StreamServer[T]

    def backpressureMedium[R: Arrayable]: StreamMedium[R]

    def backpressurePolicy: Backpressure.Policy

    def map[S](f: T => S)(implicit at: Arrayable[T], as: Arrayable[S]): Stream[S] =
      new Mapped(this, f)
<commit_msg>Make streaming lib work correctly.<commit_after>package io.reactors
package protocol



import scala.collection._
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration._
import org.scalatest._
import org.scalatest.concurrent.AsyncTimeLimitedTests
import scala.concurrent.Promise



class StreamingLibraryTest extends AsyncFunSuite with AsyncTimeLimitedTests {
  val system = ReactorSystem.default("streaming-lib")

  def timeLimit = 10.seconds

  implicit override def executionContext = ExecutionContext.Implicits.global

  import StreamingLibraryTest._

  test("streaming map") {
    val total = 4096
    val done = Promise[Seq[Int]]()

    system.spawnLocal[Unit] { self =>
      val seen = mutable.Buffer[Int]()
      val source = new Source[Int](system)
      val ready = source.map(_ * 2).foreach { x =>
        seen += x
        if (seen.size == total) done.success(seen)
      }

      (ready ? ()) on {
        var i = 0
        source.valve.available.is(true) on {
          while (source.valve.available() && i < total) {
            source.valve.channel ! i
            i += 1
          }
        }
      }
    }

    done.future.map(t => assert(t == (0 until total).map(_ * 2)))
  }
}


object StreamingLibraryTest {
  type StreamReq[T] = Channel[Reliable.TwoWay.Req[Int, T]]

  type StreamServer[T] = Channel[StreamReq[T]]

  type StreamMedium[T] = Backpressure.Medium[Reliable.TwoWay.Req[Int, T], T]

  trait Stream[T] {
    def system: ReactorSystem

    def streamServer: StreamServer[T]

    def backpressureMedium[R: Arrayable]: StreamMedium[R]

    def backpressurePolicy: Backpressure.Policy

    def map[S](f: T => S)(implicit at: Arrayable[T], as: Arrayable[S]): Stream[S] =
      new Mapped(this, f)

{'fraction_non_alphanumeric': 0.11398521887436043, 'fraction_numerical': 0.005969300739056282, 'mean_word_length': 3.135135135135135, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, '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': '4049058', 'n_tokens_mistral': 1141, 'n_tokens_neox': 1071, 'n_words': 257}