<commit_before>package com.blinkbox.books.search.ingester

import akka.actor.ActorRef
import com.blinkbox.books.messaging.ErrorHandler
import com.blinkbox.books.messaging.Event
import com.blinkbox.books.messaging.ReliableEventHandler
import com.typesafe.scalalogging.slf4j.StrictLogging
import java.io.IOException
import java.util.concurrent.TimeoutException
import spray.can.Http.ConnectionException
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.Future
import scala.util.Try
import scala.xml.XML
import java.io.ByteArrayInputStream
import scala.xml.NodeSeq
import javax.xml.transform.TransformerFactory
import javax.xml.transform.stream.StreamSource
import java.io.StringReader
import java.io.InputStreamReader
import javax.xml.transform.stream.StreamResult
import java.io.StringWriter

trait XmlHandler {
  def handleXml(data: String): Future[Unit]
}

class BookMetadataTransformer(xmlHandler: XmlHandler, errorHandler: ErrorHandler, retryInterval: FiniteDuration)
  extends ReliableEventHandler(errorHandler, retryInterval) with StrictLogging {

  override def handleEvent(event: Event, originalSender: ActorRef): Future[Unit] = {
    val xml = XML.load(new ByteArrayInputStream(event.body.content))

    // Get a suitable Transformer.
    val transformer = xml.label match {
      case "book" => bookTransformer
      case "undistribute" => undistributeTransformer
      case "book-price" => priceTransformer
      case _ => throw new IllegalArgumentException(s"Unexpected XML with root element: ${xml.label}")
    }

    // Transform the input.
    val xmlSource = new StreamSource(new ByteArrayInputStream(event.body.content))
    val output = new StringWriter
    transformer.transform(xmlSource, new StreamResult(output))

    // Pass it on.
    xmlHandler.handleXml(output.toString)
  }

  override def isTemporaryFailure(e: Throwable) =
    e.isInstanceOf[IOException] || e.isInstanceOf[TimeoutException] ||
      Option(e.getCause).exists(isTemporaryFailure)

  // Transformers used for incoming XML messages.
  // These are not thread safe hence an instance created for each Actor.
  val bookTransformer = transformer(xsltSource("/book-to-solr.xsl"))
  val undistributeTransformer = transformer(xsltSource("/undistribute-book.xsl"))
  val priceTransformer = transformer(xsltSource("/price-to-solr.xsl"))

  private def transformer(xsltSource: StreamSource) = TransformerFactory.newInstance().newTransformer(xsltSource)
  private def xsltSource(filename: String) = new StreamSource(new InputStreamReader(getClass.getResourceAsStream(filename)))

}
<commit_msg>Make ingestion faster by doing all of the processing work inside futures.<commit_after>package com.blinkbox.books.search.ingester

import akka.actor.ActorRef
import com.blinkbox.books.messaging.ErrorHandler
import com.blinkbox.books.messaging.Event
import com.blinkbox.books.messaging.ReliableEventHandler
import com.typesafe.scalalogging.slf4j.StrictLogging
import java.io.IOException
import java.util.concurrent.TimeoutException
import spray.can.Http.ConnectionException
import scala.concurrent.duration.FiniteDuration
import scala.concurrent.Future
import scala.util.Try
import scala.xml.XML
import java.io.ByteArrayInputStream
import scala.xml.NodeSeq
import javax.xml.transform.TransformerFactory
import javax.xml.transform.stream.StreamSource
import java.io.StringReader
import java.io.InputStreamReader
import javax.xml.transform.stream.StreamResult
import java.io.StringWriter

trait XmlHandler {
  def handleXml(data: String): Future[Unit]
}

class BookMetadataTransformer(xmlHandler: XmlHandler, errorHandler: ErrorHandler, retryInterval: FiniteDuration)
  extends ReliableEventHandler(errorHandler, retryInterval) with StrictLogging {

  override def handleEvent(event: Event, originalSender: ActorRef): Future[Unit] = {
    for (
        outputMsg <- Future(transformMessage(event.body.content));
        _ <- xmlHandler.handleXml(outputMsg))
      yield ()
  }

  override def isTemporaryFailure(e: Throwable) =
    e.isInstanceOf[IOException] || e.isInstanceOf[TimeoutException] ||
      Option(e.getCause).exists(isTemporaryFailure)

  private def transformMessage(bytes: Array[Byte]): String = {
    val xml = XML.load(new ByteArrayInputStream(bytes))

    // Get a suitable Transformer.
    val transformer = xml.label match {
      case "book" => bookTransformer
      case "undistribute" => undistributeTransformer
      case "book-price" => priceTransformer
      case _ => throw new IllegalArgumentException(s"Unexpected XML with root element: ${xml.label}")
    }

    // Transform the input.
    val xmlSource = new StreamSource(new ByteArrayInputStream(bytes))
    val output = new StringWriter
    transformer.transform(xmlSource, new StreamResult(output))
    output.toString
  }

  // Transformers used for incoming XML messages.
  // These are not thread safe hence an instance created for each Actor.
  val bookTransformer = transformer(xsltSource("/book-to-solr.xsl"))
  val undistributeTransformer = transformer(xsltSource("/undistribute-book.xsl"))
  val priceTransformer = transformer(xsltSource("/price-to-solr.xsl"))

  private def transformer(xsltSource: StreamSource) = TransformerFactory.newInstance().newTransformer(xsltSource)
  private def xsltSource(filename: String) = new StreamSource(new InputStreamReader(getClass.getResourceAsStream(filename)))

}

{'fraction_non_alphanumeric': 0.09087537091988131, 'fraction_numerical': 0.000370919881305638, 'mean_word_length': 5.256380510440835, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 24}, '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': '2408007', 'n_tokens_mistral': 1532, 'n_tokens_neox': 1485, 'n_words': 342}