MicroKernel.scala

/*
 * Copyright (c) 2010, Preston Skupinski <skupinsk@cse.msu.edu>
 *
 * Permission to use, copy, modify, and/or distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE
 */ 

import scala.actors._
import Actor._
import scala.collection.mutable._
import scala.concurrent.ops._
import java.nio.channels._
import java.nio._
import java.nio.charset._
import java.io.IOException

case class Tick(a: Actor)
case class PlayerCommand(key: SelectionKey, s: String)
case class TelnetCommand(key: SelectionKey, s: String)
case class CancelKey(key: SelectionKey)
case class AddKey(key: SelectionKey)
case class Quit()
case class Done(a: Actor)

object MicroKernel {
  def main(args: Array[String]): Unit = {
    PlayerManager.start()
    NetworkManager.start()
    heartBeatLoop()  
    0 
  }

  def heartBeatLoop() {
    var continue = true
    while(continue) {
      val startTime = System.currentTimeMillis()

      PlayerManager ! Tick(self)
      NetworkManager ! Tick(self)
      // Send ticks to all of the other managers...

      // All of the actors are busy processing changes and sending each other
      // messages to be processed next turn while this thread moves to wait
      // for a done response.

      var playersDone = false
      var networkDone = false
      while((!playersDone)&&(!networkDone)) {
        receive {
          case Done(PlayerManager) => playersDone = true
          case Done(NetworkManager) => networkDone = true 
        }
      }
 
      val diff = System.currentTimeMillis() - startTime
      if(diff<100) {
        println("Diff: " + (100-diff) + " milliseconds " + Thread.currentThread)
        Thread.sleep(100-diff) 
      } else if(diff>100) {
        println("WARNING: diff>100, something is taking way too long!")
      }
    }
  }
}

object NetworkManager extends Actor {
  val charset = Charset.forName("ISO-8859-1")
  val encoder = charset.newEncoder
  val decoder = charset.newDecoder
  val selector = Selector.open
  val server = ServerSocketChannel.open
  server.socket.bind(new java.net.InetSocketAddress(8988))
  server.configureBlocking(false)
  val serverKey = server.register(selector, SelectionKey.OP_ACCEPT)
  val buffer = ByteBuffer.allocate(1024)

  def act() {
    var continue = true
    loopWhile(continue) {
      react {
        case Tick(caller) =>
          // Check for changes in regards to any new clients to accept, who 
          // should be removed and any messages sent to the server. 
          println("NetworkManager - Tick! " + Thread.currentThread)
          ManageNetworkClients

          caller ! Done(self) // Let the heart know we're done.
        case Quit() => 
          println("NetworkManager - Quit! " + Thread.currentThread)
          continue = false
      }
    }
  }

  def ManageNetworkClients() {
    // Accept new clients(sending new keys to PlayerManager), ditches clients 
    // who disconnected(notifying PlayerManager that the keys are cancelled), 
    // and sends player requests to the PlayerManager.
    selector.selectNow
    val keys = selector.selectedKeys
    val iterator = keys.iterator
    while(iterator.hasNext) {
      val key = iterator.next
      if(key eq serverKey) {
        if(key.isAcceptable) {
          val client = server.accept
          if(client ne null)  {
            client.configureBlocking(false)
            val clientKey = client.register(selector, SelectionKey.OP_READ)
            PlayerManager ! AddKey(clientKey)
          }
        }
      } else {
        val client = key.channel.asInstanceOf[SocketChannel]
        try {
          if(key.isReadable()) {
            val numRead = client.read(buffer)

            if(numRead != -1) {
              buffer.flip
              val request = decoder.decode(buffer).toString()
              buffer.clear
              if(request.size!=0) {
                // If multiple messages made it in during the turn then split 
                // them up into separate messages.  We preserve the carriage 
                // return and split on "\n" because lines are ended with "\r\n".
                // Lines will end in \r with how this is split and a message
                // containing just \r means that the user only hit enter for
                // their message.  Example of why this is necessary:
                // Say \r\n\r\nquit\r\n is received during the course of one 
                // turn, to properly handle that we would need to send two new
                // prompts and then execute the quit command.
                // DOES NOT PROPERLY SEPARATE TELNET PROTOCOL STUFF YET.
                val s = request.split("""\n""")
                s.foreach { str =>
                  PlayerManager ! PlayerCommand(key, str)
                } 
              }
            } else {
              client.close 
              key.cancel
              PlayerManager ! CancelKey(key)
            }
          }
        } catch {
          case ex: CancelledKeyException => // Ignore it.
          case ex: ClosedChannelException => // Ignore it.
        }
      }
    }
  }
}

object PlayerManager extends Actor {
  var queue = new Queue[(Int, (SelectionKey, String))]
  var keySet = new LinkedHashSet[SelectionKey]()
  var newKeySet = new LinkedHashSet[SelectionKey]()
  var cancelledKeySet = new LinkedHashSet[SelectionKey]()

  def handleQueue(queue: Queue[(Int, (SelectionKey, String))]) {
    var i = 0
    for(tup <- queue) {
      println(i + ": " + tup)
      if(tup._1==0) {
        // Message from player to be executed.
        val keyMessage = tup._2
        println("Player message received from " + keyMessage._1 + " \"" +
		keyMessage._2.trim + "\"")
        // PlayerCommand could possibly parse the message and get the command
        // requested through some getCommand message and that could be passed
        // here rather than a string(once I write out the base command trait 
        // and some commands for it).  The below is pretty ugly as-is and 
        // should be considered temporary.
        if(keyMessage._2.trim=="quit")
          cancelKey(keyMessage._1)
        else 
          write(keyMessage._1, keyMessage._2+"\n")
      }
      i += 1
    }  
  }

  def write(s: SelectionKey, str: String) {
    val chan = s.channel.asInstanceOf[SocketChannel]
    try {
      chan.write(NetworkManager.encoder.encode(CharBuffer.wrap(str)))
    } catch {
      case ex: Exception =>
    }
  }

  def cancelKey(s: SelectionKey) {
    val client = s.channel.asInstanceOf[SocketChannel]
    client.close
    s.cancel
  }

  def act() {
    var continue = true
    loopWhile(continue) {
      react {
        case Tick(caller) =>
          // Process data received during the previous turn and prepare for 
          // new data.
          println("PlayerManager - Tick! " + Thread.currentThread)
 
          // Handle the job queue.
          val oldQueue = queue.clone
          queue.clear
          handleQueue(oldQueue)

          // Remove cancelled keys and add new keys.
          keySet --= cancelledKeySet
          keySet ++= newKeySet
          newKeySet.clear
          cancelledKeySet.clear

          caller ! Done(self) // Let the heart know that we're done.
        case Quit() => 
          println("PlayerManager - Quit! " + Thread.currentThread)
          continue = false
        case AddKey(s) =>
          println("PlayerManager - Add! " + Thread.currentThread)
          newKeySet += s
        case CancelKey(s) =>
          println("PlayerManager - Cancel! " + Thread.currentThread)
          println("Got a key to cancel: " + s)
          cancelKey(s)
          cancelledKeySet += s
        case PlayerCommand(key, str) => 
          println("PlayerManager - Job! " + Thread.currentThread)
          queue enqueue ((0, (key, str)))
      }
    } 
  }  
}