<commit_before>/**
 * Copyright 2015 Otto (GmbH & Co KG)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.schedoscope.scheduler.actors

import scala.collection.mutable.HashMap
import org.schedoscope.SchedoscopeSettings
import org.schedoscope.dsl.View
import org.schedoscope.scheduler.messages._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.event.Logging
import akka.event.LoggingReceive
import scala.collection.mutable.HashSet
import akka.actor.OneForOneStrategy
import akka.actor.SupervisorStrategy.Escalate
import org.schedoscope.Schedoscope
import org.schedoscope.AskPattern._

/**
 * The view manager actor is the factory and import org.schedoscope.scheduler.actors.ViewActor
 * supervisor of view actors. Upon creation of view actors
 * it is responsible for creating non-existing tables or partitions in the Hive metastore, for reading
 * the last transformation timestamps and version checksums from the metastore for already materialized
 * views.
 *
 * It does this by cooperating with the parition creator actor and metadata logger actor.
 */
class ViewManagerActor(settings: SchedoscopeSettings, actionsManagerActor: ActorRef, partitionCreatorActor: ActorRef, metadataLoggerActor: ActorRef) extends Actor {
  import context._

  val log = Logging(system, ViewManagerActor.this)

  val viewStatusMap = HashMap[String, ViewStatusResponse]()

  /**
   * Supervisor strategy: Escalate any problems because view actor failures are not recoverable.
   */
  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = -1) {
      case _: Throwable => Escalate
    }

  /**
   * Message handler.
   */
  def receive = LoggingReceive({
    case vsr: ViewStatusResponse => viewStatusMap.put(sender.path.toStringWithoutAddress, vsr)

    case GetViews(views, status, filter, dependencies) => {
      val viewActors = if (views.isDefined) initializeViewActors(views.get, dependencies) else List()
      val viewStates = viewStatusMap.values
        .filter(vs => !views.isDefined || viewActors.contains(vs.actor))
        .filter(vs => !status.isDefined || status.get.equals(vs.status))
        .filter(vs => !filter.isDefined || vs.view.urlPath.matches(filter.get))
        .toList

      sender ! ViewStatusListResponse(viewStates)
    }

    case v: View => {
      sender ! initializeViewActors(List(v), false).headOption.getOrElse(List())
    }<commit_msg>Made view manager actor throw an exception and terminate the system if for some reason it is not able to return a view actor for a view<commit_after>/**
 * Copyright 2015 Otto (GmbH & Co KG)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.schedoscope.scheduler.actors

import scala.collection.mutable.HashMap
import org.schedoscope.SchedoscopeSettings
import org.schedoscope.dsl.View
import org.schedoscope.scheduler.messages._
import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.event.Logging
import akka.event.LoggingReceive
import scala.collection.mutable.HashSet
import akka.actor.OneForOneStrategy
import akka.actor.SupervisorStrategy.Escalate
import org.schedoscope.Schedoscope
import org.schedoscope.AskPattern._

/**
 * The view manager actor is the factory and import org.schedoscope.scheduler.actors.ViewActor
 * supervisor of view actors. Upon creation of view actors
 * it is responsible for creating non-existing tables or partitions in the Hive metastore, for reading
 * the last transformation timestamps and version checksums from the metastore for already materialized
 * views.
 *
 * It does this by cooperating with the parition creator actor and metadata logger actor.
 */
class ViewManagerActor(settings: SchedoscopeSettings, actionsManagerActor: ActorRef, partitionCreatorActor: ActorRef, metadataLoggerActor: ActorRef) extends Actor {
  import context._

  val log = Logging(system, ViewManagerActor.this)

  val viewStatusMap = HashMap[String, ViewStatusResponse]()

  /**
   * Supervisor strategy: Escalate any problems because view actor failures are not recoverable.
   */
  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = -1) {
      case _: Throwable => Escalate
    }

  /**
   * Message handler.
   */
  def receive = LoggingReceive({
    case vsr: ViewStatusResponse => viewStatusMap.put(sender.path.toStringWithoutAddress, vsr)

    case GetViews(views, status, filter, dependencies) => {
      val viewActors = if (views.isDefined) initializeViewActors(views.get, dependencies) else List()
      val viewStates = viewStatusMap.values
        .filter(vs => !views.isDefined || viewActors.contains(vs.actor))
        .filter(vs => !status.isDefined || status.get.equals(vs.status))
        .filter(vs => !filter.isDefined || vs.view.urlPath.matches(filter.get))
        .toList

      sender ! ViewStatusListResponse(viewStates)
    }

    case v: View => {
      sender ! initializeViewActors(List(v), false).headOption.get
    }
{'fraction_non_alphanumeric': 0.08269930532583526, 'fraction_numerical': 0.0033079722130334105, 'mean_word_length': 4.235497835497836, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7254130', 'n_tokens_mistral': 1709, 'n_tokens_neox': 1630, 'n_words': 554}