<commit_before>package eu.themerius.docelemstore

import eu.themerius.docelemstore.utils.Stats.time

import akka.actor.{ Props, Actor }
import akka.routing.ActorRefRoutee
import akka.routing.Router
import akka.routing.RoundRobinRoutingLogic
import akka.event.Logging

import org.fusesource.stomp.jms._
import javax.jms._

import java.net.URI

import org.fusesource.stomp.client.Constants._
import org.fusesource.stomp.codec.StompFrame
import org.fusesource.stomp.client.Stomp

// Stomp Callbacks
import org.fusesource.stomp.client.Callback
import org.fusesource.stomp.client.CallbackConnection

import scala.collection.JavaConverters._

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.TimeoutException

import com.typesafe.config.ConfigFactory

case class Consume(header: Map[String, String], message: String)
case class CustomModel(msg: String, model: ModelTransRules)
case class Reply(content: String, to: String, trackingNr: String)
case class Accounting(event: String, query: String, trackingNr: String, unit: String)

class Gate extends Actor {

  val log = Logging(context.system, this)
  log.info(s"Starting ${BuildInfo.name}:${BuildInfo.version}.")

  var latestErrorLog = ""

  val conf = ConfigFactory.load
  val brokerUri = conf.getString("docelem-store.broker.uri")
  val brokerUsr = conf.getString("docelem-store.broker.usr")
  val brokerPwd = conf.getString("docelem-store.broker.pwd")
  val brokerQueue = conf.getString("docelem-store.broker.queue")
  val brokerBilling = conf.getString("docelem-store.broker.billing")

  // Connect to the broker
  val factory = new StompJmsConnectionFactory
  factory.setBrokerURI(brokerUri)

  val elistener = new ExceptionListener {
    override def onException(jmse: JMSException) {
      log.error(s"Connection Error: $jmse")
      // TODO: cleanup connection, retry connection and don't die... :)
      log.warning("Because I'm depressed of the connection error, I'll doing suicide!")
      System.exit(1)
    }
  }

  val connection = factory.createConnection(brokerUsr, brokerPwd)
  connection.setExceptionListener(elistener);
  connection.setClientID(s"jms.topic.${java.util.UUID.randomUUID.toString}")
  connection.start

  val session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)

  val queue = new StompJmsDestination(brokerQueue)
  val consumer = session.createConsumer(queue)

  val listener = new MessageListener {

    override def onMessage(message: Message) {

      val headers = for(prop <- message.getPropertyNames.asScala) yield {
        val strProp = prop.asInstanceOf[String]
        strProp -> message.getStringProperty(strProp)
      }

      val replyTo = if (message.getJMSReplyTo != null) {
        message.getJMSReplyTo.toString
      } else {
        ""
      }

      val headerMap = headers.toMap.updated("reply-to", replyTo)

      if (message.isInstanceOf[TextMessage]) {
        self ! Consume(headerMap, message.asInstanceOf[TextMessage].getText)
      }

    }

  }

  // consumer.setMessageListener(listener)

  // Blocking alternative to message listener.
  // This is a easy way to handle back pressure.
  Future {
    while(true) {
      val message = consumer.receive  // block until getting a message
      if (message != null) {  // unwrap message and route it
        val headers = for(prop <- message.getPropertyNames.asScala) yield {
          val strProp = prop.asInstanceOf[String]
          strProp -> message.getStringProperty(strProp)
        }

        val replyTo = if (message.getJMSReplyTo != null) {
          message.getJMSReplyTo.toString
        } else {
          ""
        }

        val headerMap = headers.toMap.updated("reply-to", replyTo)

        if (message.isInstanceOf[TextMessage]) {
          self ! Consume(headerMap, message.asInstanceOf[TextMessage].getText)
        }
      }
    }
  }


  val billing = new StompJmsDestination(brokerBilling)
  val accounting = session.createProducer(billing)

  val routerAccumuloFeeder = {
    val routees = Vector.fill(100) {
      val r = context.actorOf(Props[AccumuloFeeder].withMailbox("akka.actor.blocking-mailbox"))
      context.watch(r)
      ActorRefRoutee(r)
    }
    Router(RoundRobinRoutingLogic(), routees)
  }

  val inMemory = context.actorOf(Props[InMemoryStore])
  // var UNSCHOENER_HACK: akka.actor.ActorRef = null
  //
  // val routerInMemory = {
  //   val routees = Vector.fill(1) {
  //     val r = context.actorOf(Props[InMemoryStore])
  //     UNSCHOENER_HACK = r
  //     context.watch(r)
  //     ActorRefRoutee(r)
  //   }
  //   Router(RoundRobinRoutingLogic(), routees)
  // }

  val accumuloQuery = context.actorOf(Props[AccumuloQueryer])
  // val accumuloQuery = UNSCHOENER_HACK

  // Consume and distribute messages
  def receive = {
    case Consume(header: Map[String, String], textContent: String) => {
      // unwrap message and route it
      var event = header.getOrElse("event", "")
      val contentType = header.getOrElse("content-type", "")
      val replyTo = header.getOrElse("reply-to", "")
      val trackingNr = header.getOrElse("tracking-nr", "")
      var documentId = header.get("document-id")
      var documentLabel = header.get("document-label")

      (contentType, event) match {

        case ("gzip-xml", "onlyStoreXMI") => {

          log.info("(Gate) got gzipped XCAS, simply write into Accumulo.")

          val model = new Model with ModelTransRules {
            def deserialize(m: Array[Byte]) = this
            def serialize: Array[Byte] = Array[Byte]()
            override def getDocumentId = documentId<commit_msg>Gate corrects documentLabel on demand...<commit_after>package eu.themerius.docelemstore

import eu.themerius.docelemstore.utils.Stats.time

import akka.actor.{ Props, Actor }
import akka.routing.ActorRefRoutee
import akka.routing.Router
import akka.routing.RoundRobinRoutingLogic
import akka.event.Logging

import org.fusesource.stomp.jms._
import javax.jms._

import java.net.URI

import org.fusesource.stomp.client.Constants._
import org.fusesource.stomp.codec.StompFrame
import org.fusesource.stomp.client.Stomp

// Stomp Callbacks
import org.fusesource.stomp.client.Callback
import org.fusesource.stomp.client.CallbackConnection

import scala.collection.JavaConverters._

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.concurrent.TimeoutException

import com.typesafe.config.ConfigFactory

case class Consume(header: Map[String, String], message: String)
case class CustomModel(msg: String, model: ModelTransRules)
case class Reply(content: String, to: String, trackingNr: String)
case class Accounting(event: String, query: String, trackingNr: String, unit: String)

class Gate extends Actor {

  val log = Logging(context.system, this)
  log.info(s"Starting ${BuildInfo.name}:${BuildInfo.version}.")

  var latestErrorLog = ""

  val conf = ConfigFactory.load
  val brokerUri = conf.getString("docelem-store.broker.uri")
  val brokerUsr = conf.getString("docelem-store.broker.usr")
  val brokerPwd = conf.getString("docelem-store.broker.pwd")
  val brokerQueue = conf.getString("docelem-store.broker.queue")
  val brokerBilling = conf.getString("docelem-store.broker.billing")

  // Connect to the broker
  val factory = new StompJmsConnectionFactory
  factory.setBrokerURI(brokerUri)

  val elistener = new ExceptionListener {
    override def onException(jmse: JMSException) {
      log.error(s"Connection Error: $jmse")
      // TODO: cleanup connection, retry connection and don't die... :)
      log.warning("Because I'm depressed of the connection error, I'll doing suicide!")
      System.exit(1)
    }
  }

  val connection = factory.createConnection(brokerUsr, brokerPwd)
  connection.setExceptionListener(elistener);
  connection.setClientID(s"jms.topic.${java.util.UUID.randomUUID.toString}")
  connection.start

  val session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)

  val queue = new StompJmsDestination(brokerQueue)
  val consumer = session.createConsumer(queue)

  val listener = new MessageListener {

    override def onMessage(message: Message) {

      val headers = for(prop <- message.getPropertyNames.asScala) yield {
        val strProp = prop.asInstanceOf[String]
        strProp -> message.getStringProperty(strProp)
      }

      val replyTo = if (message.getJMSReplyTo != null) {
        message.getJMSReplyTo.toString
      } else {
        ""
      }

      val headerMap = headers.toMap.updated("reply-to", replyTo)

      if (message.isInstanceOf[TextMessage]) {
        self ! Consume(headerMap, message.asInstanceOf[TextMessage].getText)
      }

    }

  }

  // consumer.setMessageListener(listener)

  // Blocking alternative to message listener.
  // This is a easy way to handle back pressure.
  Future {
    while(true) {
      val message = consumer.receive  // block until getting a message
      if (message != null) {  // unwrap message and route it
        val headers = for(prop <- message.getPropertyNames.asScala) yield {
          val strProp = prop.asInstanceOf[String]
          strProp -> message.getStringProperty(strProp)
        }

        val replyTo = if (message.getJMSReplyTo != null) {
          message.getJMSReplyTo.toString
        } else {
          ""
        }

        val headerMap = headers.toMap.updated("reply-to", replyTo)

        if (message.isInstanceOf[TextMessage]) {
          self ! Consume(headerMap, message.asInstanceOf[TextMessage].getText)
        }
      }
    }
  }


  val billing = new StompJmsDestination(brokerBilling)
  val accounting = session.createProducer(billing)

  val routerAccumuloFeeder = {
    val routees = Vector.fill(100) {
      val r = context.actorOf(Props[AccumuloFeeder].withMailbox("akka.actor.blocking-mailbox"))
      context.watch(r)
      ActorRefRoutee(r)
    }
    Router(RoundRobinRoutingLogic(), routees)
  }

  val inMemory = context.actorOf(Props[InMemoryStore])
  // var UNSCHOENER_HACK: akka.actor.ActorRef = null
  //
  // val routerInMemory = {
  //   val routees = Vector.fill(1) {
  //     val r = context.actorOf(Props[InMemoryStore])
  //     UNSCHOENER_HACK = r
  //     context.watch(r)
  //     ActorRefRoutee(r)
  //   }
  //   Router(RoundRobinRoutingLogic(), routees)
  // }

  val accumuloQuery = context.actorOf(Props[AccumuloQueryer])
  // val accumuloQuery = UNSCHOENER_HACK

  // Consume and distribute messages
  def receive = {
    case Consume(header: Map[String, String], textContent: String) => {
      // unwrap message and route it
      val event = header.getOrElse("event", "")
      val contentType = header.getOrElse("content-type", "")
      val replyTo = header.getOrElse("reply-to", "")
      val trackingNr = header.getOrElse("tracking-nr", "")
      val documentId = header.get("document-id")
      var documentLabel = header.get("document-label")

      if (documentLabel.nonEmpty && !documentLabel.get.startsWith("header/")) {
        documentLabel = Some(s"header/${documentLabel.get}")
      }

      (contentType, event) match {

        case ("gzip-xml", "onlyStoreXMI") => {

          log.info("(Gate) got gzipped XCAS, simply write into Accumulo.")

          val model = new Model with ModelTransRules {
            def deserialize(m: Array[Byte]) = this
            def serialize: Array[Byte] = Array[Byte]()
            override def getDocumentId = documentId
{'fraction_non_alphanumeric': 0.10934410728101092, 'fraction_numerical': 0.0008596234849136079, 'mean_word_length': 3.390188679245283, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3276544', 'n_tokens_mistral': 3637, 'n_tokens_neox': 3536, 'n_words': 756}