<commit_before>package com.stratio.connectors

import akka.actor.{Actor, ActorLogging, Props}
import akka.cluster.ClusterEvent._
import com.stratio.meta.common.connector.IConnector
import com.stratio.meta.communication.{HeartbeatSig, getConnectorName, replyConnectorName}
import com.stratio.meta2.common.data.ClusterName
import com.stratio.meta2.core.query.{MetadataInProgressQuery, SelectInProgressQuery, StorageInProgressQuery}
import com.stratio.meta2.core.statements.{MetaDataStatement, SelectStatement}


object ConnectorActor{
  def props (connectorName:String,connector:IConnector):Props = Props (new ConnectorActor(connectorName,connector) )
}

class ConnectorActor(connectorName:String,conn:IConnector) extends Actor with ActorLogging with HeartbeatActor{

  val connector=conn //TODO: test if it works with one thread and multiple threads

  //val cluster = Cluster(context.system)
  //import cluster.{ scheduler }
  //val heartbeatTask = scheduler.schedule(PeriodicTasksInitialDelay max HeartbeatInterval, HeartbeatInterval, self, HeartbeatTick)


  // subscribe to cluster changes, re-subscribe when restart

  override def handleHeartbeat(heartbeat:HeartbeatSig)={
    println("ConnectorActor receives a heartbeat message")
  }

  override def receive = super.receive orElse{

    case stop=>{
      connector.close(new ClusterName(""))
    }
    case inProgressQuery:MetadataInProgressQuery=>{
      log.info("->"+"Receiving MetadataInProgressQuery")
      //val statement:MetaDataStatement=null
      val statement=inProgressQuery.getStatement()
      statement match{
        case ms:MetaDataStatement =>
          log.info("->receiving MetadataStatement")
        case _ =>
          log.info("->receiving a statement of a type it shouldn't")
      }
    }

     case inProgressQuery:SelectInProgressQuery=>{
      log.info("->"+"Receiving SelectInProgressQuery")
      //val statement:MetaDataStatement=null
      val statement=inProgressQuery.getStatement()
      statement match{
        case ms:SelectStatement =>
          log.info("->receiving SelectStatement")
          //val catalog=ms.g
          //val catalogs=inProgressQuery.getCatalogs()
          connector.getQueryEngine().execute()
        case _ =>
          log.info("->receiving a statement of a type it shouldn't")
      }
     }

    case inProgressQuery:StorageInProgressQuery=>
      log.info("->"+"Receiving StorageInProgressQuery")



    /*
    case metadataEngineRequest:MetadataEngineRequest=>
      //getMetadataEngine()
      log.info("->"+"Receiving MetadataRequest")

    case dataStoreNameRequest:DataStoreNameRequest=>
      //getDatastoreName()<commit_msg>adding new 'abstract' method shutdown, that will be overriden by connector implementations<commit_after>package com.stratio.connectors

import akka.actor.{Actor, ActorLogging, Props}
import akka.cluster.ClusterEvent._
import com.stratio.meta.common.connector.IConnector
import com.stratio.meta.communication.{HeartbeatSig, getConnectorName, replyConnectorName}
import com.stratio.meta2.core.query.{MetadataInProgressQuery, SelectInProgressQuery, StorageInProgressQuery}
import com.stratio.meta2.core.statements.{MetaDataStatement, SelectStatement}


object ConnectorActor{
  def props (connectorName:String,connector:IConnector):Props = Props (new ConnectorActor(connectorName,connector) )
}

class ConnectorActor(connectorName:String,conn:IConnector) extends Actor with ActorLogging with HeartbeatActor{

  val connector=conn //TODO: test if it works with one thread and multiple threads

  //val cluster = Cluster(context.system)
  //import cluster.{ scheduler }
  //val heartbeatTask = scheduler.schedule(PeriodicTasksInitialDelay max HeartbeatInterval, HeartbeatInterval, self, HeartbeatTick)


  // subscribe to cluster changes, re-subscribe when restart

  override def handleHeartbeat(heartbeat:HeartbeatSig)={
    println("ConnectorActor receives a heartbeat message")
  }

  def shutdown()={
    println("ConnectorActor is shutting down")
    //connector.close(new ClusterName(""))
    //connector.shutdown(new ClusterName(""))
  }

  override def receive = super.receive orElse{

    case shutdown=>{
      this.shutdown()
    }
    case inProgressQuery:MetadataInProgressQuery=>{
      log.info("->"+"Receiving MetadataInProgressQuery")
      //val statement:MetaDataStatement=null
      val statement=inProgressQuery.getStatement()
      statement match{
        case ms:MetaDataStatement =>
          log.info("->receiving MetadataStatement")
        case _ =>
          log.info("->receiving a statement of a type it shouldn't")
      }
    }

     case inProgressQuery:SelectInProgressQuery=>{
      log.info("->"+"Receiving SelectInProgressQuery")
      //val statement:MetaDataStatement=null
      val statement=inProgressQuery.getStatement()
      statement match{
        case ms:SelectStatement =>
          log.info("->receiving SelectStatement")
          //val catalog=ms.g
          //val catalogs=inProgressQuery.getCatalogs()
          //connector.getQueryEngine().execute()
        case _ =>
          log.info("->receiving a statement of a type it shouldn't")
      }
     }

    case inProgressQuery:StorageInProgressQuery=>
      log.info("->"+"Receiving StorageInProgressQuery")



    /*
    case metadataEngineRequest:MetadataEngineRequest=>
      //getMetadataEngine()
      log.info("->"+"Receiving MetadataRequest")

    case dataStoreNameRequest:DataStoreNameRequest=>
      //getDatastoreName()
{'fraction_non_alphanumeric': 0.10107448552176289, 'fraction_numerical': 0.0009105809506465125, 'mean_word_length': 4.010948905109489, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 39, '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': '3142369', 'n_tokens_mistral': 1533, 'n_tokens_neox': 1460, 'n_words': 323}