<commit_before>package com.bwsw.sj.engine.batch.task

import java.util.concurrent.Callable

import com.bwsw.common.LeaderLatch
import com.bwsw.sj.common.DAL.model.module.BatchInstance
import com.bwsw.sj.common.utils.EngineLiterals
import com.bwsw.sj.engine.core.entities._
import com.bwsw.sj.engine.core.state.CommonModuleService
import com.bwsw.sj.engine.core.batch.{BatchCollector, WindowRepository, BatchStreamingExecutor, BatchStreamingPerformanceMetrics}
import com.bwsw.sj.engine.batch.task.input.EnvelopeFetcher
import org.apache.curator.framework.CuratorFrameworkFactory
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier
import org.apache.curator.retry.ExponentialBackoffRetry
import org.slf4j.LoggerFactory

import scala.collection.mutable

class BatchTaskEngine(batchCollector: BatchCollector,
                      instance: BatchInstance,
                      moduleService: CommonModuleService,
                      inputService: EnvelopeFetcher,
                      performanceMetrics: BatchStreamingPerformanceMetrics) extends Callable[Unit] {

  private val currentThread = Thread.currentThread()
  currentThread.setName(s"batch-task-engine")
  private val logger = LoggerFactory.getLogger(this.getClass)
  private val inputs = instance.getInputsWithoutStreamMode()
  private val executor = moduleService.executor.asInstanceOf[BatchStreamingExecutor[AnyRef]]
  private val moduleTimer = moduleService.moduleTimer
  private var retrievableStreams = instance.getInputsWithoutStreamMode()
  private var counterOfBatchesPerStream = createCountersOfBatches()
  private val currentWindowPerStream = createStorageOfWindows()
  private val collectedWindowPerStream = mutable.Map[String, Window]()
  private val windowRepository = new WindowRepository(instance)
  private val barrierMasterNode = EngineLiterals.batchInstanceBarrierPrefix + instance.name
  private val leaderMasterNode = EngineLiterals.batchInstanceLeaderPrefix + instance.name
  private val zkHosts = instance.coordinationService.provider.hosts.toSet
  private val curatorClient = createCuratorClient()
  private val barrier = new DistributedDoubleBarrier(curatorClient, barrierMasterNode, instance.executionPlan.tasks.size())
  private val leaderLatch = new LeaderLatch(zkHosts, leaderMasterNode)
  leaderLatch.start()

  private def createCountersOfBatches() = {
    mutable.Map(inputs.map(x => (x, 0)): _*)
  }

  private def createStorageOfWindows() = {
    mutable.Map(inputs.map(x => (x, new Window(x))): _*)
  }

  private def createCuratorClient() = {
    val curatorClient = CuratorFrameworkFactory.newClient(zkHosts.mkString(","), new ExponentialBackoffRetry(1000, 3))
    curatorClient.start()
    curatorClient.getZookeeperClient.blockUntilConnectedOrTimedOut()

    curatorClient
  }

  /**
    * It is in charge of running a basic execution logic of batch task engine
    */
  override def call(): Unit = {
    logger.info(s"Run batch task engine in a separate thread of execution service.")
    logger.debug(s"Invoke onInit() handler.")
    executor.onInit()

    while (true) {
      retrieveAndProcessEnvelopes()

      if (moduleTimer.isTime) {
        logger.debug(s"Invoke onTimer() handler.")
        executor.onTimer(System.currentTimeMillis() - moduleTimer.responseTime)
        moduleTimer.reset()
      }
    }
  }

  private def retrieveAndProcessEnvelopes() = {
    retrievableStreams.foreach(stream => {
      logger.debug(s"Retrieve an available envelope from '$stream' stream.")
      inputService.get(stream) match {
        case Some(envelope) =>
          batchCollector.onReceive(envelope)
          processBatches()

          if (allWindowsCollected) {
            onWindow()
          }

        case None => println("None") //todo
      }
    })
  }<commit_msg>update t-streams and delete a 'todo'<commit_after>package com.bwsw.sj.engine.batch.task

import java.util.concurrent.Callable

import com.bwsw.common.LeaderLatch
import com.bwsw.sj.common.DAL.model.module.BatchInstance
import com.bwsw.sj.common.utils.EngineLiterals
import com.bwsw.sj.engine.core.entities._
import com.bwsw.sj.engine.core.state.CommonModuleService
import com.bwsw.sj.engine.core.batch.{BatchCollector, WindowRepository, BatchStreamingExecutor, BatchStreamingPerformanceMetrics}
import com.bwsw.sj.engine.batch.task.input.EnvelopeFetcher
import org.apache.curator.framework.CuratorFrameworkFactory
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier
import org.apache.curator.retry.ExponentialBackoffRetry
import org.slf4j.LoggerFactory

import scala.collection.mutable

class BatchTaskEngine(batchCollector: BatchCollector,
                      instance: BatchInstance,
                      moduleService: CommonModuleService,
                      inputService: EnvelopeFetcher,
                      performanceMetrics: BatchStreamingPerformanceMetrics) extends Callable[Unit] {

  private val currentThread = Thread.currentThread()
  currentThread.setName(s"batch-task-engine")
  private val logger = LoggerFactory.getLogger(this.getClass)
  private val inputs = instance.getInputsWithoutStreamMode()
  private val executor = moduleService.executor.asInstanceOf[BatchStreamingExecutor[AnyRef]]
  private val moduleTimer = moduleService.moduleTimer
  private var retrievableStreams = instance.getInputsWithoutStreamMode()
  private var counterOfBatchesPerStream = createCountersOfBatches()
  private val currentWindowPerStream = createStorageOfWindows()
  private val collectedWindowPerStream = mutable.Map[String, Window]()
  private val windowRepository = new WindowRepository(instance)
  private val barrierMasterNode = EngineLiterals.batchInstanceBarrierPrefix + instance.name
  private val leaderMasterNode = EngineLiterals.batchInstanceLeaderPrefix + instance.name
  private val zkHosts = instance.coordinationService.provider.hosts.toSet
  private val curatorClient = createCuratorClient()
  private val barrier = new DistributedDoubleBarrier(curatorClient, barrierMasterNode, instance.executionPlan.tasks.size())
  private val leaderLatch = new LeaderLatch(zkHosts, leaderMasterNode)
  leaderLatch.start()

  private def createCountersOfBatches() = {
    mutable.Map(inputs.map(x => (x, 0)): _*)
  }

  private def createStorageOfWindows() = {
    mutable.Map(inputs.map(x => (x, new Window(x))): _*)
  }

  private def createCuratorClient() = {
    val curatorClient = CuratorFrameworkFactory.newClient(zkHosts.mkString(","), new ExponentialBackoffRetry(1000, 3))
    curatorClient.start()
    curatorClient.getZookeeperClient.blockUntilConnectedOrTimedOut()

    curatorClient
  }

  /**
    * It is in charge of running a basic execution logic of batch task engine
    */
  override def call(): Unit = {
    logger.info(s"Run batch task engine in a separate thread of execution service.")
    logger.debug(s"Invoke onInit() handler.")
    executor.onInit()

    while (true) {
      retrieveAndProcessEnvelopes()

      if (moduleTimer.isTime) {
        logger.debug(s"Invoke onTimer() handler.")
        executor.onTimer(System.currentTimeMillis() - moduleTimer.responseTime)
        moduleTimer.reset()
      }
    }
  }

  private def retrieveAndProcessEnvelopes() = {
    retrievableStreams.foreach(stream => {
      logger.debug(s"Retrieve an available envelope from '$stream' stream.")
      inputService.get(stream) match {
        case Some(envelope) =>
          batchCollector.onReceive(envelope)
          processBatches()

          if (allWindowsCollected) {
            onWindow()
          }

        case None =>
      }
    })
  }
{'fraction_non_alphanumeric': 0.08948751642575559, 'fraction_numerical': 0.0018396846254927727, 'mean_word_length': 4.153012863913338, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '4269308', 'n_tokens_mistral': 2182, 'n_tokens_neox': 2142, 'n_words': 427}