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

import java.net.InetSocketAddress

import com.twitter.finagle
import com.twitter.finagle.Service
import com.twitter.finagle.http.{Request, Response}
import com.twitter.util.Future
import org.scalatest._
import placeholder.Main

import scala.concurrent.Await
import scala.concurrent.duration._

trait IntegrationTestBase extends FeatureSpec with GivenWhenThen with Matchers with BeforeAndAfterAll {
  val server = Main

  var serverAddress: InetSocketAddress = _
  var client: Service[Request, Response] = _
  var requestHost: String = _

  server.main(Array())

  override def beforeAll(): Unit = {
    serverAddress = Await.result(server.getServerAddress, 10.seconds)
    requestHost = s"localhost:${serverAddress.getPort.toString}"
    client = finagle.Http.newService(requestHost)
  }

  def performRequest(request: Request): Future[Response] = {
    request.host = requestHost
    client(request)
  }
} 

//设置package包名称以及导入依赖的类
import java.io.File
import java.net.InetSocketAddress

import play.sbt.PlayRunHook
import sbt._

object EmberRunner {
  def apply(logger: Logger,
            baseDir: File,
            port: Int = 4200): PlayRunHook = {

    object EmberRunHook extends PlayRunHook {
      var watchProcess: Option[Process] = None

      override def beforeStarted(): Unit = {
        if (baseDir.exists()) {
          logger.info("Executing npm install")
          val npm = Process(Seq("npm", "install"), baseDir).lines
          npm.foreach(logger.info(_))

          logger.info("Executing bower install")
          val bower = Process(Seq("bower", "install"), baseDir).lines
          bower.foreach(logger.info(_))
        } else {
          logger.info(s"Skipping npm and bower install. UI application directory ${baseDir.getAbsolutePath} not found.")
        }
      }

      override def afterStarted(addr: InetSocketAddress): Unit = {
        addr.getAddress.getHostAddress
        val url = s"http://localhost:${addr.getPort}"
        if (baseDir.exists()) {
          logger.info(s"Starting ember server in development mode. Setting proxy to $url")
          watchProcess = Some(Process(Seq("ember", "serve", "--proxy", url, "--port", port.toString), baseDir).run(logger))
        } else {
          logger.info(s"Skipping ember server start. UI application directory ${baseDir.getAbsolutePath} not found.")
        }
      }

      override def afterStopped(): Unit = {
        logger.info("Attempting to stop ember server")
        watchProcess.foreach(_.destroy())
        watchProcess = None
      }
    }
    EmberRunHook
  }
} 

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

import java.io.File
import java.net.InetSocketAddress

import kafka.utils.Utils
import org.apache.zookeeper.server.{NIOServerCnxnFactory, ZooKeeperServer}


class EmbeddedZookeeper(zkPort: Int, zkSnapshotDir: File, zkLogDir: File, autoStart: Boolean) {

  val connectString = s"127.0.0.1:${zkPort}"
  val tickTime = 500
  val factory = new NIOServerCnxnFactory()
  val maxZkConnections = 100
  factory.configure(new InetSocketAddress("127.0.0.1", zkPort), maxZkConnections)

  var zookeeper: ZooKeeperServer = null

  def this(port: Int, autoStart: Boolean = true) = this(port, TestUtils.tempDir, TestUtils.tempDir, autoStart)

  if (autoStart) {
    start()
  }

  def start(): Unit = {
    // With Zookeeper 3.4, the startup logic of the ZookeeperServer has changed where a sequence of:
    // zookeeper.start() -> zookeeper.shutdown() -> zookeeper.start()
    // will fail to restart the ZookeeperServer. Because of this, a new ZookeeperServer needs to be instantiated if
    // we want to simulate Zookeeper unavailability during tests
    zookeeper = new ZooKeeperServer(zkSnapshotDir, zkLogDir, tickTime)
    factory.startup(zookeeper)
  }

  def stop(): Unit = {
    zookeeper.shutdown()
    zookeeper = null
  }

  def snapshotDir: File = zkSnapshotDir

  def logDir: File = zkLogDir

  def cleanShutdown(): Unit = {
    shutdown()
    Utils.rm(zkLogDir)
    Utils.rm(zkSnapshotDir)
  }

  def shutdown(): Unit = {
    Utils.swallow(zookeeper.shutdown())
    Utils.swallow(factory.shutdown())
    zookeeper = null
  }
} 

//设置package包名称以及导入依赖的类
package eu.inn.kafka.mimic

import java.net.{InetAddress, InetSocketAddress}
import java.util.concurrent.TimeUnit
import java.util.{Timer, TimerTask}

import com.codahale.metrics.graphite.{Graphite, GraphiteReporter}
import nl.grons.metrics.scala.{InstrumentedBuilder, MetricName}

object Metrics extends Logging {
  val metricRegistry = new com.codahale.metrics.MetricRegistry()

  def startReporter(host: String, port: Int, prefix: String, reportPeriod: Long) = {

    val graphite = new Graphite(new InetSocketAddress(host, port))

    val reporter = GraphiteReporter.forRegistry(metricRegistry)
      .prefixedWith(prefix + "." + InetAddress.getLocalHost.getHostName.replaceAll("\\.", "-"))
      .build(graphite)

    new Timer("graphite-reporter-timer").schedule(
      new TimerTask { def run() = reporter.report() },
      reportPeriod,
      reportPeriod
    )
  }
}

trait Metrics extends InstrumentedBuilder {
  override lazy val metricBaseName = MetricName("kafka-mimic")

  lazy val metricRegistry = Metrics.metricRegistry
}

trait MetricsComponent extends Metrics with Logging {

  this: ConfigComponent =>

  def startMetricsReporting() = {
    if (config.hasPath("enabled") && config.getBoolean("enabled")) {
      val host = config.getString("host")
      val port = config.getInt("port")
      log.info(s"Starting graphite reporter for $host:$port")
      Metrics.startReporter(
        host,
        port,
        config.getString("prefix"),
        config.getDuration("report-period", TimeUnit.MILLISECONDS)
      )
    }

    metrics.gauge("heartbeat") { 1 }
  }
} 

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

import akka.actor.Address
import akka.remote.transport.AssociationHandle
import akka.remote.transport.AssociationHandle.{ HandleEventListener, InboundPayload }
import akka.remote.transport.Transport.AssociationEventListener
import akka.util.ByteString
import java.net.{ SocketAddress, InetAddress, InetSocketAddress }
import org.jboss.netty.buffer.{ ChannelBuffer, ChannelBuffers }
import org.jboss.netty.channel._
import scala.concurrent.{ Future, Promise }


private[remote] class UdpAssociationHandle(val localAddress: Address,
                                           val remoteAddress: Address,
                                           private val channel: Channel,
                                           private val transport: NettyTransport) extends AssociationHandle {

  override val readHandlerPromise: Promise[HandleEventListener] = Promise()

  override def write(payload: ByteString): Boolean = {
    if (!channel.isConnected)
      channel.connect(new InetSocketAddress(InetAddress.getByName(remoteAddress.host.get), remoteAddress.port.get))

    if (channel.isWritable && channel.isOpen) {
      channel.write(ChannelBuffers.wrappedBuffer(payload.asByteBuffer))
      true
    } else false
  }

  override def disassociate(): Unit = try channel.close()
  finally transport.udpConnectionTable.remove(transport.addressToSocketAddress(remoteAddress))

} 

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

import akka.testkit._
import scala.concurrent.duration._
import akka.actor.{ Address, ExtendedActorSystem, ActorSystem }
import com.typesafe.config.ConfigFactory
import java.nio.channels.ServerSocketChannel
import java.net.InetSocketAddress
import scala.collection.JavaConverters._

@org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner])
class DaemonicSpec extends AkkaSpec {

  def addr(sys: ActorSystem, proto: String) =
    sys.asInstanceOf[ExtendedActorSystem].provider.getExternalAddressFor(Address(s"akka.$proto", "", "", 0)).get

  def unusedPort = {
    val ss = ServerSocketChannel.open().socket()
    ss.bind(new InetSocketAddress("localhost", 0))
    val port = ss.getLocalPort
    ss.close()
    port
  }

  "Remoting configured with daemonic = on" must {

    "shut down correctly after getting connection refused" in {
      // get all threads running before actor system i started
      val origThreads: Set[Thread] = Thread.getAllStackTraces().keySet().asScala.to[Set]
      // create a separate actor system that we can check the threads for
      val daemonicSystem = ActorSystem("daemonic", ConfigFactory.parseString("""
        akka.daemonic = on
        akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        akka.remote.netty.tcp.transport-class = "akka.remote.transport.netty.NettyTransport"
        akka.remote.netty.tcp.port = 0
        akka.log-dead-letters-during-shutdown = off
      """))

      val unusedAddress = addr(daemonicSystem, "tcp").copy(port = Some(unusedPort))
      val selection = daemonicSystem.actorSelection(s"${unusedAddress}/user/SomeActor")
      selection ! "whatever"
      Thread.sleep(2.seconds.dilated.toMillis)

      // get new non daemonic threads running
      val newNonDaemons: Set[Thread] = Thread.getAllStackTraces().keySet().asScala.seq.
        filter(t ? !origThreads(t) && t.isDaemon == false).to[Set]

      newNonDaemons should be(Set.empty[Thread])
      shutdown(daemonicSystem)
    }
  }
} 

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

import akka.testkit.AkkaSpec
import akka.actor.{ PoisonPill, Props, AddressFromURIString }
import akka.testkit.ImplicitSender
import akka.remote.testconductor.Controller.NodeInfo
import java.net.InetSocketAddress
import java.net.InetAddress

object ControllerSpec {
  val config = """
    akka.testconductor.barrier-timeout = 5s
    akka.actor.provider = akka.remote.RemoteActorRefProvider
    akka.actor.debug.fsm = on
    akka.actor.debug.lifecycle = on
    """
}

class ControllerSpec extends AkkaSpec(ControllerSpec.config) with ImplicitSender {

  val A = RoleName("a")
  val B = RoleName("b")

  "A Controller" must {

    "publish its nodes" in {
      val c = system.actorOf(Props(classOf[Controller], 1, new InetSocketAddress(InetAddress.getLocalHost, 0)))
      c ! NodeInfo(A, AddressFromURIString("akka://sys"), testActor)
      expectMsg(ToClient(Done))
      c ! NodeInfo(B, AddressFromURIString("akka://sys"), testActor)
      expectMsg(ToClient(Done))
      c ! Controller.GetNodes
      expectMsgType[Iterable[RoleName]].toSet should be(Set(A, B))
      c ! PoisonPill // clean up so network connections don't accumulate during test run
    }

  }

} 

//设置package包名称以及导入依赖的类
import java.net.InetSocketAddress

import play.sbt.PlayRunHook
import sbt._

object Webpack {
  def apply(base: File): PlayRunHook = {
    object WebpackHook extends PlayRunHook {
      var process: Option[Process] = None

      override def beforeStarted() = {
        process = Option(
          Process("webpack", base).run()
        )
      }

      override def afterStarted(addr: InetSocketAddress) = {
        process = Option(
          Process("webpack --watch", base).run()
        )
      }

      override def afterStopped() = {
        process.foreach(_.destroy())
        process = None
      }
    }

    WebpackHook
  }
} 

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

import java.net.InetSocketAddress

import akka.actor.{ActorSystem, Props}
import chatapp.client.ClientMessage.SendMessage


object ClientMain extends App {

  val Port = 18573

  val system = ActorSystem("ClientMain")
  val clientConnection = system.actorOf(Props(new ClientActor(new InetSocketAddress("localhost", Port), system)))
  val bufferedReader = io.Source.stdin.bufferedReader()
  loop("")

  def loop(message: String): Boolean = message match {
    case "~quit" =>
      system.terminate()
      false
    case _ =>
      val msg = bufferedReader.readLine()
      clientConnection ! SendMessage(msg)
      loop(msg)
  }
} 

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

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorSystem, Kill}
import akka.io.Tcp._
import akka.io.{IO, Tcp}
import akka.util.ByteString
import chatapp.client.ClientMessage.SendMessage


class ClientActor(address: InetSocketAddress, actorSystem: ActorSystem) extends Actor {

  IO(Tcp)(actorSystem) ! Connect(address)

  def receive = {
    case CommandFailed(command: Tcp.Command) =>
      println("Failed to connect to " + address.toString)
      self ! Kill
      actorSystem.terminate()
    case Connected(remote, local) =>
      println("Successfully connected to " + address)
      val connection = sender()
      connection ! Register(self)
      context become {
        case Received(data) =>
          println(data.decodeString("US-ASCII"))
        case SendMessage(message) =>
          connection ! Write(ByteString(message))
      }
  }
} 

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

import java.net.InetSocketAddress

import akka.actor._
import akka.io.Tcp.Bind
import akka.io.{IO, Tcp}

class GameServer(handlerProps: (ActorRef, InetSocketAddress) => Props)
  extends Actor with ActorLogging {

  import Tcp._

  override def receive: Receive = {
    case [email protected](localAddress) => log.debug("Bound server to {}.", localAddress)
    case [email protected](cmd) =>
      log.error("Command {} failed so closing server.", cmd)
      context stop self
    case [email protected](remote, local) =>
      val connection = sender()
      val handler = context.actorOf(handlerProps(connection, remote))
      connection ! Register(handler)
  }
}

class GameServerFactory(host: String, port: Int, handlerProps: (ActorRef, InetSocketAddress) => Props)
                       (implicit system: ActorSystem) extends GameEndPoint {
  override val connectionString: String = host + ":" + port

  val server = system.actorOf(Props(classOf[GameServer], handlerProps))

  override def init(): Unit = {
    IO(Tcp) ! Bind(server, new InetSocketAddress(host, port))
  }
} 

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

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.io.Tcp

object ServerActor {
  def props(bindAddress: InetSocketAddress, tcp: ActorRef, handler:ActorRef) =
    Props(classOf[ServerActor], bindAddress, tcp, handler)
}

class ServerActor(bindAddress: InetSocketAddress, tcp: ActorRef, handler: ActorRef) extends Actor with ActorLogging {
  import Tcp._

  // TODO: verify why we bind from within the actor
 tcp ! Bind(self, bindAddress)

  def receive = {
    case b @ Bound(localAddress) =>
      log.info(s"Tcp Server bound to <$localAddress>")

    case CommandFailed(_: Bind) =>
      log.warning("Tcp ServerActor failed to bind. Stopping...")
      context stop self

    case c @ Connected(remote, local) =>
      log.info(s"Tcp Server Connected. remote=<$remote>, local=<$local>. Registering handler...")
      val connection = sender()
      connection ! Register(handler)
  }
} 

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

import akka.actor.{ Actor, ActorRef, Props }
import akka.io.Tcp
import akka.util.ByteString
import java.net.InetSocketAddress

object ClientActor {
  def props(remote: InetSocketAddress, tcp: ActorRef, replies: ActorRef) =
    Props(classOf[ClientActor], remote, tcp, replies)
}

class ClientActor(remote: InetSocketAddress, tcp:ActorRef, listener: ActorRef) extends Actor {
  import Tcp._
  import context.system

  tcp ! Connect(remote)

  def receive = {
    case CommandFailed(_: Connect) =>
      listener ! "connect failed"
      context stop self

    case c @ Connected(remote, local) =>
      listener ! c
      val connection = sender()
      connection ! Register(self)
      context become {
        case data: ByteString =>
          connection ! Write(data)
        case CommandFailed(w: Write) =>
          // O/S buffer was full
          listener ! "write failed"
        case Received(data) =>
          listener ! data
        case "close" =>
          connection ! Close
        case _: ConnectionClosed =>
          listener ! "connection closed"
          context stop self
      }
  }
} 

//设置package包名称以及导入依赖的类
package wow.common.network

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.io.{IO, Tcp}
import akka.util.ByteString


object Client {
  def props(remote: InetSocketAddress, listener: ActorRef) = Props(classOf[Client], remote, listener)
}


class Client(remote: InetSocketAddress, listener: ActorRef) extends Actor with ActorLogging {

  import akka.io.Tcp._
  import context.system

  IO(Tcp) ! Connect(remote)

  def receive: PartialFunction[Any, Unit] = {
    case CommandFailed(_: Connect) =>
      listener ! "connect failed"
      context stop self

    case [email protected](dist, local) =>
      log.debug("[CLIENT] Connected from: " + local + " to: " + dist)
      listener ! c
      val connection = sender()
      connection ! Register(self)
      context become {
        case 42 => sender() ! 42
        case data: ByteString =>
          log.debug("[CLIENT] Sent : " + data)
          connection ! Write(data)
        case CommandFailed(_: Write) =>
          // O/S buffer was full
          listener ! "write failed"
        case Received(data) =>
          log.debug("[CLIENT] Received : " + data)
          listener ! data
        case "close" =>
          connection ! Close
        case _: ConnectionClosed =>
          listener ! "connection closed"
          context stop self
      }
  }
} 

//设置package包名称以及导入依赖的类
package knot.remote.endpoints

import java.net.InetSocketAddress
import java.nio.channels.SelectionKey

import knot.remote.channel.{NioChannel, NioServerSocketChannel, NioSocketChannel}

class NioServerEndpoint extends NioEndpoint {

  override def processKey(key: SelectionKey, channel: NioSocketChannel): Unit = {
    if (key.isValid && (key.isReadable || key.readyOps() == 0)) {
      channel.read()
    }
  }

  override def processKey(key: SelectionKey, channel: NioServerSocketChannel): Unit = {
    if (key.isAcceptable) {
      log.debug("SERVER: New Connection")
      val c = channel.accept
      c.register(this, SelectionKey.OP_READ)
    }
  }

  override def init(channel: NioChannel, address: InetSocketAddress): Unit = {
    channel.bind(address)
    channel.register(this, SelectionKey.OP_ACCEPT)
  }
} 

//设置package包名称以及导入依赖的类
package knot.remote.endpoints

import java.io.IOException
import java.net.InetSocketAddress
import java.nio.channels.SelectionKey

import knot.remote.channel.{NioChannel, NioSocketChannel}

class NioClientEndPoint extends NioEndpoint {

  override def processKey(key: SelectionKey, channel: NioSocketChannel): Unit = {
    if (key.isConnectable) {
      if (!channel.finishConnect) throw new IOException("Connect Fail")
      val ops = key.interestOps() & ~SelectionKey.OP_CONNECT
      key.interestOps(ops)
      log.debug("CLIENT: Connected")
    }

    if (key.isReadable || key.readyOps() == 0) {
      channel.read()
    }
  }

  override def init(channel: NioChannel, address: InetSocketAddress): Unit = {
    channel.connect(address)
    channel.register(this, SelectionKey.OP_CONNECT)
  }
} 

//设置package包名称以及导入依赖的类
package com.amadornes.modcast.bot.servers

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorRef}
import akka.io.{IO, Tcp}
import akka.util.ByteString
import com.amadornes.modcast.bot.Configuration
import grizzled.slf4j.Logging

import scala.collection.mutable.ArrayBuffer


class MCWhitelistServer extends Actor with Logging {
	
	import Tcp._
	import context.system
	
	IO(Tcp) ! Bind(self, new InetSocketAddress(Configuration.config.getString("mc.host"), Configuration.config.getInt("mc.port")))
	
	val connections = new ArrayBuffer[ActorRef]()
	
	def receive = {
		case str: String =>
		//Ignored
		
		case [email protected](localAddress) =>
			info("MC Whitelist TCP Server is online and listening.")
		
		case CommandFailed(_: Bind) =>
			error("Bind failed. MC Whitelist Server is not functional.")
			context stop self
		
		case [email protected](remote, local) =>
			if (Configuration.config.getStringList("mc.acceptedIPs").contains(remote.getAddress.getHostAddress)) {
				sender() ! Register(self)
				connections += sender()
			} else {
				warn(s"Refusing connection from ${remote.getAddress.getHostAddress}")
				sender() ! Close
			}
		
		case MCWhitelistServer.WhitelistUser(userID) =>
			for (connection <- connections)
				connection ! Write(ByteString(s"W$userID\n"))
		case MCWhitelistServer.UnWhitelistUser(userID) =>
			for (connection <- connections)
				connection ! Write(ByteString(s"U$userID\n"))
	}
}

object MCWhitelistServer {
	
	case class WhitelistUser(id: String)
	
	case class UnWhitelistUser(id: String)
	
} 

//设置package包名称以及导入依赖的类
package net.shiroka.cat
import java.net.InetSocketAddress
import com.sun.net.httpserver.{ HttpExchange, HttpHandler, HttpServer }

object TinyHttpServer {
  def serve(port: Int): Unit = {
    val server = HttpServer.create(new InetSocketAddress(port), 0)
    server.createContext("/", new HttpHandler() {
      def handle(he: HttpExchange): Unit = sendResponse(he, "ok\n")
    })
    server.setExecutor(null)
    server.start()
    println("HTTP server started")
    //server.stop(0)
  }

  private def sendResponse(he: HttpExchange, response: String): Unit = {
    he.sendResponseHeaders(200, response.length)
    val os = he.getResponseBody
    try { os.write(response.getBytes) }
    finally { os.close() }
  }
} 

//设置package包名称以及导入依赖的类
package com.github.wildprairie.common.actors.shared

import java.net.InetSocketAddress

import akka.actor.{ActorRef, Props}
import com.github.wakfutcp.actors.server.TcpServer
import com.github.wildprairie.common.actors.ClusteredActor


object WakfuServer {
  val ROLE_AUTH = "auth"
  val ROLE_WORLD = "world"
  val ROLE_MASTER = "master"
}

abstract class WakfuServer extends ClusteredActor {
  override def preStart(): Unit =
    super.preStart()
  context.actorOf(
    TcpServer.props(
      new InetSocketAddress(host, port),
      WakfuServerConnection.props,
      newHandlerProps
    ),
    "tcp-server"
  )

  def newHandlerProps: (ActorRef) => Props
  def host: String
  def port: Int

  override def receive: Receive = PartialFunction.empty
} 

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

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.scaladsl.Source
import akka.util.ByteString

import cassandra.streams.{Complete, Partial}

import java.net.InetSocketAddress

import scala.concurrent.{ExecutionContext, Future}
import scala.util.Random

trait Session {
  def close()(implicit ec: ExecutionContext): Future[Unit]
  def execute(cl: ConsistencyLevel)(query: String, values: AnyRef*)(implicit ec: ExecutionContext): Future[ResultSource]
  def cluster(): Cluster
  def keyspace(): Option[String]
  def hosts(): List[String]
}

class SimpleSession(
  val cluster: Cluster,
  val keyspace: Option[String],
  val hosts: List[String],
  actorSystem: ActorSystem
) extends Session {
  val address = "(.+):(.+)".r
  val random = new Random()

  val connections: List[Connection] = hosts.map {
    case address(host, port) => new Connection(new InetSocketAddress(host, port.toInt))(actorSystem)
    case malformed => throw new IllegalArgumentException(s"${malformed} is not a host:port")
  }

  def connect()(implicit ec: ExecutionContext): Future[Unit] = Future.sequence(connections.map { conn =>
    keyspace match {
      case Some(ks) => conn.connect(ks)
      case None => conn.connect()
    }
  }).map(_ => ())

  def close()(implicit ec: ExecutionContext): Future[Unit] = ???

  def execute(cl: ConsistencyLevel)(query: String, values: AnyRef*)(implicit ec: ExecutionContext): Future[ResultSource] = {
    val conn = connections(random.nextInt() % connections.length)
    println("VALUES " + values)
    conn.stream(cl)(query, values).map {
      case Partial((fh, fb), source) => ResultSource(fh, fb, source)
      case Complete((fh, fb), bytes) => ResultSource(fh, fb, Source.single(bytes))
    }
  }
}