import java.net.InetSocketAddress

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

import scala.collection.mutable

object SocketProducer {
  //Needed to create an Actor with a given configuration
  def props(host: String, port: Int, messages: mutable.MutableList[String]): Props =
    Props(new SocketProducer(host,port,messages))

}

class SocketProducer(host: String, port: Int, messages: mutable.MutableList[String]) extends Actor with ActorLogging{

  import SocketProducer._

  // function that allows us to do some logic before the Actor is up
  override def preStart(): Unit = {
    log.info("Starting socket PRODUCER actor with following config {}:{}",host,port)
  }

  // function that allows us to do some logic once the Actor is down
  override def postStop(): Unit = {
    log.info("Stopped socket PRODUCER actor")
  }

}

import java.net.InetSocketAddress

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

import scala.collection.mutable

object SocketProducer {

  //Needed to create an Actor with a given configuration
  def props(host: String, port: Int, messages: mutable.MutableList[String]): Props =
    Props(new SocketProducer(host,port,messages))

}

class SocketProducer(host: String, port: Int, messages: mutable.MutableList[String]) extends Actor with ActorLogging{

  import SocketProducer._

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

  // Actor that manages the low level communication with the socket
  // this Actor can send messages to out SocketProducer Actor, because
  // a reference to the SocketProducer Actor is implicit passed
  // when we invoke IO(Tcp)
  IO(Tcp) ! Connect(new InetSocketAddress(host,port))

  // function that allows us to do some logic before the Actor is up
  override def preStart(): Unit = {
    log.info("Starting socket PRODUCER actor with following config {}:{}",host,port)
  }

  // function that allows us to do some logic once the Actor is down
  override def postStop(): Unit = {
    log.info("Stopped socket PRODUCER actor")
  }

  override def receive = {
    case CommandFailed(_: Connect) =>
      context stop self

    //Needed to establish a connection
    case c @ Connected(remote, local) =>

      //opens connection
      val connection = sender()
      //establish connection
      connection ! Register(self)
      //sends messages
      messages.foreach(message => {
        log.info("Trying to send message: "+message)
        connection ! Write(ByteString(message))
      })
      //close connection
      connection ! Close

      //this is some black magic that will be explained later
      //but the idea it's that it manages the socket status
      context become {
        case data: ByteString =>
          connection ! Write(data)
        case CommandFailed(w: Write) =>
          // O/S buffer was full
          log.warning("write failed")
        case Received(data) =>
        case "close" =>
          connection ! Close
        case _: ConnectionClosed =>
          log.warning("connection closed")
          //kills the Actor
          context stop self
      }

    //if the Actor receives a message that it doesn't understand, it sends a warning through the logger
    case x @ _ => log.warning("Something else is up. ---> " + x.toString)

  }
}

import actors.SocketProducer
import akka.actor.ActorSystem


import scala.collection.mutable

object Main {

  def main(args: Array[String]): Unit = {

    //Creates and Actor system in wich will reside/live our actor
    val actorSystem = ActorSystem.create("MyActorSystem")

    //data that we want to send with our actor
    val list = mutable.MutableList("1","2","Three","0100").map(_+"\n")
    //I append a \n to each element in the list for visibility

    //creates and runs the actor
    val actor = actorSystem.actorOf(SocketProducer.props("localhost",9000,list))


  }

}

import java.net.InetSocketAddress

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

import scala.collection.mutable

object SocketConsumer {

  //Needed to create an Actor with a given configuration
  def props(port: Int): Props = Props(new SocketConsumer(port))

}

class SocketConsumer(port: Int) extends Actor with ActorLogging{

  import SocketConsumer._

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

  // Actor that manages the low level communication with the socket
  // this Actor can send messages to out SocketProducer Actor, because
  // a reference to the SocketProducer Actor is implicit passed
  // when we invoke IO(Tcp)
  IO(Tcp) ! Bind(self, new InetSocketAddress("localhost", port))

  // function that allows us to do some logic before the Actor is up
  override def preStart(): Unit = {
    log.info("Starting socket CONSUMER actor in port {}",port)
  }

  // function that allows us to do some logic once the Actor is down
  override def postStop(): Unit = {
    log.info("Stopped socket CONSUMER actor")
  }

  override def receive = {

    case b @ Bound(localAddress) =>
      log.info("CONSUMER bound to: "+b)

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

    case c @ Connected(remote, local) =>
      val connection = sender()
      connection ! Register(self)

    case Received(message) => log.info("Received: \n"+ message.decodeString("UTF8"))
    case PeerClosed     => context stop self

    //if the Actor receives a message that it doesn't understand, it sends a warning through the logger
    case x @ _ => log.warning("Something else is up. ---> " + x.toString)

  }
}

import actors.{SocketConsumer, SocketProducer}
import akka.actor.ActorSystem

import scala.collection.mutable

object Main {

  def main(args: Array[String]): Unit = {

    //Creates and Actor system in wich will reside/live our actor
    val actorSystem = ActorSystem.create("MyActorSystem")

    //data that we want to send with our actor
    val list = mutable.MutableList("1","2","Three","0100").map(_+"\n")

    //creates and runs the actor
    val actorConsumer = actorSystem.actorOf(SocketConsumer.props(9000))
    val actorProducer = actorSystem.actorOf(SocketProducer.props("localhost",9000,list))

  }

}