//设置package包名称以及导入依赖的类
package services.Kafka

import java.util.Properties
import java.util.concurrent.Executors

import akka.actor.{Props, DeadLetter, ActorSystem}
import kafka.consumer.{Consumer, ConsumerConfig}
import scala.concurrent.{ExecutionContext, Future}

object KafkaClient {

  val config = {
    val properties = new Properties()
    properties.put("bootstrap.servers", "localhost:9092")
    properties.put("group.id", "pogo_consumer")
    properties.put("auto.offset.reset", "largest")
    properties.put("zookeeper.connect", "localhost:2181")
    properties.put("zookeeper.session.timeout.ms", "400")
    properties.put("zookeeper.sync.time.ms", "200")
    properties.put("auto.commit.interval.ms", "500")
    new ConsumerConfig(properties)
  }

  // Our actor system managing our actors
  val system = ActorSystem("es-sharpshooter")

  // Taking care of dead letters
  system.eventStream.subscribe(system.actorOf(Props[IndexService], "dead-letters"), classOf[DeadLetter])

  // Dedicated Kafka Execution context
  implicit val KafkaContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(20))

  def start = {
    // Starting our consumer
    val consumer = Consumer.create(config)

    val topics = Map(
      "pokemons" -> 1,
      "spawnpoints" -> 1
    )

    val streams = consumer.createMessageStreams(topics)

    // Start the consumer asynchronously
    Future {
      streams.get("pokemons").get.foreach(PokemonService.cycle(system))
    } onFailure { case ec => println(ec) }
    Future {
      streams.get("spawnpoints").get.foreach(SpawnService.cycle(system))
    } onFailure { case ec => println(ec) }
  }
} 

//设置package包名称以及导入依赖的类
package akka.event

import akka.actor.Actor
import akka.actor.DeadLetter
import akka.event.Logging.Info

class DeadLetterListener extends Actor {

  val eventStream = context.system.eventStream
  val maxCount = context.system.settings.LogDeadLetters
  var count = 0

  override def preStart(): Unit =
    eventStream.subscribe(self, classOf[DeadLetter])

  // don't re-subscribe, skip call to preStart
  override def postRestart(reason: Throwable): Unit = ()

  // don't remove subscription, skip call to postStop, no children to stop
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = ()

  override def postStop(): Unit =
    eventStream.unsubscribe(self)

  def receive = {
    case DeadLetter(message, snd, rcp) ?
      count += 1
      val done = maxCount != Int.MaxValue && count >= maxCount
      val doneMsg = if (done) ", no more dead letters will be logged" else ""
      eventStream.publish(Info(rcp.path.toString, rcp.getClass,
        s"Message [${message.getClass.getName}] from $snd to $rcp was not delivered. [$count] dead letters encountered$doneMsg. " +
          "This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' " +
          "and 'akka.log-dead-letters-during-shutdown'."))
      if (done) context.stop(self)
  }

} 

//设置package包名称以及导入依赖的类
package au.csiro.data61.magda

import akka.actor.{ Actor, ActorLogging, ActorSystem, DeadLetter, Props }
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import au.csiro.data61.magda.api.SearchApi
import au.csiro.data61.magda.search.elasticsearch.{ DefaultClientProvider, ElasticSearchQueryer }

object MagdaApp extends App {
  implicit val config = AppConfig.conf()
  implicit val system = ActorSystem("search-api", config)
  implicit val executor = system.dispatcher
  implicit val materializer = ActorMaterializer()
  implicit val clientProvider = new DefaultClientProvider

  implicit val logger = Logging(system, getClass)

  logger.info("Starting API in env {} on port {}", AppConfig.getEnv, config.getString("http.port"))

  val listener = system.actorOf(Props(classOf[Listener]))
  system.eventStream.subscribe(listener, classOf[DeadLetter])

  logger.debug("Starting API")
  val searchQueryer = ElasticSearchQueryer.apply
  val api = new SearchApi(searchQueryer)

  val interface = Option(System.getenv("npm_package_config_interface")).orElse(Option(config.getString("http.interface"))).getOrElse("127.0.0.1")
  val port = Option(System.getenv("npm_package_config_port")).map(_.toInt).orElse(Option(config.getInt("http.port"))).getOrElse(6101)

  Http().bindAndHandle(api.routes, interface, port)
}

class Listener extends Actor with ActorLogging {
  def receive = {
    case d: DeadLetter => log.debug(d.message.toString())
  }
} 

//设置package包名称以及导入依赖的类
package au.csiro.data61.magda.registry

import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.ActorSystem
import akka.actor.DeadLetter
import akka.actor.Props
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import au.csiro.data61.magda.AppConfig

object RegistryApp extends App {
  class Listener extends Actor with ActorLogging {
    def receive = {
      case d: DeadLetter => //log.info(d.message)
    }
  }

  implicit val config = AppConfig.conf()
  implicit val system = ActorSystem("registry-api", config)
  implicit val executor = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val logger = Logging(system, getClass)

  logger.info("Starting MAGDA! Registry with env {}", AppConfig.getEnv)

  val listener = system.actorOf(Props(classOf[Listener]))
  system.eventStream.subscribe(listener, classOf[DeadLetter])

  val webHookActor = system.actorOf(WebHookActor.props, name = "WebHookActor")

  val api = new Api(webHookActor, config, system, executor, materializer)

  val interface = Option(System.getenv("npm_package_config_interface")).orElse(Option(config.getString("http.interface"))).getOrElse("127.0.0.1")
  val port = Option(System.getenv("npm_package_config_port")).map(_.toInt).orElse(Option(config.getInt("http.port"))).getOrElse(6100)

  Http().bindAndHandle(api.routes, interface, port)
} 

//设置package包名称以及导入依赖的类
package au.csiro.data61.magda

import akka.actor.{ Actor, ActorLogging, ActorSystem, DeadLetter, Props }
import akka.event.Logging
import akka.stream.ActorMaterializer
import au.csiro.data61.magda.api.Api
import au.csiro.data61.magda.crawler.Supervisor
import au.csiro.data61.magda.external.InterfaceConfig
import com.typesafe.config.{ ConfigObject, ConfigValue }

import scala.collection.JavaConversions._

object MagdaApp extends App {
  implicit val system = ActorSystem()
  implicit val executor = system.dispatcher
  implicit val materializer = ActorMaterializer()
  implicit val config = AppConfig.conf

  val logger = Logging(system, getClass)

  logger.info("Starting MAGDA CKAN Crawler with env {}", AppConfig.env)

  val listener = system.actorOf(Props(classOf[Listener]))
  system.eventStream.subscribe(listener, classOf[DeadLetter])

  val interfaceConfigs = config.getConfig("indexedServices").root().map {
    case (name: String, serviceConfig: ConfigValue) =>
      InterfaceConfig(serviceConfig.asInstanceOf[ConfigObject].toConfig)
  }.toSeq

  val supervisor = system.actorOf(Props(new Supervisor(system, config, interfaceConfigs)))

  // Index erryday 
  //  system.scheduler.schedule(0 millis, 1 days, supervisor, Start(List((ExternalInterfaceType.CKAN, new URL(config.getString("services.dga-api.baseUrl"))))))

  val api = new Api()
}

class Listener extends Actor with ActorLogging {
  def receive = {
    case d: DeadLetter => log.debug(d.message.toString())
  }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.newsletter.util

import akka.actor.{Actor, ActorLogging, DeadLetter, Props, UnhandledMessage}

object UnhandledMessageWatcher {
  def props() = Props(new UnhandledMessageWatcher)
}

class UnhandledMessageWatcher extends Actor with ActorLogging {

  context.system.eventStream.subscribe(self, classOf[DeadLetter])
  context.system.eventStream.subscribe(self, classOf[UnhandledMessage])

  override def receive: Receive = {
    case UnhandledMessage(msg, msgSender, recipient) =>
      log.error(s"UnhandledMessage: $msgSender sent $msg to $recipient")

    case DeadLetter(msg, msgSender, recipient) =>
      log.warning(s"DeadLetter: $msgSender sent $msg to $recipient")
  }
} 

//设置package包名称以及导入依赖的类
package actors

import akka.actor.{Actor, ActorLogging, DeadLetter, Props}

class DeadLetterHandler extends Actor with ActorLogging {
  override def receive: Receive = {
    case DeadLetter(msg, from, to) =>
      msg.getClass.getTypeName match {
        case "akka.dispatch.sysmsg.Terminate" => ()
        case _ => log.warning(s"Non-terminating dead-letter received: MSG: $msg FROM: $from TO: $to");
      }
  }
}

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

//设置package包名称以及导入依赖的类
package actors

import actors.workflow.AmazonCredentials
import akka.actor.{ActorRef, ActorSystem, DeadLetter, Props}
import com.typesafe.config.ConfigFactory
import utils.ActorFactory

object ChadashSystem {
  val config = ConfigFactory.load().getConfig("chadash")
  implicit val system = ActorSystem("ChadashSystem", config)

  val credentials       = system.actorOf(Props[AmazonCredentials], "awsCredentials")
  val userCredentials   = system.actorOf(UserCredentialsLoader.props(), "userCredentials")
  val deadLetterHandler = system.actorOf(DeadLetterHandler.props(), "deadLetterHandler")

  credentials ! AmazonCredentials.Initialize
  system.eventStream.subscribe(deadLetterHandler, classOf[DeadLetter])
}

trait DeploymentActor {
  val actor: ActorRef
}

object DeploymentActor extends DeploymentActor {
  val actor = ChadashSystem.system.actorOf(DeploymentSupervisor.props(ActorFactory), "deploymentSupervisor")
} 

//设置package包名称以及导入依赖的类
package com.taintech

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorLogging, DeadLetter, Props}


object SceneLightStart {
  def main(args: Array[String]): Unit = {
    println("hello world")
    akka.Main.main(Array(classOf[Master].getName))
  }
}

class Master extends Actor {
  override def preStart(): Unit = {
    val slave = context.actorOf(Props[Slave], "slave")
    slave ! Work
    val noone = context.actorOf(Props[DeadLetterReader], "noone")
    noone ! Unknown
  }
  def receive = {
    case Done => context.stop(self)
  }
}

class Slave extends Actor with ActorLogging {
  override def receive = {
    case Work =>
      log.info("I hate my Master!")
      sender() ! Done
  }
}

case object Work

case object Done

case object Unknown

class DeadLetterReader extends Actor with ActorLogging{
  override def preStart(): Unit = context.system.eventStream.subscribe(self, classOf[DeadLetter])
  override def receive: Receive = {
    case Unknown => log.info("Received unknown message")
  }
} 

//设置package包名称以及导入依赖的类
package services.Kafka

import akka.actor.{DeadLetter, Actor}
import models.Pokemon
import com.github.nscala_time.time.Imports._
import services.ES.{IndexInto, DeleteFrom}

class IndexService extends Actor {

  import context._

  def indexWithTTL(pokemon: Pokemon) = {
    // Index pokemon into ES and schedule another message containing the ID
    IndexInto.pokemon(pokemon).map { id =>
      val duration = (DateTime.now to pokemon.expireAt).toDuration.toScalaDuration

      system.scheduler.scheduleOnce(duration) {
        self ! id
      }
    }
  }

  def deleteAfterExpired(id: String) = {
    DeleteFrom.pokemon(id)
  }

  override def receive: Receive = {
    case pk: Pokemon               => indexWithTTL(pk)
    case id: String                => deleteAfterExpired(id)
    case DeadLetter(msg, from, to) => deleteAfterExpired(msg.asInstanceOf[String])
  }
} 

//设置package包名称以及导入依赖的类
package akka.event

import akka.actor.Actor
import akka.actor.DeadLetter
import akka.event.Logging.Info

class DeadLetterListener extends Actor {

  val eventStream: EventStream = context.system.eventStream
  val maxCount = context.system.settings.LogDeadLetters
  var count = 0

  override def preStart(): Unit =
    eventStream.subscribe(self, classOf[DeadLetter])

  // don't re-subscribe, skip call to preStart
  override def postRestart(reason: Throwable): Unit = ()

  // don't remove subscription, skip call to postStop, no children to stop
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = ()

  override def postStop(): Unit =
    eventStream.unsubscribe(self)

  def receive: PartialFunction[Any, Unit] = {
    case DeadLetter(message, snd, rcp) ?
      count += 1
      val done = maxCount != Int.MaxValue && count >= maxCount
      val doneMsg = if (done) ", no more dead letters will be logged" else ""
      eventStream.publish(Info(rcp.path.toString, rcp.getClass,
        s"Message [${message.getClass.getName}] from $snd to $rcp was not delivered. [$count] dead letters encountered$doneMsg. " +
          "This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' " +
          "and 'akka.log-dead-letters-during-shutdown'."))
      if (done) context.stop(self)
  }

} 

//设置package包名称以及导入依赖的类
package com.github.sorhus.webalytics.akka.util

import akka.actor.{Actor, DeadLetter, Props}
import org.slf4j.LoggerFactory

class DeadLetterLoggingActor extends Actor {

  val log = LoggerFactory.getLogger(getClass)

  var count = 0
  def receive = {
    case DeadLetter(msg, from, to) =>
      count += 1
      log.warn(s"received dead letter no $count, {}", (msg, from, to))
  }
}

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

//设置package包名称以及导入依赖的类
package nl.bridgeworks.akka.rete

import akka.actor.{DeadLetter, Props, ActorSystem}

object ConsoleApp {
  def main(args: Array[String]) {
    val system = ActorSystem("Shakespeare")
    val listener = system.actorOf(Props(new DeadListener))
    system.eventStream.subscribe(listener, classOf[DeadLetter])

    val r1 = Rule("1", Vector[Expr](Simple("nasal congestion"), Simple("viremia")), Vector[Fact](ConceptOnly("diagnosis influenza")))
    val r2 = Rule("2", Vector[Expr](Simple("runny nose")), Vector[Fact](ConceptOnly("nasal congestion")))
    val r3 = Rule("3", Vector[Expr](Simple("body aches")), Vector[Fact](ConceptOnly("achiness")))
    val r4 = Rule("4", Vector[Expr](Simple("headache")), Vector[Fact](ConceptOnly("achiness")))
    val r5 = Rule("5", Vector[Expr](ValueOp("temp", GreaterThan, 100)), Vector[Fact](ConceptOnly("fever")))
    val r6 = Rule("6", Vector[Expr](Simple("fever"), Simple("achiness")), Vector[Fact](ConceptOnly("viremia")))

    val r7 = Rule("7", Vector[Expr](Simple("Single with few financial burdens")), Vector[Fact](ConceptOnly("5 points")))
    val r8 = Rule("8", Vector[Expr](Simple("A couple with children")), Vector[Fact](ConceptOnly("3 points")))
    val r9 = Rule("9", Vector[Expr](Simple("Young family with a home")), Vector[Fact](ConceptOnly("1 points")))
    val r10 = Rule("10", Vector[Expr](Simple("Mature family")), Vector[Fact](ConceptOnly("5 points")))
    val r11 = Rule("11", Vector[Expr](Simple("Preparing for retirement")), Vector[Fact](ConceptOnly("3 points")))
    val r12 = Rule("12", Vector[Expr](Simple("Retired")), Vector[Fact](ConceptOnly("1 points")))

    val rules = Vector[Rule](r1, r2, r3, r4, r5, r6)

    val cs = buildReteNetwork(rules, system)
    cs ! Assertion(Vector(ConceptOnly("runny nose"), ConceptWithValue("temp", 101), ConceptOnly("headache")), java.util.UUID.randomUUID.toString)

    Thread.sleep(50)

    cs ! "print"

    system.terminate
  }
}