<commit_before>package scalaz.stream

import java.io.{InputStream,OutputStream}

import scalaz.concurrent.Task
import Process._

/** 
 * Module of `Process` functions and combinators for file and network I/O.
 */
trait io {

  /* 
   * Generic combinator for producing a `Process[Task,O]` from some
   * effectful `O` source. The source is tied to some resource,
   * `R` (like a file handle) that we want to ensure is released.
   * See `lines` below for an example use. 
   */
  def resource[R,O](acquire: Task[R])(
                    release: R => Task[Unit])(
                    step: R => Task[O]): Process[Task,O] = {
    def go(step: Task[O], onExit: Task[Unit]): Process[Task,O] =
      await[Task,O,O](step) ( 
        o => emit(o) ++ go(step, onExit) // Emit the value and repeat 
      , await[Task,Unit,O](onExit)()  // Release resource when exhausted
      , await[Task,Unit,O](onExit)()) // or in event of error
    await(acquire) ( r => go(step(r), release(r)), Halt, Halt )
  }

  /* 
   * Create a `Process[Task,String]` from the lines of a file, using
   * the `resource` combinator to ensure the file is closed
   * when processing the stream of lines is finished. 
   */
  def linesR(filename: String): Process[Task,String] = 
    resource(Task.delay(scala.io.Source.fromFile(filename)))(
             src => Task.delay(src.close)) { src => 
      lazy val lines = src.getLines // A stateful iterator 
      Task.delay { if (lines.hasNext) lines.next else throw End }
    }

  /** 
   * Create a `Process[Task,String]` from an `InputStream`, by 
   * repeatedly requesting chunks of size `n`. The last chunk may 
   * have size less than `n`. This implementation reuses the
   * same buffer for consecutive reads, which is only safe if 
   * whatever consumes this `Process` never stores the `Bytes` 
   * returned or passes it to a combinator (like `buffer`) that
   * does. Use `chunkR` for a safe version of this combinator.
   * 
   * This implementation closes the `InputStream` when finished
   * or in the event of an error.
   */
  def unsafeChunkR(n: Int)(is: => InputStream): Process[Task,Bytes] = {
    require(n > 0, "chunk size must be greater than 0, was: " + n)
    resource(Task.delay((is, new Array[Byte](n))))(
             src => Task.delay(src._1.close)) { case (src,buf) => 
      Task.delay { val m = src.read(buf); if (m == -1) throw End else new Bytes(buf, m) } 
    }
  }

  /** 
   * Create a `Process[Task,String]` from an `InputStream`, by 
   * repeatedly requesting chunks of size `n`. The last chunk may 
   * have size less than `n`.
   *
   * This implementation closes the `InputStream` when finished
   * or in the event of an error. 
   */
  def chunkR(n: Int)(is: => InputStream): Process[Task, Array[Byte]] =
    unsafeChunkR(n)(is).map(_.toArray)

  /** 
   * Create a `Sink` from an `OutputStream`, which will be closed<commit_msg>more flexible model for chunked reads - can change buffer size on each step if desired<commit_after>package scalaz.stream

import java.io.{InputStream,OutputStream}

import scalaz.concurrent.Task
import Process._

/** 
 * Module of `Process` functions and combinators for file and network I/O.
 */
trait io {

  /* 
   * Generic combinator for producing a `Process[Task,O]` from some
   * effectful `O` source. The source is tied to some resource,
   * `R` (like a file handle) that we want to ensure is released.
   * See `lines` below for an example use. 
   */
  def resource[R,O](acquire: Task[R])(
                    release: R => Task[Unit])(
                    step: R => Task[O]): Process[Task,O] = {
    def go(step: Task[O], onExit: Task[Unit]): Process[Task,O] =
      await[Task,O,O](step) ( 
        o => emit(o) ++ go(step, onExit) // Emit the value and repeat 
      , await[Task,Unit,O](onExit)()  // Release resource when exhausted
      , await[Task,Unit,O](onExit)()) // or in event of error
    await(acquire) ( r => go(step(r), release(r)), Halt, Halt )
  }

  /* 
   * Create a `Process[Task,String]` from the lines of a file, using
   * the `resource` combinator to ensure the file is closed
   * when processing the stream of lines is finished. 
   */
  def linesR(filename: String): Process[Task,String] = 
    resource(Task.delay(scala.io.Source.fromFile(filename)))(
             src => Task.delay(src.close)) { src => 
      lazy val lines = src.getLines // A stateful iterator 
      Task.delay { if (lines.hasNext) lines.next else throw End }
    }

  /** 
   * Create a `Channel[Task,Array[Byte],Bytes]` from an `InputStream` by 
   * repeatedly filling the input buffer. The last chunk may be less 
   * than the requested size.
   * 
   * Because this implementation returns a read-only view of the given
   * buffer, it is safe to recyle the same buffer for consecutive reads
   * as long as whatever consumes this `Process` never stores the `Bytes` 
   * returned or pipes it to a combinator (like `buffer`) that does. 
   * Use `chunkR` for a safe version of this combinator.
   * 
   * This implementation closes the `InputStream` when finished
   * or in the event of an error.
   */
  def unsafeChunkR(is: => InputStream): Channel[Task,Array[Byte],Bytes] = {
    resource(Task.delay(is))(
             src => Task.delay(src.close)) { src =>
      Task.now { (buf: Array[Byte]) => Task.delay {
        val m = src.read(buf)
        if (m == -1) throw End 
        else new Bytes(buf, m) 
      }} 
    }
  }

  /** 
   * Create a `Channel[Task,Int,Bytes]` from an `InputStream` by 
   * repeatedly requesting the given number of bytes. The last chunk 
   * may be less than the requested size.
   * 
   * This implementation requires an array allocation for each read.
   * To recycle the input buffer, use `unsafeChunkR`. 
   *
   * This implementation closes the `InputStream` when finished
   * or in the event of an error. 
   */
  def chunkR(is: => InputStream): Channel[Task, Int, Array[Byte]] =
    unsafeChunkR(is).map(f => (n: Int) => { 
      val buf = new Array[Byte](n)
      f(buf).map(_.toArray)
    })

  /** 
   * Create a `Sink` from an `OutputStream`, which will be closed
{'fraction_non_alphanumeric': 0.12477543687734771, 'fraction_numerical': 0.0008165931732810714, 'mean_word_length': 2.8564231738035266, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 27, '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': '5309283', 'n_tokens_mistral': 1919, 'n_tokens_neox': 1838, 'n_words': 671}