<commit_before>package teleporter.integration.core

import java.util.UUID
import java.util.concurrent.{Callable, TimeUnit}

import akka.Done
import akka.actor.{Actor, ActorRef, Props}
import akka.stream.KillSwitch
import com.google.common.base.Charsets
import com.google.common.cache.CacheBuilder
import com.google.common.hash.Hashing
import com.markatta.akron.CronTab.{UnSchedule, UnScheduled}
import com.markatta.akron.{CronExpression, CronTab}
import org.apache.logging.log4j.scala.Logging
import teleporter.integration.core.Streams.{ExecuteStream, _}
import teleporter.integration.script.ScriptEngines

import scala.collection.concurrent.TrieMap
import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration.{Duration, FiniteDuration}
import scala.util.{Failure, Success}

/**
  * Created by kui.dai on 2016/7/7.
  */
case class StreamState(returnValues: ReturnValues)

class StreamsActor()(implicit center: TeleporterCenter) extends Actor with Logging {

  import center.materializer
  import context.dispatcher

  private val streamLogicCache = CacheBuilder.newBuilder()
    .maximumSize(100)
    .expireAfterWrite(5, TimeUnit.MINUTES)
    .build[String, StreamLogic]()
  private val streamStates = TrieMap[String, StreamState]()
  private val cronCache = mutable.Map[Any, UUID]()

  @scala.throws[Exception](classOf[Exception])
  override def preStart(): Unit = {
    super.preStart()
  }

  override def receive: Receive = schedule.orElse(command)

  def schedule: Receive = {
    case DelayCommand(command, delay) ⇒
      delay match {
        case Duration.Zero ⇒ self ! command
        case d: FiniteDuration ⇒ context.system.scheduler.scheduleOnce(d, self, command)
        case _ ⇒
      }
    case CronCommand(command, cron) ⇒
      cronCache.remove(command).foreach {
        jobId ⇒
          logger.info(s"Cancel exists cron job, $command, $jobId")
          center.crontab ! UnSchedule(jobId)
          center.crontab ! CronTab.Schedule(self, command, CronExpression(cron))
      }
      center.crontab ! CronTab.Schedule(self, command, CronExpression(cron))
    case cronSchedule@CronTab.Scheduled(jobId, _, message) ⇒
      logger.info(s"CronSchedule $cronSchedule was start")
      cronCache += (message → jobId)
    case UnScheduled(jobId) ⇒
      logger.info(s"Cron job was cancel, $jobId")
  }<commit_msg>Update Streams.scala<commit_after>package teleporter.integration.core

import java.util.UUID
import java.util.concurrent.{Callable, TimeUnit}

import akka.Done
import akka.actor.{Actor, ActorRef, Props}
import akka.stream.KillSwitch
import com.google.common.base.Charsets
import com.google.common.cache.CacheBuilder
import com.google.common.hash.Hashing
import com.markatta.akron.CronTab.{UnSchedule, UnScheduled}
import com.markatta.akron.{CronExpression, CronTab}
import org.apache.logging.log4j.scala.Logging
import teleporter.integration.core.Streams.{ExecuteStream, _}
import teleporter.integration.script.ScriptEngines

import scala.collection.concurrent.TrieMap
import scala.collection.mutable
import scala.concurrent.Future
import scala.concurrent.duration.{Duration, FiniteDuration}
import scala.util.{Failure, Success}

/**
  * Created by kui.dai on 2016/7/7.
  */
case class StreamState(returnValues: ReturnValues)

class StreamsActor()(implicit center: TeleporterCenter) extends Actor with Logging {

  import center.materializer
  import context.dispatcher

  private val streamLogicCache = CacheBuilder.newBuilder()
    .maximumSize(100)
    .expireAfterWrite(5, TimeUnit.MINUTES)
    .build[String, StreamLogic]()
  private val streamStates = TrieMap[String, StreamState]()
  private val cronCache = mutable.Map[Any, UUID]()

  @scala.throws[Exception](classOf[Exception])
  override def preStart(): Unit = {
    super.preStart()
  }

  override def receive: Receive = schedule.orElse(command)

  def schedule: Receive = {
    case DelayCommand(command, delay) ⇒
      delay match {
        case Duration.Zero ⇒ self ! command
        case d: FiniteDuration ⇒ context.system.scheduler.scheduleOnce(d, self, command)
        case _ ⇒
      }
    case CronCommand(command, cron) ⇒
      cronCache.remove(command).foreach {
        jobId ⇒
          logger.info(s"Cancel exists cron job, $command, $jobId")
          center.crontab ! UnSchedule(jobId)
      }
      logger.info(s"Schedule new job, $command")
      center.crontab ! CronTab.Schedule(self, command, CronExpression(cron))
    case <EMAIL>@Cron<EMAIL>.Scheduled(jobId, _, message) ⇒
      logger.info(s"CronSchedule $cronSchedule was start")
      cronCache += (message → jobId)
    case UnScheduled(jobId) ⇒
      logger.info(s"Cron job was cancel, $jobId")
  }
{'fraction_non_alphanumeric': 0.10574468085106382, 'fraction_numerical': 0.004680851063829788, 'mean_word_length': 4.217536071032186, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '5322199', 'n_tokens_mistral': 1514, 'n_tokens_neox': 1473, 'n_words': 325}