<commit_before>package com.stratio.meta2.server.actors

import akka.actor.{Actor, ActorLogging, ActorRef, Props, ReceiveTimeout, RootActorPath}
import akka.cluster.ClusterEvent._
import com.stratio.meta.communication._
import com.stratio.meta2.core.query.InProgressQuery
import akka.cluster.Cluster

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

class ConnectorManagerActor extends Actor with ActorLogging {

  log.info("Lifting connector actor")
  val coordinatorActorRef = context.actorSelection("../CoordinatorActor")
  //coordinatorActorRef ! "hola"

  var connectorsMap: Map[String, ActorRef] = Map()

  override def preStart(): Unit = {
    //#subscribe
    Cluster(context.system).subscribe(self, classOf[MemberEvent])
    //cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent], classOf[UnreachableMember])
  }
  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {

    case mu: MemberUp => {
      log.info("Member is Up: {}" + mu.toString+mu.member.getRoles)
      val it=mu.member.getRoles.iterator()
      while(it.hasNext()){
    	  var rol=it.next()
    	  rol match{
    	    case "connector"=>
    	    	val connectorActorRef = context.actorSelection(RootActorPath(mu.member.address) / "user" / "meta-connector")
    	    	connectorActorRef ! "hola"
    	  }
    	  log.info("has role: {}" + rol)
      }
      // connectorsMap += (member.toString -> memberActorRef)
      //memberActorRef ! "hola pichi, estás metaregistrado"
    }


    case query: InProgressQuery => {
      log.info("Connector Actor received InProgressQuery")
    }


    case state: CurrentClusterState =>
      log.info("Current members: {}", state.members.mkString(", "))

    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)

    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)

    case _: MemberEvent =>
      log.info("Receiving anything else")

    case _: ClusterDomainEvent =>
      println("ClusterDomainEvent")

    case ReceiveTimeout =>
      println("ReceiveTimeout")
      
    case _:ConnectToConnector=>
      println("connecting to connector ")

    case _: DisconnectFromConnector=>
      println("disconnecting from connector")

      /*
    case ex:Execute=>
      println("Executing workflow "+ex.workflow.toString())
      println("initial steps->"+ex.workflow.getInitialSteps())
     */
    //case _: Request=>
    //case _: Response=>
    //case _: MetadataStruct=>
    //case _: StorageQueryStruct=>
    //case _: WorkflowStruct=>

    case toConnector: MetadataStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case toConnector: StorageQueryStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case toConnector: WorkflowStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case response: Response =>
      connectorsMap += (response.msg -> sender)

    case other =>
      println("connector actor receives event")
    //      sender ! "OK"
    //memberActorRef.tell(objetoConWorkflow, context.sender)
  }

}<commit_msg>the connectorManager saves the connector references in a Hash<commit_after>package com.stratio.meta2.server.actors

import akka.actor.{Actor, ActorLogging, ActorRef, Props, ReceiveTimeout, RootActorPath}
import akka.cluster.ClusterEvent._
import com.stratio.meta.communication._
import com.stratio.meta2.core.query.InProgressQuery
import akka.cluster.Cluster
import java.util.HashMap
import akka.actor.ActorSelection

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

class ConnectorManagerActor extends Actor with ActorLogging {

  log.info("Lifting connector actor")
  val coordinatorActorRef = context.actorSelection("../CoordinatorActor")
  //coordinatorActorRef ! "hola"

  //var connectorsMap:scala.collection.mutable.Map[String, ActorRef] = scala.collection.mutable.Map[String,ActorRef]()
  var connectorsMap:scala.collection.mutable.Map[String, ActorSelection] = scala.collection.mutable.Map[String,ActorSelection]()

  override def preStart(): Unit = {
    //#subscribe
    Cluster(context.system).subscribe(self, classOf[MemberEvent])
    //cluster.subscribe(self, initialStateMode = InitialStateAsEvents, classOf[MemberEvent], classOf[UnreachableMember])
  }
  override def postStop(): Unit =
    Cluster(context.system).unsubscribe(self)

  def receive = {

    case mu: MemberUp => {
      log.info("Member is Up: {}" + mu.toString+mu.member.getRoles)
      val it=mu.member.getRoles.iterator()
      while(it.hasNext()){
    	  var rol=it.next()
    	  rol match{
    	    case "connector"=>
    	    	val connectorActorRef = context.actorSelection(RootActorPath(mu.member.address) / "user" / "meta-connector")
    	    	val id=java.util.UUID.randomUUID.toString()
    	    	connectorsMap.put(mu.member.toString, connectorActorRef)
    	    	//connectorActorRef ! "hola"
    	  }
    	  log.info("has role: {}" + rol)
      }
      // connectorsMap += (member.toString -> memberActorRef)
      //memberActorRef ! "h<NAME>, estás metaregistrado"
    }


    case query: InProgressQuery => {
      log.info("Connector Actor received InProgressQuery")
    }


    case state: CurrentClusterState =>
      log.info("Current members: {}", state.members.mkString(", "))

    case UnreachableMember(member) =>
      log.info("Member detected as unreachable: {}", member)

    case MemberRemoved(member, previousStatus) =>
      log.info("Member is Removed: {} after {}",
        member.address, previousStatus)

    case _: MemberEvent =>
      log.info("Receiving anything else")

    case _: ClusterDomainEvent =>
      println("ClusterDomainEvent")

    case ReceiveTimeout =>
      println("ReceiveTimeout")
      
    case _:ConnectToConnector=>
      println("connecting to connector ")

    case _: DisconnectFromConnector=>
      println("disconnecting from connector")

      /*
    case ex:Execute=>
      println("Executing workflow "+ex.workflow.toString())
      println("initial steps->"+ex.workflow.getInitialSteps())
     */
    //case _: Request=>
    //case _: Response=>
    //case _: MetadataStruct=>
    //case _: StorageQueryStruct=>
    //case _: WorkflowStruct=>

    case toConnector: MetadataStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case toConnector: StorageQueryStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case toConnector: WorkflowStruct =>
      connectorsMap(toConnector.connectorName) ! toConnector

    case response: Response =>
      //connectorsMap += (response.msg -> sender)

    case other =>
      println("connector actor receives event")
    //      sender ! "OK"
    //memberActorRef.tell(objetoConWorkflow, context.sender)
  }

}
{'fraction_non_alphanumeric': 0.1191881387649345, 'fraction_numerical': 0.000575788109975529, 'mean_word_length': 3.491273432449903, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 54, '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': '1120502', 'n_tokens_mistral': 2089, 'n_tokens_neox': 2019, 'n_words': 406}