pekko/kernel/src/main/scala/ActiveObject.scala

/**
 * Copyright (C) 2009 Scalable Solutions.
 */

package se.scalablesolutions.akka.kernel

import java.util.{List => JList, ArrayList}
import java.lang.reflect.{Method, Field, InvocationHandler, Proxy, InvocationTargetException}
import java.lang.annotation.Annotation

//import voldemort.client.{SocketStoreClientFactory, StoreClient, StoreClientFactory}
//import voldemort.versioning.Versioned

sealed class ActiveObjectException(msg: String) extends RuntimeException(msg)
class ActiveObjectInvocationTimeoutException(msg: String) extends ActiveObjectException(msg)

object Annotations {
  import se.scalablesolutions.akka.annotation._
  val transactional = classOf[transactional]
  val oneway =        classOf[oneway]
  val immutable =     classOf[immutable]
  val state =         classOf[state]
}

/**
 * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
 */
class ActiveObjectFactory {
  def newInstance[T](intf: Class[_], proxy: ActiveObjectProxy): T = ActiveObject.newInstance(intf, proxy)

  def supervise(restartStrategy: RestartStrategy, components: JList[Worker]): Supervisor =
    ActiveObject.supervise(restartStrategy, components.toArray.toList.asInstanceOf[List[Worker]])  
}

/**
 * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
 */
object ActiveObject {
   private[kernel] val threadBoundTx: ThreadLocal[Option[Transaction]] = {
     val tl = new ThreadLocal[Option[Transaction]]
     tl.set(None)
     tl
   }

   def newInstance[T](intf: Class[_], proxy: ActiveObjectProxy): T = {
     Proxy.newProxyInstance(
       intf.getClassLoader,
       Array(intf),
       proxy).asInstanceOf[T]
   }

  def newInstance[T](intf: Class[_], target: AnyRef, timeout: Int): T = {
    val proxy = new ActiveObjectProxy(intf, target.getClass, timeout)
    proxy.setTargetInstance(target)
    supervise(proxy)
    newInstance(intf, proxy)
  }

  def supervise(restartStrategy: RestartStrategy, components: List[Worker]): Supervisor = {
    object factory extends SupervisorFactory {
      override def getSupervisorConfig = SupervisorConfig(restartStrategy, components)
    }
    val supervisor = factory.newSupervisor
    supervisor ! se.scalablesolutions.akka.kernel.Start
    supervisor
  }

  private def supervise(proxy: ActiveObjectProxy): Supervisor = 
    supervise(
      RestartStrategy(OneForOne, 5, 1000),
      Worker(
        proxy.server,
        LifeCycle(Permanent, 100))
      :: Nil)
}

/**
 * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
 */
class ActiveObjectProxy(val intf: Class[_], val target: Class[_], val timeout: Int) extends InvocationHandler {
  import ActiveObject.threadBoundTx
  private[this] var activeTx: Option[Transaction] = None

  private var targetInstance: AnyRef = _
  private[kernel] def setTargetInstance(instance: AnyRef) = {
    targetInstance = instance
    val (maps, vectors, refs) = getTransactionalItemsFor(targetInstance) 
    server.transactionalMaps = maps
    server.transactionalVectors = vectors
    server.transactionalRefs = refs
  }

  private[kernel] val server = new GenericServerContainer(target.getName, () => new Dispatcher(target.getName))
  server.setTimeout(timeout)

  def invoke(proxy: AnyRef, m: Method, args: Array[AnyRef]): AnyRef = {
    if (m.isAnnotationPresent(Annotations.transactional)) {
      // FIXME: check if we are already in a transaction if so NEST (set parent)
      val newTx = new Transaction
      newTx.begin(server)
      threadBoundTx.set(Some(newTx))
    }
    val cflowTx = threadBoundTx.get
    activeTx match {
      case Some(tx) =>
        if (cflowTx.isDefined && cflowTx.get != tx) {
          // new tx in scope; try to commit
          tx.commit(server)
          threadBoundTx.set(None)
          activeTx = None
        } 
      case None =>
        if (cflowTx.isDefined) {
          val currentTx = cflowTx.get
          currentTx.join(server)
          activeTx = Some(currentTx)
        }
    }
    activeTx = threadBoundTx.get
    invoke(Invocation(m, args, targetInstance, activeTx))
  }

  private def invoke(invocation: Invocation): AnyRef =  {
    val result: AnyRef = 
      if (invocation.method.isAnnotationPresent(Annotations.oneway)) server ! invocation
      else {
        val result: ErrRef[AnyRef] =
          server !!! (invocation, {
            var ref = ErrRef(activeTx)
            ref() = throw new ActiveObjectInvocationTimeoutException("Invocation to active object [" + targetInstance.getClass.getName + "] timed out after " + timeout + " milliseconds")
            ref
          })
        try {
          result()
        } catch {
          case e => 
            rollback(result.tx)
            throw e
        }
      }
    // FIXME: clear threadBoundTx on successful commit
    if (activeTx.isDefined) activeTx.get.precommit(server)
    result
  }

  private def rollback(tx: Option[Transaction]) = tx match {
    case None => {} // no tx; nothing to do
    case Some(tx) =>
      tx.rollback(server)
      threadBoundTx.set(Some(tx))
  }

  private def getTransactionalItemsFor(targetInstance: AnyRef): 
    Tuple3[List[TransactionalMap[_, _]], List[TransactionalVector[_]], List[TransactionalRef[_]]] = {
    require(targetInstance != null)
    var maps:    List[TransactionalMap[_, _]] = Nil 
    var vectors: List[TransactionalVector[_]] = Nil 
    var refs:    List[TransactionalRef[_]] = Nil 
    for {
      field <- target.getDeclaredFields.toArray.toList.asInstanceOf[List[Field]]
      fieldType = field.getType
      if fieldType == classOf[TransactionalMap[_, _]] || 
        fieldType  == classOf[TransactionalVector[_]] || 
        fieldType  == classOf[TransactionalRef[_]]
      txItem = {
        field.setAccessible(true)
        field.get(targetInstance)
      }
      if txItem != null
    } {
      if (txItem.isInstanceOf[TransactionalMap[_, _]])      maps ::= txItem.asInstanceOf[TransactionalMap[_, _]]
      else if (txItem.isInstanceOf[TransactionalVector[_]]) vectors ::= txItem.asInstanceOf[TransactionalVector[_]]
      else if (txItem.isInstanceOf[TransactionalRef[_]])    refs ::= txItem.asInstanceOf[TransactionalRef[_]]
    }
    (maps, vectors, refs)
  }
}

/**
 * Generic GenericServer managing Invocation dispatch, transaction and error management.
 *
 * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
 */
private[kernel] class Dispatcher(val targetName: String) extends GenericServer {
  override def body: PartialFunction[Any, Unit] = {
    case invocation: Invocation =>
      val tx = invocation.tx
      ActiveObject.threadBoundTx.set(tx)
      try {
        reply(ErrRef(invocation.invoke, tx))
      } catch {
        case e: InvocationTargetException =>
          val ref = ErrRef(tx); ref() = throw e.getTargetException; reply(ref)
        case e =>
          val ref = ErrRef(tx); ref() = throw e; reply(ref)
      }
    case 'exit =>
      exit; reply()
    case unexpected =>
      throw new ActiveObjectException("Unexpected message [" + unexpected + "] to [" + this + "] from [" + sender + "]")
  }
  
  override def toString(): String = "GenericServer[" + targetName + "]"
}

/**
 * Represents a snapshot of the current invocation.
 * 
 * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
 */
private[kernel] case class Invocation(val method: Method,
                                      val args: Array[AnyRef],
                                      val target: AnyRef,
                                      val tx: Option[Transaction]) {
  method.setAccessible(true)

  def invoke: AnyRef = synchronized {
    method.invoke(target, args:_*)
  }

  override def toString: String = synchronized {
    "Invocation [method: " + method.getName + ", args: " + argsToString(args) + ", target: " + target + "]"
  }
  
  override def hashCode(): Int = synchronized {
    var result = HashCode.SEED
    result = HashCode.hash(result, method)
    result = HashCode.hash(result, args)
    result = HashCode.hash(result, target)
    result
  }

  override def equals(that: Any): Boolean = synchronized {
    that != null &&
    that.isInstanceOf[Invocation] &&
    that.asInstanceOf[Invocation].method == method &&
    that.asInstanceOf[Invocation].target == target &&
    isEqual(that.asInstanceOf[Invocation].args, args)
  }

  private[this] def isEqual(a1: Array[Object], a2: Array[Object]): Boolean =
    (a1 == null && a2 == null) ||
    (a1 != null && 
     a2 != null && 
     a1.size == a2.size && 
     a1.zip(a2).find(t => t._1 == t._2).isDefined)

  private[this] def argsToString(array: Array[Object]): String =
    array.foldLeft("(")(_ + " " + _) + ")" 
}