//设置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.server

import java.nio.charset.Charset

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

object SimplisticHandler {
  def props(encoding: Charset) = Props(classOf[SimplisticHandler], encoding)
}

class SimplisticHandler(encoding: Charset) extends Actor with ActorLogging {
  import Tcp._

  def receive: Receive = {
    case Received(data) =>
      log.info(s"SimplisticHandler received <${data.length}> bytes")
      sender() ! Write(aResponsePacketFrom(data))

    case PeerClosed =>
      log.info("SimplisticHandler received PeerClosed, stopping.")
      context stop self
  }

  def aResponsePacketFrom(data: ByteString): ByteString = {
    val map = new String(data.toArray, 0, data.length, encoding).flatMap(caesarCypher)
    ByteString(map.getBytes(encoding))
  }

  def caesarCypher(c: Char): scala.collection.mutable.ArrayOps[Char] =
    if (c != '\n' && c != '\r') Character.toChars(c + 3) else Array(c)

} 

//设置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, Props, SupervisorStrategy}
import akka.io.Tcp._
import akka.io.{IO, Tcp}

case object GetAddress


class TCPServer[A <: TCPSessionFactory](val factory: A, val address: String, val port: Int)
  extends Actor with ActorLogging {
  log.debug("Binding server with socket")
  IO(Tcp)(context.system) ! Bind(self, new InetSocketAddress(address, port))

  override def supervisorStrategy: SupervisorStrategy = SupervisorStrategy.stoppingStrategy

  override def postStop(): Unit = log.debug(s"Stopped TCP server for $address:$port")

  def receive: PartialFunction[Any, Unit] = {
    case Bound(localAddress) =>
      log.debug(s"TCP port opened at: ${localAddress.getHostString}:${localAddress.getPort}")

    case Connected(remote, local) =>
      log.debug(s"Remote connection set from $remote to $local")
      val handlerRef = context.actorOf(factory.props(sender), factory.PreferredName + TCPSession.PreferredName(remote))
      sender ! Register(handlerRef)

    case CommandFailed(_: Bind) => context stop self
  }
}

object TCPServer {
  def props[A <: TCPSessionFactory](companion: A, address: String, port: Int): Props = Props(classOf[TCPServer[A]],
    companion,
    address,
    port)

  val PreferredName = "tcp"
} 

//设置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 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 controllers

import java.net.InetSocketAddress
import akka.actor.Actor
import akka.io.{IO, Tcp}
import akka.util.ByteString
import scala.concurrent.Promise
import Tcp._

class TcpClient(remote: InetSocketAddress, requestData: String, thePromise: Promise[String]) extends Actor {
    import context.system

    println("Connecting")
    IO(Tcp) ! Connect(remote)

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

        case c @ Connected(remote, local) =>
            println ("Connect succeeded")
            val connection = sender()
            connection ! Register(self)
            println("Sending request early")
            connection ! Write(ByteString(requestData))

            context become {
                case CommandFailed(w: Write) =>
                    println("Failed to write request.")
                case Received(data) =>
                    println("Received response.")
                    thePromise.success(data.decodeString("UTF-8"))
                case "close" =>
                    println("Closing connection")
                    connection ! Close
                case _: ConnectionClosed =>
                    println("Connection closed by server.")
                    context stop self
            }
        case _ => println("Something else is up.")
    }
} 

//设置package包名称以及导入依赖的类
package build.unstable.sonicd

import java.net.InetSocketAddress

import akka.http.scaladsl.Http
import akka.io.Tcp
import akka.stream.scaladsl.{Tcp ? StreamTcp}
import build.unstable.sonicd.api.AkkaApi
import build.unstable.sonicd.system.{AkkaService, AkkaSystem}
import com.typesafe.sslconfig.akka.AkkaSSLConfig
import com.typesafe.sslconfig.akka.util.AkkaLoggerFactory
import com.typesafe.sslconfig.ssl.{ConfigSSLContextBuilder, SSLConfigFactory}

object Sonicd extends App with AkkaSystem with AkkaService with AkkaApi with SonicdLogging {

  val http = Http()
  val sslConfigFactory = AkkaSSLConfig()

  val sonicOverrides = system.settings.config.getConfig("sonicd.ssl-config")
  val defaults = system.settings.config.getConfig("ssl-config")
  val config = SSLConfigFactory.parse(sonicOverrides withFallback defaults)

  val keyManagerFactory = sslConfigFactory.buildKeyManagerFactory(config)
  val trustManagerFactory = sslConfigFactory.buildTrustManagerFactory(config)
  val sslContext = new ConfigSSLContextBuilder(new AkkaLoggerFactory(system), config, keyManagerFactory, trustManagerFactory).build()

  http.bindAndHandle(handler = httpHandler, interface = SonicdConfig.INTERFACE, port = SonicdConfig.HTTP_PORT)

  tcpIoService.tell(Tcp.Bind(tcpService,
    new InetSocketAddress(SonicdConfig.INTERFACE, SonicdConfig.TCP_PORT), options = Nil, pullMode = true), tcpService)

  log.info( "STARTING SONIC SERVICE V.{} ({} {}) on interface {}; http port: {}; tcp port: {}",
    BuildInfo.version, BuildInfo.commit, BuildInfo.builtAt, SonicdConfig.INTERFACE,
    SonicdConfig.HTTP_PORT, SonicdConfig.TCP_PORT)

  log.info( "ssl config: {} with default protocol: {}", config, config.protocol)

} 

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

import akka.actor.{ Actor, Props }
import akka.io.{ IO, Tcp }
import java.net.InetSocketAddress
import com.vinctus.venatus.handlers.LoginHandler

class LoginServer extends Actor {

  import Tcp._
  import context.system
 
  IO(Tcp) ! Tcp.Bind(self, new InetSocketAddress("localhost", 6666))

  def receive = { 
    case Tcp.CommandFailed(_: Bind) =>
      context.stop(self)

    case c @ Tcp.Connected(remote, local) =>
      val handler = context.actorOf(Props[LoginHandler])
      val connection = sender
      connection ! Tcp.Register(handler)
  }

} 

//设置package包名称以及导入依赖的类
package net.furikuri.cproxy.server

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


class PeerConnectionHandler(connection: ActorRef, clients: Map[String, ActorRef]) extends Actor with ActorLogging {

  import Tcp._

  def receive = {
    case Received(data) =>
      log.info("New request")
      val clientProxy = clients("hello")
      clientProxy ! Write(data)

    case w: Write =>
      log.info("Back to client")
      connection ! Write(w.data)

    case PeerClosed =>
      log.info("Peer closed")
      context.stop(self)
  }
} 

//设置package包名称以及导入依赖的类
package net.furikuri.cproxy.server

import java.util.UUID

import akka.actor.{Actor, ActorLogging, ActorRef}
import akka.io.Tcp
import akka.util.ByteString
import net.furikuri.cproxy.server.ProxyServer.ClientConnected

class ClientProxyHandler(connection: ActorRef, server: ActorRef) extends Actor with ActorLogging {

  import Tcp._

  var init = false
  var last: ActorRef = _

  var clientsActors:Map[UUID, ActorRef] = Map()

  def receive = {
    case Received(data) =>
      log.info("Receive new data")
      if (!init) {
        val clientId = data.decodeString("utf-8")
        println("Received: " + clientId)
        server ! ClientConnected(clientId)
        init = true
      } else {
        val value = data.decodeString("utf-8")
        val firstNewLine = value.indexOf("\n")
        val header = value.substring(0, firstNewLine)
        val rawData = data.drop(firstNewLine + 1).decodeString("utf-8")
        val uuid = header.split(":").apply(0)
        log.info("Got new header:\n" + header)
        log.info("Got new data:\n" + rawData)

        val uuidKey = UUID.fromString(uuid)
        val receiver = clientsActors(uuidKey)
        log.info("Return")

        receiver ! Write(data.drop(firstNewLine + 1))
      }

    case w: Write =>
      val uuid = UUID.randomUUID()
      log.info("Write new data")
      connection ! Write(ByteString(uuid.toString + ":" + w.data.length + "\n"))
      connection ! Write(w.data)
      clientsActors += uuid -> sender()
      last = sender()

    case PeerClosed =>
      log.info("Peer closed")
      context.stop(self)
  }
} 

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

import java.net.InetSocketAddress

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

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

class TcpClient(remote: InetSocketAddress, listener: ActorRef) extends Actor with ActorLogging {
  import Tcp._
  import context.system

  IO(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 =>
          log.info("Write data")
          connection ! Write(data)
        case CommandFailed(w: Write) =>
          listener ! "write failed"
        case Received(data) =>
          log.info("New data")
          listener ! data
        case "close" =>
          connection ! Close
        case _: ConnectionClosed =>
          listener ! "connection closed"
          context stop self
      }
  }
} 

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

import akka.actor._
import akka.io.{IO, Tcp}
import scala.concurrent.Future
import akka.util.ByteString
import java.net.InetSocketAddress

class EasyConsole(
  val inet: InetSocketAddress,
  val welcomeMessage: String,
  val func: EasyConsole.Func
) extends Actor with ActorLogging {

  import context.system
  import Tcp._
  val name = s"ConsoleNet(${inet.getHostName}:${inet.getPort})"
  val welcome = welcomeMessage + "\n"

  val manager = IO(Tcp)
  manager ! Bind(self, inet, 1)

  val binding:Receive = {
    case Bound(addr) =>
      log.info(s"${name}: Bounded")
      context become (bound,false)
    case CommandFailed(cmd: Bind) =>
      log.error(s"${name}: Binding has failed")
      context stop self
  }

  val bound:Receive = {
    case Connected(remote, local) =>
      val servantName = s"Client(${remote.getHostName}:${remote.getPort})"
      log.info(s"${name}: New Connection from $servantName")
      val servant = context actorOf Servant.props(name, servantName, func)
      sender ! Register(servant)
      sender ! Write(ByteString(welcome))
  }

  def receive = binding

}

object EasyConsole {

  type Func = PartialFunction[List[String],String]

  def props(
    host: String,
    port: Int,
    welcome: String,
    func: Func
  ):Props = props( new InetSocketAddress(host, port), welcome, func)

  def props(inet: InetSocketAddress, welcome:String, func:Func):Props = Props {
    new EasyConsole(inet, welcome, func)
  }
} 

//设置package包名称以及导入依赖的类
import org.scalatest._
import akka.testkit._
import akka.actor._
import akka.io.Tcp
import akka.util.ByteString
import com.teanlab.akkautil.consolenet._
import scala.concurrent.duration._

class MySpec(
  _system: ActorSystem
) extends TestKit(_system) 
    with ImplicitSender
    with WordSpecLike
    with Matchers 
    with BeforeAndAfterAll {

  def this() = this(ActorSystem("MySpec"))

  override def afterAll {
    TestKit.shutdownActorSystem(system)
  }

  val helloResponse = "Hello, Nice to meet you :)"

  val func:ConsoleNet.Func = {
    case "hello" => helloResponse
    case "help" => "Ops!"
  }

  "First" must {
    "success" in {
      
      val servant = _system.actorOf(Servant.props("A","B", func))

      servant ! Tcp.Received(ByteString("help\nhe"))
      receiveN(1)

      servant ! Tcp.Received(ByteString("llo\nOPS"))
      expectMsgPF(1 seconds){
        case Tcp.Write(output: ByteString, _) =>
          val its = new String(output.toArray)
          info(its)
          assert( its == helloResponse)
        case unexp => fail(s"Unexpected output with type: ${unexp.getClass}")
      }
    }
  }

} 

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

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.io.Tcp.{Bind, Bound, Connected, Register}
import akka.io.{IO, Tcp}

object ServerChannel {
  def apply(address: InetSocketAddress, game: ActorRef) =
    Props(new ServerChannel(address, game))
}


private final class ServerChannel(address: InetSocketAddress, game: ActorRef) extends Actor with ActorLogging {
  import context.system

  override def preStart(): Unit = {
    IO(Tcp) ! Bind(self, address)
  }

  override def receive: Receive = {
    case Bound(local) =>
      log.info(s"Bound at $local")

    case Connected(remote, local) =>
      val channelHandler = context actorOf PlayerChannel(game)
      sender() ! Register(channelHandler)

    case message =>
      unhandled(message)
  }
} 

//设置package包名称以及导入依赖的类
package io.iohk.ethereum.network

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.agent.Agent
import akka.io.Tcp.{Bind, Bound, CommandFailed, Connected}
import akka.io.{IO, Tcp}
import io.iohk.ethereum.utils.{NodeStatus, ServerStatus}
import org.spongycastle.util.encoders.Hex

class ServerActor(nodeStatusHolder: Agent[NodeStatus], peerManager: ActorRef) extends Actor with ActorLogging {

  import ServerActor._
  import context.system

  override def receive: Receive = {
    case StartServer(address) =>
      IO(Tcp) ! Bind(self, address)
      context become waitingForBindingResult
  }

  def waitingForBindingResult: Receive = {
    case Bound(localAddress) =>
      val nodeStatus = nodeStatusHolder()
      log.info("Listening on {}", localAddress)
      log.info("Node address: enode://{}@{}:{}",
        Hex.toHexString(nodeStatus.nodeId),
        localAddress.getAddress.getHostAddress,
        localAddress.getPort)
      nodeStatusHolder.send(_.copy(serverStatus = ServerStatus.Listening(localAddress)))
      context become listening

    case CommandFailed(b: Bind) =>
      log.warning("Binding to {} failed", b.localAddress)
      context stop self
  }

  def listening: Receive = {
    case Connected(remoteAddress, localAddress) =>
      val connection = sender()
      peerManager ! PeerManagerActor.HandlePeerConnection(connection, remoteAddress)
  }
}

object ServerActor {
  def props(nodeStatusHolder: Agent[NodeStatus], peerManager: ActorRef): Props =
    Props(new ServerActor(nodeStatusHolder, peerManager))

  case class StartServer(address: InetSocketAddress)
} 

//设置package包名称以及导入依赖的类
package com.github.nrf110.io.transport

import java.net.InetSocketAddress

import akka.actor._
import akka.io.Tcp.Connect
import akka.io.{IO, Tcp}
import akka.util.ByteString

class ConnectionHandler(remoteAddress: InetSocketAddress, localAddress: Option[InetSocketAddress], listener: ActorRef)
  extends Actor with ActorLogging {

  IO(Tcp)(context.system) ! Connect(remoteAddress, localAddress)

  def initial: Receive = {
    case fail @ Tcp.CommandFailed(_: Tcp.Connect) =>
      log.warning("Failed to connect to {}", remoteAddress)
      listener ! ConnectionFailed
      die()
    case c @ Tcp.Connected(remote, local) =>
      listener ! c
      val connection = sender()
      connection ! Tcp.Register(self)
      context become connected(connection)
  }

  def connected(connection: ActorRef): Receive = {
    case data: ByteString =>
      connection ! Tcp.Write(data)

    case Tcp.CommandFailed(w: Tcp.Write) =>
      log.warning("Write failed to {}", remoteAddress)
      listener ! WriteFailed

    case Tcp.Received(data) =>
      listener ! data

    case Close =>
      connection ! Tcp.Close

    case _: Tcp.ConnectionClosed =>
      log.info("Closed connection to {}", remoteAddress)
      listener ! Closed
      die()
  }

  def die(): Unit = context stop self

  def receive: Receive = initial
}

object ConnectionHandler {
  trait Factory {
    def apply(remoteAddress: InetSocketAddress, localAddress: Option[InetSocketAddress], handler: ActorRef): ActorRef
  }

  def apply(remoteAddress: InetSocketAddress, localAddress: Option[InetSocketAddress], handler: ActorRef)(implicit factory: ActorRefFactory): ActorRef =
    factory.actorOf(Props(classOf[ConnectionHandler], remoteAddress, handler))
} 

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

import java.net.InetSocketAddress

import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
import akka.io.{IO, Tcp}
import akka.util.ByteString
import com.mpakhomov.actors.CandlestickAggregatorActor.GetDataForLastNMinutes
import com.mpakhomov.actors.UpstreamClientActor.CloseCommand
import com.typesafe.config.ConfigFactory

class UpstreamClientActor(remote: InetSocketAddress, eventProcessor: ActorRef) extends Actor with ActorLogging {

  import Tcp._
  import context.system

  IO(Tcp) ! Connect(remote)

  def receive = {
    case CommandFailed(_: Connect) =>
      log.error("connect failed")
      context stop self

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

object UpstreamClientActor {

  // messages
  case object CloseCommand

  def props(remote: InetSocketAddress, replies: ActorRef) =
    Props(classOf[UpstreamClientActor], remote, replies)
}