<commit_before>/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) licenses this file
 * to you 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 com.stratio.crossdata.server.actors

import akka.actor._
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import com.stratio.crossdata.common.connector.ConnectorClusterConfig
import com.stratio.crossdata.common.data
import com.stratio.crossdata.common.data.ConnectorName
import com.stratio.crossdata.common.result.ConnectResult
import com.stratio.crossdata.common.statements.structures.selectors.SelectorHelper
import com.stratio.crossdata.common.utils.StringUtils
import com.stratio.crossdata.communication._
import com.stratio.crossdata.core.connector.ConnectorManager
import com.stratio.crossdata.core.execution.ExecutionManager
import com.stratio.crossdata.core.metadata.MetadataManager
import org.apache.log4j.Logger

import scala.collection.JavaConversions._

object ConnectorManagerActor {
  def props(connectorManager: ConnectorManager): Props = Props(new ConnectorManagerActor(connectorManager))
}

class ConnectorManagerActor(connectorManager: ConnectorManager) extends Actor with ActorLogging {

  lazy val logger = Logger.getLogger(classOf[ConnectorManagerActor])
  logger.info("Lifting connectormanager manager actor")
  val coordinatorActorRef = context.actorSelection("../CoordinatorActor")

  val connectors = MetadataManager.MANAGER.getConnectorNames(data.Status.ONLINE)
  MetadataManager.MANAGER.setConnectorStatus(connectors, data.Status.OFFLINE)

  override def preStart(): Unit = {
    Cluster(context.system).subscribe(self, classOf[ClusterDomainEvent])
  }

  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {

    /**
     * A new actor connects to the cluster. If the new actor is a connectormanager, we requests its name.
     */
    //TODO Check that new actors are recognized and their information stored in the MetadataManager
    case mu: MemberUp => {
      logger.info("Member is Up: " + mu.toString + mu.member.getRoles)
      val it = mu.member.getRoles.iterator()
      while (it.hasNext()) {
        val rol = it.next()
        rol match {
          case "connectormanager" => {
            val connectorActorRef = context.actorSelection(RootActorPath(mu.member.address) / "user" / "ConnectorActor")
            connectorActorRef ! getConnectorName()
          }
          case _ =>{
            logger.info("MemberUp: rol is not in this actor.")

          }
        }
      }
    }

    /**
     * CONNECTOR answers its name.
     */
    case msg: replyConnectorName => {
      logger.info("Connector Name " + msg.name + " received from " + sender)
      val actorRefUri = StringUtils.getAkkaActorRefUri(sender)
      logger.info("Registering connectormanager at: " + actorRefUri)
      val connectorName = new ConnectorName(msg.name)
      ExecutionManager.MANAGER.createEntry(actorRefUri, connectorName, true)

      MetadataManager.MANAGER.addConnectorRef(connectorName, actorRefUri)

      val connectorMetadata = MetadataManager.MANAGER.getConnector(connectorName)
      val clusterProps = connectorMetadata.getClusterProperties

      if((clusterProps != null) && (!clusterProps.isEmpty)){
        for(clusterProp <- clusterProps.entrySet()){
          val opts = MetadataManager.MANAGER.getCluster(clusterProp.getKey).getOptions;
          val connectorClusterConfig = new ConnectorClusterConfig(clusterProp.getKey,
            SelectorHelper.convertSelectorMapToStringMap(opts))
          sender ! new Connect(null, connectorClusterConfig)
        }
      }
      MetadataManager.MANAGER.setConnectorStatus(connectorName, data.Status.ONLINE)

    }

    case c: ConnectResult => {
      logger.info("Connect result from " + sender + " => " + c.getSessionId)
    }

    //Pass the message to the connectorActor to extract the member in the cluster
    case state: CurrentClusterState => {
      logger.info("Current members: " + state.members.mkString(", "))
      //TODO Check if I'm the only server
      try {
        val connectors = MetadataManager.MANAGER.getConnectorNames(data.Status.ONLINE)
        MetadataManager.MANAGER.setConnectorStatus(connectors, data.Status.OFFLINE)
      } catch {
        case e: Exception => {
          log.error("Couldn't set connectors to OFFLINE")
        }
      }
    }

    case member: UnreachableMember => {
      logger.info("Member detected as unreachable: " + member)
      //TODO Process UnreachableMember
    }

    case member: MemberRemoved => {
      logger.info("Member is Removed: " + member.member.address)
      logger.info("Member info: " + member.toString)
      val actorRefUri = StringUtils.getAkkaActorRefUri(member.member.address)
      val connectorName = ExecutionManager.MANAGER.getValue(actorRefUri +"/user/ConnectorActor/")
      MetadataManager.MANAGER.setConnectorStatus(connectorName.asInstanceOf[ConnectorName],
        data.Status.OFFLINE)
    }

    case member: MemberExited => {
      logger.info("Member is exiting: " + member.member.address)
      val actorRefUri = StringUtils.getAkkaActorRefUri(sender)
      val connectorName = ExecutionManager.MANAGER.getValue(actorRefUri)
      MetadataManager.MANAGER.setConnectorStatus(connectorName.asInstanceOf[ConnectorName],
        data.Status.SHUTTING_DOWN)<commit_msg>Connectors status restart system improved<commit_after>/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) licenses this file
 * to you 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 com.stratio.crossdata.server.actors

import akka.actor._
import akka.cluster.Cluster
import akka.cluster.ClusterEvent._
import com.stratio.crossdata.common.connector.ConnectorClusterConfig
import com.stratio.crossdata.common.data
import com.stratio.crossdata.common.data.ConnectorName
import com.stratio.crossdata.common.result.ConnectResult
import com.stratio.crossdata.common.statements.structures.selectors.SelectorHelper
import com.stratio.crossdata.common.utils.StringUtils
import com.stratio.crossdata.communication._
import com.stratio.crossdata.core.connector.ConnectorManager
import com.stratio.crossdata.core.execution.ExecutionManager
import com.stratio.crossdata.core.metadata.MetadataManager
import org.apache.log4j.Logger

import scala.collection.JavaConversions._
import scala.collection.mutable

object ConnectorManagerActor {
  def props(connectorManager: ConnectorManager): Props = Props(new ConnectorManagerActor(connectorManager))
}

class ConnectorManagerActor(connectorManager: ConnectorManager) extends Actor with ActorLogging {

  lazy val logger = Logger.getLogger(classOf[ConnectorManagerActor])
  logger.info("Lifting connectormanager manager actor")
  val coordinatorActorRef = context.actorSelection("../CoordinatorActor")

  val connectors = MetadataManager.MANAGER.getConnectorNames(data.Status.ONLINE)
  MetadataManager.MANAGER.setConnectorStatus(connectors, data.Status.OFFLINE)

  override def preStart(): Unit = {
    Cluster(context.system).subscribe(self, classOf[ClusterDomainEvent])
  }

  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {

    /**
     * A new actor connects to the cluster. If the new actor is a connectormanager, we requests its name.
     */
    //TODO Check that new actors are recognized and their information stored in the MetadataManager
    case mu: MemberUp => {
      logger.info("Member is Up: " + mu.toString + mu.member.getRoles)
      val it = mu.member.getRoles.iterator()
      while (it.hasNext()) {
        val rol = it.next()
        rol match {
          case "connector" => {
            val connectorActorRef = context.actorSelection(RootActorPath(mu.member.address) / "user" / "ConnectorActor")
            connectorActorRef ! getConnectorName()
          }
          case _ =>{
            logger.info("MemberUp: rol is not in this actor.")

          }
        }
      }
    }

    /**
     * CONNECTOR answers its name.
     */
    case msg: replyConnectorName => {
      logger.info("Connector Name " + msg.name + " received from " + sender)
      val actorRefUri = StringUtils.getAkkaActorRefUri(sender)
      logger.info("Registering connector at: " + actorRefUri)
      val connectorName = new ConnectorName(msg.name)
      ExecutionManager.MANAGER.createEntry(actorRefUri, connectorName, true)

      MetadataManager.MANAGER.addConnectorRef(connectorName, actorRefUri)

      val connectorMetadata = MetadataManager.MANAGER.getConnector(connectorName)
      val clusterProps = connectorMetadata.getClusterProperties

      if((clusterProps != null) && (!clusterProps.isEmpty)){
        for(clusterProp <- clusterProps.entrySet()){
          val opts = MetadataManager.MANAGER.getCluster(clusterProp.getKey).getOptions;
          val connectorClusterConfig = new ConnectorClusterConfig(clusterProp.getKey,
            SelectorHelper.convertSelectorMapToStringMap(opts))
          sender ! new Connect(null, connectorClusterConfig)
        }
      }
      MetadataManager.MANAGER.setConnectorStatus(connectorName, data.Status.ONLINE)

    }

    case c: ConnectResult => {
      logger.info("Connect result from " + sender + " => " + c.getSessionId)
    }

    //Pass the message to the connectorActor to extract the member in the cluster
    case state: CurrentClusterState => {
      logger.info("Current members: " + state.members.mkString(", "))
      var foundServers = mutable.HashSet[Address]()
      val members = state.getMembers
      for(member <- members){
        if(member.getRoles.contains("server")){
          foundServers += member.address
        }
      }
      if(foundServers.size == 1){
        val connectors = MetadataManager.MANAGER.getConnectorNames(data.Status.ONLINE)
        MetadataManager.MANAGER.setConnectorStatus(connectors, data.Status.OFFLINE)
      }
    }

    case member: UnreachableMember => {
      logger.info("Member detected as unreachable: " + member)
      //TODO Process UnreachableMember
    }

    case member: MemberRemoved => {
      logger.info("Member is Removed: " + member.member.address)
      logger.info("Member info: " + member.toString)
      val actorRefUri = StringUtils.getAkkaActorRefUri(member.member.address)
      val connectorName = ExecutionManager.MANAGER.getValue(actorRefUri +"/user/ConnectorActor/")
      MetadataManager.MANAGER.setConnectorStatus(connectorName.asInstanceOf[ConnectorName],
        data.Status.OFFLINE)
    }

    case member: MemberExited => {
      logger.info("Member is exiting: " + member.member.address)
      val actorRefUri = StringUtils.getAkkaActorRefUri(sender)
      val connectorName = ExecutionManager.MANAGER.getValue(actorRefUri)
      MetadataManager.MANAGER.setConnectorStatus(connectorName.asInstanceOf[ConnectorName],
        data.Status.SHUTTING_DOWN)
{'fraction_non_alphanumeric': 0.09143984869665324, 'fraction_numerical': 0.000904530877394951, 'mean_word_length': 3.6633435582822087, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 24, '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': '4235351', 'n_tokens_mistral': 3366, 'n_tokens_neox': 3285, 'n_words': 805}