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

import actors.UserSocket.{ChatMessage, Message}
import actors.UserSocket.Message.messageReads
import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.{Publish, Subscribe}
import akka.event.LoggingReceive
import play.api.libs.json.{Writes, JsValue, Json}
import play.twirl.api.HtmlFormat

import scala.xml.Utility

object UserSocket {
  def props(user: String)(out: ActorRef) = Props(new UserSocket(user, out))

  case class Message(msg: String)

  object Message {
    implicit val messageReads = Json.reads[Message]
  }

  case class ChatMessage(user: String, text: String)

  object ChatMessage {
    implicit val chatMessageWrites = new Writes[ChatMessage] {
      def writes(chatMessage: ChatMessage): JsValue = {
        Json.obj(
          "type" -> "message",
          "user" -> chatMessage.user,
          "text" -> multiLine(chatMessage.text)
        )
      }
    }

    private def multiLine(text: String) = {
      HtmlFormat.raw(text).body.replace("\n", "<br/>")
    }
  }
}

class UserSocket(uid: String, out: ActorRef) extends Actor with ActorLogging {

  val topic = "chat"

  val mediator = DistributedPubSub(context.system).mediator

  mediator ! Subscribe(topic, self)

  def receive = LoggingReceive {
    case js: JsValue =>
      js.validate[Message](messageReads)
        .map(message => Utility.escape(message.msg))
        .foreach { msg => mediator ! Publish(topic, ChatMessage(uid, msg))}

    case c:ChatMessage => out ! Json.toJson(c)
  }
} 

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

import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
import akka.cluster.pubsub.DistributedPubSub
import com.google.inject.Inject
import common.BarEvents
import common.BarEvents.EntityInserted
import play.api.libs.json.Json
import play.api.mvc._

import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import akka.pattern._
import akka.util.Timeout

import scala.concurrent.ExecutionContext
import scala.concurrent.duration._

class MainController @Inject()(system: ActorSystem)(implicit ec: ExecutionContext)
    extends Controller {

  val entities  = system.actorOf(Props(new EntitiesService))
  val random    = new Random()
  implicit val timeout = Timeout(5.seconds)

  def list() = Action.async {
    val currently = (entities ? "get").mapTo[Seq[Int]]
    currently.map(x => Ok(Json.toJson(x)))
  }

  def insert() = Action {
    val next = random.nextInt
    entities ! next
    Created(Json.toJson(random.nextInt))
  }
}

class EntitiesService extends Actor with ActorLogging {
  import akka.cluster.pubsub.DistributedPubSubMediator.Publish
  val mediator = DistributedPubSub(context.system).mediator
  val topic = BarEvents.KEY

  val existingEntities = ArrayBuffer[Int]()

  def receive = {
    case "get" => sender ! existingEntities
    case next: Int =>
      mediator ! Publish(topic, existingEntities)
      existingEntities.append(next)
      log.warning(s"publishing $existingEntities")
  }
} 

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

import akka.actor.Actor
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish
import org.a4r.Logged
import org.a4r.messages.MessageTopic


trait DistributedPublisher extends Actor with Logged{
  // activate the extension
  def topics:Seq[MessageTopic]
  val mediator = DistributedPubSub(context.system).mediator

  def publish( msg:Any ) = {
    topics.map(topic => {
      info(s"publishing $msg to ${topic.name}")
      mediator ! Publish(topic.name, msg)
    })
  }


} 

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

import akka.actor.{Props, ActorLogging, Actor}
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish
import akka.cluster.sharding.ShardRegion

// Domain
final case class PublisherUpdate(publisher: UpdateableEvent)

object PublisherActor {

  lazy val props = Props(classOf[PublisherActor])

  val idExtractor: ShardRegion.ExtractEntityId = {
    case s: PublisherUpdate => (s.publisher.id.toString, s.publisher)
  }

  val shardExtractor: ShardRegion.ExtractShardId = {
    case msg: PublisherUpdate => (msg.publisher.id % 100).toString
  }
}

class PublisherActor extends Actor with ActorLogging {

  log.info(s"Publisher ${self.path.name} actor instance created")

  val mediator = DistributedPubSub(context.system).mediator

  override def receive: Receive = {
    case event: UpdateableEvent => {
      log.info(s"$event to be routed to all listeners")
      mediator ! Publish(s"Publisher-${self.path.name}", event)
    }
  }
} 

//设置package包名称以及导入依赖的类
package tmt.demo.connectors

import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish
import com.trueaccord.scalapb.GeneratedMessageCompanion
import tmt.app.utils.{ActorRuntime, PbMessage}
import tmt.demo.zeromq_drivers.ZmqSubscriberFactory

class ZmqToAkkaFlow(actorRuntime: ActorRuntime, zmqSubscriberFactory: ZmqSubscriberFactory) {
  import actorRuntime._

  def connect[Msg <: PbMessage.Of[Msg]](
    publishingTopic: String,
    subscriberPort: Int,
    responseParser: GeneratedMessageCompanion[Msg]
  ) = {
    val zmqSubscriber = zmqSubscriberFactory.make(subscriberPort, responseParser)
    zmqSubscriber.stream
      .runForeach { message =>
        DistributedPubSub(system).mediator ! Publish(publishingTopic, message)
      }.onComplete { x =>
        zmqSubscriber.shutdown()
      }
  }
} 

//设置package包名称以及导入依赖的类
package tmt.demo.hcd_drivers

import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish
import akka.stream.scaladsl.Source
import com.trueaccord.scalapb.GeneratedMessage
import tmt.app.utils.ActorRuntime

class EventPublisher(actorRuntime: ActorRuntime) {

  import actorRuntime._

  def publish[Msg <: GeneratedMessage](
    messages: Source[Msg, Any],
    topic: String
  ) = {
    messages.runForeach { message =>
      println(s"********* PublisherClient is publishing: $message")
      DistributedPubSub(system).mediator ! Publish(topic, message)
    }
  }
} 

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

import akka.actor.{ Actor, ActorLogging, ActorRef, ActorSystem, Props }
import common.CommonMessages.{ ProcessLine, Received, fileProcessingTopic }
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish

object FileReceiver {
  def props(): Props = Props(new FileReceiver)
  def createFileReceiver(system: ActorSystem): ActorRef = {
    system.actorOf(FileReceiver.props())
  }
}

class FileReceiver extends Actor with ActorLogging {

  val mediator = DistributedPubSub(context.system).mediator

  override def receive: Receive = {
    case msg: ProcessLine =>
      mediator ! Publish(fileProcessingTopic, msg)
      sender() ! Received
  }
} 

//设置package包名称以及导入依赖的类
package com.github.mkorman9.districron.logic

import akka.actor.{ActorRef, ActorSystem}
import akka.cluster.pubsub.DistributedPubSubMediator.{Publish, Send}
import com.github.mkorman9.districron.Messages.ExecuteTask
import com.github.mkorman9.districron.model.Task
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

@Service
class MessageSender @Autowired() (actorSystem: ActorSystem) {
  def sendMessageToTasksHandlingActor(message: Any): Unit = {
    val tasksHandlingActor = actorSystem.actorSelection(com.github.mkorman9.districron.Actor.Path)
    tasksHandlingActor ! message
  }

  def broadcastExecutionMessage(clusterMediator: ActorRef, sender: ActorRef, executionId: Long, task: Task): Unit = {
    val executeTaskMessage = ExecuteTask(executionId, task.name)

    if (task.globalDistribution)
      clusterMediator.tell(
        Publish(task.recipient, executeTaskMessage),
        sender
      )
    else
      clusterMediator.tell(
        Send(task.recipient, executeTaskMessage, localAffinity = false),
        sender
      )
  }
} 

//设置package包名称以及导入依赖的类
package org.dmonix.area51.akka.cluster.extensions

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
import akka.cluster.Cluster
import akka.cluster.pubsub.DistributedPubSub
import akka.cluster.pubsub.DistributedPubSubMediator.Publish
import org.dmonix.area51.akka.cluster.Messages.{Broadcast, Message, Response, Unicast}
import org.dmonix.area51.akka.cluster.{ClusterSettings, SameThreadExecutionContext}


class Publisher(mediator:ActorRef) extends Actor with ActorLogging{

  def receive = {
    //sends the provided message to ONE instance of all the provided service names
    case Unicast(serviceNames, message) =>
      serviceNames.foreach(sName => mediator ! Publish(sName,message,true))
    //sends the provided message to ALL instances of all the provided service names
    case Broadcast(serviceNames, message) =>
      serviceNames.foreach(sName => mediator ! Publish(sName,message,false))
    case Response(rsp) =>
      log.info(s"Publisher [$self] got response [$rsp] from [$sender]")
    case a:Any ?
      log.warning(s"Publisher [$self] got unexpected message [$a] from [$sender]")
  }

}

object PublisherStarter extends App with SameThreadExecutionContext with ClusterSettings {
  System.setProperty("config.file", "src/main/resources/akka-cfg/cluster-ext-member-tcp.conf");
  val actorSystem = ActorSystem(actorSystemName)
  val mediator = DistributedPubSub(actorSystem).mediator
  val cluster = Cluster(actorSystem)
  cluster.joinSeedNodes(seedNodes)

  val actor = actorSystem.actorOf(Props(new Publisher(mediator)), "Publisher")

  //appears that this sleep is needed for the communication to be up
  //sending something too early just causes the message to disappear into void without any trace nor log
  Thread.sleep(2000)

  actor ! Unicast(Seq("ServiceA", "ServiceB"), Message("Unicast Hello-1!!!"))
  actor ! Unicast(Seq("ServiceA", "ServiceB"), Message("Unicast Hello-2!!!"))
  actor ! Broadcast(Seq("ServiceA"), Message("Broadcast Hello Again!!!"))

  //stupid pause to allow for responses to propagate and be logged
  Thread.sleep(2000)
  actorSystem.terminate().onComplete(_ => System.exit(1))
}