<commit_before>package com.mindcandy.waterfall.io

import java.util.concurrent.TimeUnit

import akka.actor.ActorSystem
import akka.util.Timeout
import com.mindcandy.waterfall.IntermediateFormat
import com.mindcandy.waterfall.Intermediate
import com.mindcandy.waterfall.IOConfig
import com.mindcandy.waterfall.IOSource
import com.typesafe.scalalogging.slf4j.Logging
import com.mindcandy.waterfall.IOOps
import com.mindcandy.waterfall.RowSeparator._
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.util.Try
import com.mindcandy.waterfall.intermediate.FileIntermediate
import spray.http._
import spray.client.pipelining._

case class HttpIOConfig(url: String, timeout: Int = 5000) extends IOConfig

case class HttpIOSource[A <: AnyRef](config: HttpIOConfig, override val columnSeparator: Option[String] = None, val rowSeparator: RowSeparator = NewLine)
    extends IOSource[A]
    with IOOps[A] {

  def retrieveInto[I <: Intermediate[A]](intermediate: I)(implicit format: IntermediateFormat[A]) = {
    val inputContent = rowSeparator match {
      case NewLine => fileContent.map {
        _.lines.map(fromLine(_))
      }
      case NoSeparator => {
        fileContent.map {
          _.lines.mkString("") match {
            case combinedData if !combinedData.isEmpty => Iterator(fromLine(combinedData))
            case _ => Iterator[A]()
          }
        }
      }
    }

    inputContent.flatMap { content =>
      intermediate.write(content).map { _ =>
        logger.info("Retrieving into %s from %s completed".format(intermediate, config))
      }
    }
  }

  implicit val system = ActorSystem("waterfall-httpio")
  implicit val executionContext: ExecutionContext = system.dispatcher
  implicit val timeout = Timeout(config.timeout, TimeUnit.MILLISECONDS)
  val pipeline: HttpRequest => Future[String] = sendReceive ~> unmarshal[String]

  private[this] def fileContent: Try[String] = Try {
    Await.result(pipeline(Get(config.url)), Duration.Inf)
  }
}

trait MultipleHttpIOConfig extends IOConfig {
  def urls: List[String]
  def combinedFileUrl: String
  def timeout: Int = 5000
  override def url = urls.mkString(";")
  override def toString = "MultipleHttpIOConfig(%s)".format(urls)
}

case class MultipleHttpIOSource[A <: AnyRef](config: MultipleHttpIOConfig) extends IOSource[A] with Logging {
  def retrieveInto[I <: Intermediate[A]](intermediate: I)(implicit format: IntermediateFormat[A]) = {
    val combinedIntermediate = FileIntermediate[A](config.combinedFileUrl)<commit_msg>* adding proper actor system termination<commit_after>package com.mindcandy.waterfall.io

import java.util.concurrent.TimeUnit

import akka.actor.ActorSystem
import akka.util.Timeout
import com.mindcandy.waterfall.IntermediateFormat
import com.mindcandy.waterfall.Intermediate
import com.mindcandy.waterfall.IOConfig
import com.mindcandy.waterfall.IOSource
import com.typesafe.scalalogging.slf4j.Logging
import com.mindcandy.waterfall.IOOps
import com.mindcandy.waterfall.RowSeparator._
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, ExecutionContext, Future }
import scala.util.Try
import com.mindcandy.waterfall.intermediate.FileIntermediate
import spray.http._
import spray.client.pipelining._

case class HttpIOConfig(url: String, timeout: Int = 5000) extends IOConfig

case class HttpIOSource[A <: AnyRef](config: HttpIOConfig, override val columnSeparator: Option[String] = None, val rowSeparator: RowSeparator = NewLine)
    extends IOSource[A]
    with IOOps[A] {

  def retrieveInto[I <: Intermediate[A]](intermediate: I)(implicit format: IntermediateFormat[A]) = {
    val inputContent = rowSeparator match {
      case NewLine => fileContent.map {
        _.lines.map(fromLine(_))
      }
      case NoSeparator => {
        fileContent.map {
          _.lines.mkString("") match {
            case combinedData if !combinedData.isEmpty => Iterator(fromLine(combinedData))
            case _ => Iterator[A]()
          }
        }
      }
    }
    system.shutdown()
    system.awaitTermination()

    inputContent.flatMap { content =>
      intermediate.write(content).map { _ =>
        logger.info("Retrieving into %s from %s completed".format(intermediate, config))
      }
    }
  }

  implicit val system = ActorSystem("waterfall-httpio")
  implicit val executionContext: ExecutionContext = system.dispatcher
  implicit val timeout = Timeout(config.timeout, TimeUnit.MILLISECONDS)
  val pipeline: HttpRequest => Future[String] = sendReceive ~> unmarshal[String]

  private[this] def fileContent: Try[String] = Try {
    Await.result(pipeline(Get(config.url)), Duration.Inf)
  }
}

trait MultipleHttpIOConfig extends IOConfig {
  def urls: List[String]
  def combinedFileUrl: String
  def timeout: Int = 5000
  override def url = urls.mkString(";")
  override def toString = "MultipleHttpIOConfig(%s)".format(urls)
}

case class MultipleHttpIOSource[A <: AnyRef](config: MultipleHttpIOConfig) extends IOSource[A] with Logging {
  def retrieveInto[I <: Intermediate[A]](intermediate: I)(implicit format: IntermediateFormat[A]) = {
    val combinedIntermediate = FileIntermediate[A](config.combinedFileUrl)
{'fraction_non_alphanumeric': 0.10331608203948629, 'fraction_numerical': 0.0034502587694077054, 'mean_word_length': 4.412863070539419, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, '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': '5321387', 'n_tokens_mistral': 1594, 'n_tokens_neox': 1584, 'n_words': 340}