<commit_before>package io.github.binaryfoo.lagotto.reader

import java.io.{BufferedInputStream, File, FileInputStream}
import java.util.concurrent.ArrayBlockingQueue
import java.util.zip.GZIPInputStream

import io.github.binaryfoo.lagotto._

import scala.annotation.tailrec
import scala.collection.AbstractIterator
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent._
import scala.concurrent.duration._
import scala.io.{BufferedSource, Source}

case class LogReader[T <: LogEntry](strict: Boolean = false, keepFullText: Boolean = true, progressMeter: ProgressMeter = NullProgressMeter, logType: LogType[T] = JposLog) {

  def readFilesOrStdIn(args: Iterable[String]): Iterator[T] = {
    if (args.isEmpty)
      read(new BufferedSource(System.in))
    else
      read(args.map(new File(_)))
  }

  def read(files: File*): Iterator[T] = read(files.toIterable)

  def read(files: Iterable[File]): Iterator[T] = {
    progressMeter.startRun(files.size)
    files.toIterator.flatMap(f => read(open(f), f.getName))
  }

  private def open(f: File): BufferedSource = {
    if (f.getName.endsWith(".gz"))
      Source.fromInputStream(new GZIPInputStream(new BufferedInputStream(new FileInputStream(f))))
    else
      Source.fromFile(f)
  }

  def read(source: Source, sourceName: String = ""): Iterator[T] = new LogEntryIterator(source, sourceName, progressMeter)

  private val processors = Runtime.getRuntime.availableProcessors()

  class LogEntryIterator(source: Source, sourceName: String = "", progressMeter: ProgressMeter) extends AbstractIterator[T] {

    private val lines = new SourceLineIterator(source.getLines(), sourceName, strict, keepFullText)
    private var recordCount = 0
    private val queue = new ArrayBlockingQueue[Future[T]](processors * processors)
    private var current: T = null.asInstanceOf[T]
    private var started = false
    private val reader = new Thread(new Runnable {
      override def run(): Unit = {
        var more = true
        do {
          val f = try {
            val entry = logType.readLinesForNextRecord(lines)
            if (entry != null) {
              parseInTheFuture(entry)
            } else {
              more = false
              Future.successful(null.asInstanceOf[T])
            }
          }
          catch {
            case e: Exception => Future.failed(e)
          }
          queue.put(f)
        } while (more)
      }
    }, s"$sourceName-reader")
    reader.setDaemon(true)

    if (sourceName != "")
      progressMeter.startFile(sourceName)

    override def hasNext: Boolean = {
      ensureStarted()
      current != null
    }<commit_msg>Make the queue a little larger.<commit_after>package io.github.binaryfoo.lagotto.reader

import java.io.{BufferedInputStream, File, FileInputStream}
import java.util.concurrent.ArrayBlockingQueue
import java.util.zip.GZIPInputStream

import io.github.binaryfoo.lagotto._

import scala.annotation.tailrec
import scala.collection.AbstractIterator
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent._
import scala.concurrent.duration._
import scala.io.{BufferedSource, Source}

case class LogReader[T <: LogEntry](strict: Boolean = false, keepFullText: Boolean = true, progressMeter: ProgressMeter = NullProgressMeter, logType: LogType[T] = JposLog) {

  def readFilesOrStdIn(args: Iterable[String]): Iterator[T] = {
    if (args.isEmpty)
      read(new BufferedSource(System.in))
    else
      read(args.map(new File(_)))
  }

  def read(files: File*): Iterator[T] = read(files.toIterable)

  def read(files: Iterable[File]): Iterator[T] = {
    progressMeter.startRun(files.size)
    files.toIterator.flatMap(f => read(open(f), f.getName))
  }

  private def open(f: File): BufferedSource = {
    if (f.getName.endsWith(".gz"))
      Source.fromInputStream(new GZIPInputStream(new BufferedInputStream(new FileInputStream(f))))
    else
      Source.fromFile(f)
  }

  def read(source: Source, sourceName: String = ""): Iterator[T] = new LogEntryIterator(source, sourceName, progressMeter)

  private val processors = Runtime.getRuntime.availableProcessors()

  class LogEntryIterator(source: Source, sourceName: String = "", progressMeter: ProgressMeter) extends AbstractIterator[T] {

    private val lines = new SourceLineIterator(source.getLines(), sourceName, strict, keepFullText)
    private var recordCount = 0
    private val queue = new ArrayBlockingQueue[Future[T]](processors * processors * 2)
    private var current: T = null.asInstanceOf[T]
    private var started = false
    private val reader = new Thread(new Runnable {
      override def run(): Unit = {
        var more = true
        do {
          val f = try {
            val entry = logType.readLinesForNextRecord(lines)
            if (entry != null) {
              parseInTheFuture(entry)
            } else {
              more = false
              Future.successful(null.asInstanceOf[T])
            }
          }
          catch {
            case e: Exception => Future.failed(e)
          }
          queue.put(f)
        } while (more)
      }
    }, s"$sourceName-reader")
    reader.setDaemon(true)

    if (sourceName != "")
      progressMeter.startFile(sourceName)

    override def hasNext: Boolean = {
      ensureStarted()
      current != null
    }
{'fraction_non_alphanumeric': 0.10814870446864439, 'fraction_numerical': 0.0005632745024408562, 'mean_word_length': 3.258193445243805, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '2276342', 'n_tokens_mistral': 1577, 'n_tokens_neox': 1546, 'n_words': 356}