<commit_before>package actors.execution

import com.rabbitmq.client.Connection
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.Channel
import play.api.Play
import play.api.Play.current
import play.api.Logger
import play.api.i18n.Lang
import com.rabbitmq.client.QueueingConsumer
import plm.core.model.lesson.Exercise
import plm.core.lang.ProgrammingLanguage
import java.util.{ Map, HashMap }
import play.api.libs.json.JsObject
import play.api.libs.json.Json
import akka.actor.Props
import org.json.simple.JSONObject
import org.json.simple.parser._
import plm.core.model.lesson.ExecutionProgress
import akka.actor.ActorRef
import akka.pattern.AskTimeoutException
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.UUID
import java.util.Locale
import plm.core.model.json.JSONUtils
import com.fasterxml.jackson.databind.JsonNode

/**
 * @author matthieu
 */

object TribunalActor {
  
  def props(initialLang: Lang) = Props(new TribunalActor(initialLang))
  
  val QUEUE_NAME_REQUEST : String = "worker_in"
  val QUEUE_NAME_REPLY : String = "worker_out"

  val QUEUE_ADDR = Play.configuration.getString("messagequeue.addr").getOrElse("localhost")
  val QUEUE_PORT = Play.configuration.getString("messagequeue.port").getOrElse("5672")
  
  // Connection
  val factory : ConnectionFactory = new ConnectionFactory()
  factory.setHost(QUEUE_ADDR)
  factory.setPort(QUEUE_PORT.toInt)
  var connection : Connection = null
  try {
    connection = factory.newConnection()
  } catch {
    case _ : Exception =>
      Logger.error("[judge] ERROR : no connection with message queue")
  }
}

/**
* Tribunal state
*/
object TribunalState extends Enumeration {
  type TribunalState = Value
  val Off, Waiting, Ack, Launched, Streaming, Replied = Value
}

class TribunalActor(initialLang: Lang) extends ExecutionActor {
  import ExecutionActor._
  import TribunalActor._
  import TribunalState._
  
  val defaultTimeout : Long = 10000

  var state : TribunalState = Off
  var currentLocale: Locale = initialLang.toLocale
  var timeout : Long = 0
  var executionStopped: Boolean = false

  val argsOut: Map[String, Object] = new HashMap[String, Object]
  argsOut.put("x-message-ttl", defaultTimeout.asInstanceOf[Object])
  val argsIn: Map[String, Object] = new HashMap[String, Object]
  argsIn.put("x-message-ttl", 5000.asInstanceOf[Object])

  val channelOut : Channel = connection.createChannel
  channelOut.queueDeclare(QUEUE_NAME_REQUEST, false, false, false, argsOut)

  def receive =  {
    case StartExecution(out, exercise, progLang, code) =>
      startExecution(sender, out, exercise, progLang, code)
    case StopExecution =>
      executionStopped = true
    case UpdateLang(lang: Lang) =>
      currentLocale = lang.toLocale
    case _ =>
      Logger.error("LessonsActor: not supported message")
  }

  def startExecution(plmActor: ActorRef, client: ActorRef, exercise: Exercise, progLang: ProgrammingLanguage, code: String) {
    Future {
      val replyQueue: String = QUEUE_NAME_REPLY + UUID.randomUUID.toString
      val channelIn : Channel = connection.createChannel
      channelIn.queueDeclare(replyQueue, false, false, true, argsIn)
 
      val consumer : QueueingConsumer = new QueueingConsumer(channelIn)
      channelIn.basicConsume(replyQueue, true, consumer)

      val parameters: Map[String, Object] = new HashMap[String, Object]
      parameters.put("exercise", JSONUtils.exerciseToJudgeJSON(exercise)) // To remove incorrect entities' type
      parameters.put("code", code)
      parameters.put("language", progLang.getLang)
      parameters.put("localization", currentLocale.getLanguage)
      parameters.put("replyQueue", replyQueue)
      
      val json: String = JSONUtils.mapToJSON(parameters)

      channelOut.basicPublish("", QUEUE_NAME_REQUEST, null, json.getBytes("UTF-8"))
      timeout = System.currentTimeMillis + defaultTimeout
      state = Waiting
      executionStopped = false
      while(state != Replied && state != Off) {
        val delivery : QueueingConsumer.Delivery = consumer.nextDelivery(1000)
        if (executionStopped) {
          handleStopExecution(plmActor, progLang)
          state = Off
        }
        else if (System.currentTimeMillis > timeout) {
          handleTimeout(plmActor, progLang)
          state = Off
        }
        // The delivery will be "null" if nextDelivery timed out.
        else if (delivery != null) {
          handleMessage(plmActor, client, new String(delivery.getBody, "UTF-8"))
        }
      }
      channelIn.close
    }
  }

  def containsOperations(message: String): Boolean = {
    message.startsWith("""{"cmd":"operations"""")
  }

  def handleMessage(plmActor: ActorRef, client: ActorRef, msg: String): TribunalState = {
    var state: TribunalState = Streaming
    if(containsOperations(msg)) {
      // Handle separately operations to avoid too much parsing
      client ! msg
    }
    else {
      val json: JsonNode = JSONUtils.mapper.readTree(msg)
      val msgType: String = json.path("cmd").asText
      msgType match {
        case "executionResult" =>
          val result: ExecutionProgress = JSONUtils.mapper.treeToValue(json.path("args").path("result"), classOf[ExecutionProgress])
          plmActor ! result
          state = Replied
        case _ =>
      }
    }
    state
  }<commit_msg>Add progLang to the ExecutionProgress send back by judges<commit_after>package actors.execution

import com.rabbitmq.client.Connection
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.Channel
import play.api.Play
import play.api.Play.current
import play.api.Logger
import play.api.i18n.Lang
import com.rabbitmq.client.QueueingConsumer
import plm.core.model.lesson.Exercise
import plm.core.lang.ProgrammingLanguage
import java.util.{ Map, HashMap }
import play.api.libs.json.JsObject
import play.api.libs.json.Json
import akka.actor.Props
import org.json.simple.JSONObject
import org.json.simple.parser._
import plm.core.model.lesson.ExecutionProgress
import akka.actor.ActorRef
import akka.pattern.AskTimeoutException
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.UUID
import java.util.Locale
import plm.core.model.json.JSONUtils
import com.fasterxml.jackson.databind.JsonNode

/**
 * @author matthieu
 */

object TribunalActor {
  
  def props(initialLang: Lang) = Props(new TribunalActor(initialLang))
  
  val QUEUE_NAME_REQUEST : String = "worker_in"
  val QUEUE_NAME_REPLY : String = "worker_out"

  val QUEUE_ADDR = Play.configuration.getString("messagequeue.addr").getOrElse("localhost")
  val QUEUE_PORT = Play.configuration.getString("messagequeue.port").getOrElse("5672")
  
  // Connection
  val factory : ConnectionFactory = new ConnectionFactory()
  factory.setHost(QUEUE_ADDR)
  factory.setPort(QUEUE_PORT.toInt)
  var connection : Connection = null
  try {
    connection = factory.newConnection()
  } catch {
    case _ : Exception =>
      Logger.error("[judge] ERROR : no connection with message queue")
  }
}

/**
* Tribunal state
*/
object TribunalState extends Enumeration {
  type TribunalState = Value
  val Off, Waiting, Ack, Launched, Streaming, Replied = Value
}

class TribunalActor(initialLang: Lang) extends ExecutionActor {
  import ExecutionActor._
  import TribunalActor._
  import TribunalState._
  
  val defaultTimeout : Long = 10000

  var state : TribunalState = Off
  var currentLocale: Locale = initialLang.toLocale
  var timeout : Long = 0
  var executionStopped: Boolean = false

  val argsOut: Map[String, Object] = new HashMap[String, Object]
  argsOut.put("x-message-ttl", defaultTimeout.asInstanceOf[Object])
  val argsIn: Map[String, Object] = new HashMap[String, Object]
  argsIn.put("x-message-ttl", 5000.asInstanceOf[Object])

  val channelOut : Channel = connection.createChannel
  channelOut.queueDeclare(QUEUE_NAME_REQUEST, false, false, false, argsOut)

  def receive =  {
    case StartExecution(out, exercise, progLang, code) =>
      startExecution(sender, out, exercise, progLang, code)
    case StopExecution =>
      executionStopped = true
    case UpdateLang(lang: Lang) =>
      currentLocale = lang.toLocale
    case _ =>
      Logger.error("LessonsActor: not supported message")
  }

  def startExecution(plmActor: ActorRef, client: ActorRef, exercise: Exercise, progLang: ProgrammingLanguage, code: String) {
    Future {
      val replyQueue: String = QUEUE_NAME_REPLY + UUID.randomUUID.toString
      val channelIn : Channel = connection.createChannel
      channelIn.queueDeclare(replyQueue, false, false, true, argsIn)
 
      val consumer : QueueingConsumer = new QueueingConsumer(channelIn)
      channelIn.basicConsume(replyQueue, true, consumer)

      val parameters: Map[String, Object] = new HashMap[String, Object]
      parameters.put("exercise", JSONUtils.exerciseToJudgeJSON(exercise)) // To remove incorrect entities' type
      parameters.put("code", code)
      parameters.put("language", progLang.getLang)
      parameters.put("localization", currentLocale.getLanguage)
      parameters.put("replyQueue", replyQueue)
      
      val json: String = JSONUtils.mapToJSON(parameters)

      channelOut.basicPublish("", QUEUE_NAME_REQUEST, null, json.getBytes("UTF-8"))
      timeout = System.currentTimeMillis + defaultTimeout
      state = Waiting
      executionStopped = false
      while(state != Replied && state != Off) {
        val delivery : QueueingConsumer.Delivery = consumer.nextDelivery(1000)
        if (executionStopped) {
          handleStopExecution(plmActor, progLang)
          state = Off
        }
        else if (System.currentTimeMillis > timeout) {
          handleTimeout(plmActor, progLang)
          state = Off
        }
        // The delivery will be "null" if nextDelivery timed out.
        else if (delivery != null) {
          handleMessage(plmActor, client, progLang, new String(delivery.getBody, "UTF-8"))
        }
      }
      channelIn.close
    }
  }

  def containsOperations(message: String): Boolean = {
    message.startsWith("""{"cmd":"operations"""")
  }

  def handleMessage(plmActor: ActorRef, client: ActorRef, progLang: ProgrammingLanguage, msg: String): TribunalState = {
    var state: TribunalState = Streaming
    if(containsOperations(msg)) {
      // Handle separately operations to avoid too much parsing
      client ! msg
    }
    else {
      val json: JsonNode = JSONUtils.mapper.readTree(msg)
      val msgType: String = json.path("cmd").asText
      msgType match {
        case "executionResult" =>
          val result: ExecutionProgress = JSONUtils.mapper.treeToValue(json.path("args").path("result"), classOf[ExecutionProgress])
          result.language = progLang
          plmActor ! result
          state = Replied
        case _ =>
      }
    }
    state
  }
{'fraction_non_alphanumeric': 0.09870579657309515, 'fraction_numerical': 0.004010207801676996, 'mean_word_length': 3.7216006884681585, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 19, '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': '6802035', 'n_tokens_mistral': 3276, 'n_tokens_neox': 3131, 'n_words': 751}