<commit_before>package com.bwsw.sj.crud.rest.runner

import com.bwsw.sj.common.DAL.model.module.Instance
import com.bwsw.sj.common.DAL.model.{SjStream, TStreamSjStream}
import com.bwsw.sj.common.StreamConstants
import org.slf4j.LoggerFactory

/**
  * One-thread deleting object for instance
  * using synchronous apache http client
  * Created: 16/05/2016
  *
  * @author <NAME>
  */
class InstanceDestroyer(instance: Instance, delay: Long) extends Runnable {
  private val logger = LoggerFactory.getLogger(getClass.getName)

  import InstanceMethods._
  import com.bwsw.sj.common.ModuleConstants._

  def run() = {
    logger.debug(s"Instance: ${instance.name}. Destroy instance.")
    stageUpdate(instance, instance.name, deleting)
    deleteGenerators(instance)
    deleteInstance(instance)
    instanceDAO.delete(instance.name)
    logger.debug(s"Instance: ${instance.name}. Instance is deleted.")
  }

  /**
    * Stopping all running generators for streams of instance,
    * if generators is not using any streams of started instances
    *
    * @param instance - Instance for stopping
    * @return - Response from marathon
    */
  def deleteGenerators(instance: Instance) = {
    logger.debug(s"Instance: ${instance.name}. Deleting generators.")
    val allStreams = instance.inputs.map(_.replaceAll("/split|/full", "")).union(instance.outputs).map(streamDAO.get)
    val startedInstances = instanceDAO.getByParameters(Map("status" -> started))
    val startingInstance = startedInstances.union(instanceDAO.getByParameters(Map("status" -> starting)))
    val tStreamStreamsToStop = allStreams
      .filter { stream: SjStream =>
        if (stream.streamType.equals(StreamConstants.tStream)) {
          val stage = instance.stages.get(stream.name)
          !stream.asInstanceOf[TStreamSjStream].generator.generatorType.equals("local") &&
            (stage.state.equals(failed) ||
              !startingInstance.exists { instance =>
                val streamGeneratorName = createGeneratorTaskName(stream.asInstanceOf[TStreamSjStream])
                val instanceStreamGenerators = instance.inputs.map(_.replaceAll("/split|/full", ""))
                  .union(instance.outputs)
                  .map(name => streamDAO.get(name))
                  .filter(s => s.streamType.equals(StreamConstants.tStream))
                  .map(sjStream => createGeneratorTaskName(sjStream.asInstanceOf[TStreamSjStream]))
                instanceStreamGenerators.contains(streamGeneratorName)
              })<commit_msg>SJ-2110 Fixed deleting of input instance<commit_after>package com.bwsw.sj.crud.rest.runner

import com.bwsw.sj.common.DAL.model.module.Instance
import com.bwsw.sj.common.DAL.model.{SjStream, TStreamSjStream}
import com.bwsw.sj.common.StreamConstants
import org.slf4j.LoggerFactory

/**
  * One-thread deleting object for instance
  * using synchronous apache http client
  * Created: 16/05/2016
  *
  * @author <NAME>
  */
class InstanceDestroyer(instance: Instance, delay: Long) extends Runnable {
  private val logger = LoggerFactory.getLogger(getClass.getName)

  import InstanceMethods._
  import com.bwsw.sj.common.ModuleConstants._

  def run() = {
    logger.debug(s"Instance: ${instance.name}. Destroy instance.")
    stageUpdate(instance, instance.name, deleting)
    deleteGenerators(instance)
    deleteInstance(instance)
    instanceDAO.delete(instance.name)
    logger.debug(s"Instance: ${instance.name}. Instance is deleted.")
  }

  /**
    * Stopping all running generators for streams of instance,
    * if generators is not using any streams of started instances
    *
    * @param instance - Instance for stopping
    * @return - Response from marathon
    */
  def deleteGenerators(instance: Instance) = {
    logger.debug(s"Instance: ${instance.name}. Deleting generators.")
    var instanceStreams = instance.outputs
    if (!instance.moduleType.equals(inputStreamingType)) {
      instanceStreams = instanceStreams.union(instance.inputs.map(_.replaceAll("/split|/full", "")))
    }
    val allStreams = instanceStreams.map(streamDAO.get)
    val startedInstances = instanceDAO.getByParameters(Map("status" -> started))
    val startingInstance = startedInstances.union(instanceDAO.getByParameters(Map("status" -> starting)))
    val tStreamStreamsToStop = allStreams
      .filter { stream: SjStream =>
        if (stream.streamType.equals(StreamConstants.tStream)) {
          val stage = instance.stages.get(stream.name)
          !stream.asInstanceOf[TStreamSjStream].generator.generatorType.equals("local") &&
            (stage.state.equals(failed) ||
              !startingInstance.exists { instance =>
                val streamGeneratorName = createGeneratorTaskName(stream.asInstanceOf[TStreamSjStream])
                val instanceStreamGenerators = instance.inputs.map(_.replaceAll("/split|/full", ""))
                  .union(instance.outputs)
                  .map(name => streamDAO.get(name))
                  .filter(s => s.streamType.equals(StreamConstants.tStream))
                  .map(sjStream => createGeneratorTaskName(sjStream.asInstanceOf[TStreamSjStream]))
                instanceStreamGenerators.contains(streamGeneratorName)
              })
{'fraction_non_alphanumeric': 0.11302681992337164, 'fraction_numerical': 0.00421455938697318, 'mean_word_length': 3.439625850340136, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, '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': '7324897', 'n_tokens_mistral': 1511, 'n_tokens_neox': 1466, 'n_words': 275}