<commit_before>package actors

import akka.actor.{Props, Actor}
import play.api.libs.json.{JsObject, Json, JsValue}
import play.api.libs.iteratee._
import akka.util.Timeout
import play.api.libs.concurrent.Akka
import scala.concurrent.duration._
import akka.pattern.ask
import org.apache.commons.lang3.RandomStringUtils
import play.api.Play.current
import play.api.libs.concurrent.Execution.Implicits.defaultContext
import play.api.mvc.AnyContent

/**
 * Monitor Manager actor which handles all connections
 */
class MonitorManager extends Actor {

  var sessions = Map[String, Session]()

  /**
   * Request a new session
   *
   * @return
   */
  def requestSession() = {
    var sessionId = ""
    do {
       sessionId = RandomStringUtils.randomAlphanumeric(10)
    } while(sessions.contains(sessionId))

    val (enumerator, channel) = Concurrent.broadcast[JsValue]

    val session = Session(sessionId, enumerator, channel)
    sessions += sessionId -> session
    session
  }

  def receive = {
    case RequestSession() => sender ! requestSession()
    case GetSession(id)   => sender ! sessions.get(id).getOrElse(SessionNotFound())
    case BroadcastRequest(session, request) =>
      sessions.get(session.id) map { found =>
        val result = Json.obj(
          "content" -> request.body,
          "headers" -> request.headers,
          "contentType" -> request.contentType)
        found.channel.push(result)
    }
  }
}

object MonitorManager {

  implicit val timeout = Timeout(4 second)

  lazy val actor = Akka.system.actorOf(Props[MonitorManager])

  def join(sessionId: String): scala.concurrent.Future[(Iteratee[JsValue, _], Enumerator[JsValue])] = {

    actor ? GetSession(sessionId) map {
      case Session(id, enumerator, channel) =>
        val iteratee = Iteratee.foreach[JsValue] { event =>
          // On Message
        }.map { _ =>
          // On Connection Closed
        }

        (iteratee, Enumerator[JsValue](Json.obj("result" -> "Connected")).andThen(enumerator))

      case SessionNotFound() =>
        val iteratee = Done[JsValue, Unit]((), Input.EOF)
        val enumerator = Enumerator[JsValue](Json.obj("error" -> "session not found")).andThen(Enumerator.enumInput(Input.EOF))
        (iteratee, enumerator)
    }
  }<commit_msg>renamed content to body because of conflicts in ember<commit_after>package actors

import akka.actor.{Props, Actor}
import play.api.libs.json.{JsObject, Json, JsValue}
import play.api.libs.iteratee._
import akka.util.Timeout
import play.api.libs.concurrent.Akka
import scala.concurrent.duration._
import akka.pattern.ask
import org.apache.commons.lang3.RandomStringUtils
import play.api.Play.current
import play.api.libs.concurrent.Execution.Implicits.defaultContext
import play.api.mvc.AnyContent

/**
 * Monitor Manager actor which handles all connections
 */
class MonitorManager extends Actor {

  var sessions = Map[String, Session]()

  /**
   * Request a new session
   *
   * @return
   */
  def requestSession() = {
    var sessionId = ""
    do {
       sessionId = RandomStringUtils.randomAlphanumeric(10)
    } while(sessions.contains(sessionId))

    val (enumerator, channel) = Concurrent.broadcast[JsValue]

    val session = Session(sessionId, enumerator, channel)
    sessions += sessionId -> session
    session
  }

  def receive = {
    case RequestSession() => sender ! requestSession()
    case GetSession(id)   => sender ! sessions.get(id).getOrElse(SessionNotFound())
    case BroadcastRequest(session, request) =>
      sessions.get(session.id) map { found =>
        val result = Json.obj(
          "body" -> request.body,
          "headers" -> request.headers,
          "contentType" -> request.contentType)
        found.channel.push(result)
    }
  }
}

object MonitorManager {

  implicit val timeout = Timeout(4 second)

  lazy val actor = Akka.system.actorOf(Props[MonitorManager])

  def join(sessionId: String): scala.concurrent.Future[(Iteratee[JsValue, _], Enumerator[JsValue])] = {

    actor ? GetSession(sessionId) map {
      case Session(id, enumerator, channel) =>
        val iteratee = Iteratee.foreach[JsValue] { event =>
          // On Message
        }.map { _ =>
          // On Connection Closed
        }

        (iteratee, Enumerator[JsValue](Json.obj("result" -> "Connected")).andThen(enumerator))

      case SessionNotFound() =>
        val iteratee = Done[JsValue, Unit]((), Input.EOF)
        val enumerator = Enumerator[JsValue](Json.obj("error" -> "session not found")).andThen(Enumerator.enumInput(Input.EOF))
        (iteratee, enumerator)
    }
  }
{'fraction_non_alphanumeric': 0.12241341755608799, 'fraction_numerical': 0.0017425397516880855, 'mean_word_length': 3.271627906976744, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, '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': '6791356', 'n_tokens_mistral': 1407, 'n_tokens_neox': 1363, 'n_words': 298}