<commit_before>package com.jeffharwell.commoncrawl.createcorpus

import java.io.File;
import java.io.BufferedInputStream
import java.io.InputStream
import java.io.FileInputStream
import java.io.BufferedReader
import java.io.InputStreamReader
import java.util.zip.GZIPInputStream
import scala.collection.mutable.ListBuffer
import scala.collection.immutable.List


class Parser(inputstream: InputStream) extends Iterator[List[String]] {
  val linebuffer = scala.collection.mutable.ArrayBuffer.empty[Byte]
  val recordbuffer = scala.collection.mutable.ArrayBuffer.empty[Byte]
  val newlinebyte = "\n".getBytes("UTF-8").last
  var readbuffersize: Int = 1024
  var linecount: Int = 0
  var recordcount: Int = 0
  var recordlength: Int = 0
  var contentbytesread: Int = 0
  var bufferbytesread: Int = 0
  var printnextline: Boolean = false
  var wetrecords: ListBuffer[String] = new ListBuffer()
  var gzcInputStream: GZIPCompatibilityWrapper = new GZIPCompatibilityWrapper(inputstream)
  var gzInputStream: GZIPInputStream = new GZIPInputStream(gzcInputStream)
  var hasnext: Boolean = true

  /* Here is a wrapper class that wraps an InputStream
   * but always returns > 0 when .available() is called.
   * This will cause GZIPInputStream to always make another 
   * call to the InputStream to check for an additional 
   * concatenated GZIP file in the stream.
   */
  class GZIPCompatibilityWrapper(inputstream: InputStream) extends InputStream {
      val is: InputStream = inputstream

      override def read(): Int = {
          return(is.read());
      }

      override def read(b: Array[Byte]): Int = {
          return(is.read(b));
      }

      override def read(b: Array[Byte], off: Int, len: Int): Int = {
          return(is.read(b, off, len));
      }

      override def close(): Unit = {
          is.close();
      }

      override def available(): Int = {
          // Always say that we have 1 more byte in the
          // buffer, even when we don't
          val a: Int = is.available();
          if (a == 0) {
              return(1);
          } else {
              return(a);
          }
      }
  }

  /*
   * This is the next() implementation for our Iterator
   *
   * @return A list of individual WET records
   */
  def next(): List[String] = {
    var result = false
    do {
      result = this.parseBuffer
    } while (!result)

    return this.returnRecord() // will return an empty list if we hit end of stream
  }

  /*
   * This is the hasNext() implementation for the iterator. The variable hasnext
   * is set to false by the parseBuffer() method when we reach the end of the InputStream
   *
   * Note that this isn't entirely accurate. Because we are reading an InputStream on the fly
   * we actually don't know if there is an entire record left or not, we won't know until we 
   * try next() and actually hit the end of the stream. So. next() may return an empty list
   * after .hasNext() has indicated that there is actually another record present.
   *
   * I suppose we could get around this by reading ahead one record ... hmm
   *
   * @return boolean indicating if there is another record in the iterator
   */
  def hasNext(): Boolean = {
    return hasnext
  }
      

  /*
   * This is the method called by the program populating the a read buffer.
   * The read buffer will contain some small chunk of a WET file, the job of this
   * class is to take a sequence of those buffers and return a set of distinct records
   * from the WET file.
   *
   * @return true if we have found a complete record or reach end of stream, false otherwise
   */

  def parseBuffer(): Boolean = {
    val buffer = new Array[ Byte ]( readbuffersize )
    var length: Int = 0<commit_msg>Modified Parser.scala iterator implementation so that hasNext() works correctly.<commit_after>package com.jeffharwell.commoncrawl.createcorpus

import java.io.File;
import java.io.BufferedInputStream
import java.io.InputStream
import java.io.FileInputStream
import java.io.BufferedReader
import java.io.InputStreamReader
import java.util.zip.GZIPInputStream
import scala.collection.mutable.ListBuffer
import scala.collection.immutable.List


class Parser(inputstream: InputStream) extends Iterator[List[String]] {

  /* Here is a wrapper class that wraps an InputStream
   * but always returns > 0 when .available() is called.
   * This will cause GZIPInputStream to always make another 
   * call to the InputStream to check for an additional 
   * concatenated GZIP file in the stream.
   */
  class GZIPCompatibilityWrapper(inputstream: InputStream) extends InputStream {
      val is: InputStream = inputstream

      override def read(): Int = {
          return(is.read());
      }

      override def read(b: Array[Byte]): Int = {
          return(is.read(b));
      }

      override def read(b: Array[Byte], off: Int, len: Int): Int = {
          return(is.read(b, off, len));
      }

      override def close(): Unit = {
          is.close();
      }

      override def available(): Int = {
          // Always say that we have 1 more byte in the
          // buffer, even when we don't
          val a: Int = is.available();
          if (a == 0) {
              return(1);
          } else {
              return(a);
          }
      }
  }

  /*
   * Here is the constructor.
   */

  val linebuffer = scala.collection.mutable.ArrayBuffer.empty[Byte]
  val recordbuffer = scala.collection.mutable.ArrayBuffer.empty[Byte]
  val newlinebyte = "\n".getBytes("UTF-8").last
  var readbuffersize: Int = 1024
  var linecount: Int = 0
  var recordcount: Int = 0
  var recordlength: Int = 0
  var contentbytesread: Int = 0
  var bufferbytesread: Int = 0
  var printnextline: Boolean = false
  var wetrecords: ListBuffer[String] = new ListBuffer()
  var gzcInputStream: GZIPCompatibilityWrapper = new GZIPCompatibilityWrapper(inputstream)
  var gzInputStream: GZIPInputStream = new GZIPInputStream(gzcInputStream)
  var hasnext: Boolean = true
  var nextrecord = List[String]()

  /*
   * Upon initialization of the object we want to read the first chunk of data from the 
   * buffer and populate the next result so that if .hasNext() is called we can tell the 
   * truth. We do this because there is no way to reliably tell if an inputstream is really 
   * empty without trying to read the next value. Much like Drivesavers, we won't say we can
   * get the information unless we already have it ;)
   */

  /*
   * This is the next() implementation for our Iterator
   *
   * @return A list of individual WET records
   */
  def next(): List[String] = {
    // docs say behaviour when next() is called after hasNext() would
    // report false is undefined, in our case we will throw a runtime
    // as I think it likely the code has made a mistake that the programmer
    // should know about.
    if (!this.hasNext()) {
      throw new RuntimeException(".next() called when no records remain in the iterator")
    }

    // Copy nextrecord into the record we are going to return
    var recordtoreturn: List[String] = nextrecord;

    // Go ahead and try to get the record we will return if next() is called again
    nextrecord = privatenext()

    // Return the record we already had
    return recordtoreturn
  }

  /*
   * This command actually calls parseBuffer and extracts the next record
   *
   * @return A list of individual WET records
   */
  def privatenext(): List[String] = {
    // parseBuffer will return true when it gets a complete record. It will also set
    // the hasnext variable to false when it hits the end of the stream
    var result = false
    do {
      result = this.parseBuffer
    } while (!result && hasnext) // if the input stream ends in the middle of a record
                                 // we want to catch that (hasnext == false) and break
                                 // the loop. Otherwise an incomplete record at the end of 
                                 // a file could cause an infinite loop.

    return this.returnRecord()
  }


  /*
   * This is the hasNext() implementation for the iterator. The variable hasnext
   * is set to false by the parseBuffer() method when we reach the end of the InputStream
   *
   * Note that this isn't entirely accurate. Because we are reading an InputStream on the fly
   * we actually don't know if there is an entire record left or not, we won't know until we 
   * try next() and actually hit the end of the stream. So. next() may return an empty list
   * after .hasNext() has indicated that there is actually another record present.
   *
   * I suppose we could get around this by reading ahead one record ... hmm
   *
   * @return boolean indicating if there is another record in the iterator
   */
  def hasNext(): Boolean = {
    return hasnext
  }
      

  /*
   * This is the method called by the program populating the a read buffer.
   * The read buffer will contain some small chunk of a WET file, the job of this
   * class is to take a sequence of those buffers and return a set of distinct records
   * from the WET file.
   *
   * @return true if we have found a complete record or reach end of stream, false otherwise
   */

  def parseBuffer(): Boolean = {
    val buffer = new Array[ Byte ]( readbuffersize )
    var length: Int = 0
{'fraction_non_alphanumeric': 0.07786752827140549, 'fraction_numerical': 0.0032310177705977385, 'mean_word_length': 2.723336006415397, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7253218', 'n_tokens_mistral': 2592, 'n_tokens_neox': 2500, 'n_words': 1095}