diff --git a/akka-actor/src/main/scala/actor/Actor.scala b/akka-actor/src/main/scala/actor/Actor.scala index dc74028964..6667f7fbc4 100644 --- a/akka-actor/src/main/scala/actor/Actor.scala +++ b/akka-actor/src/main/scala/actor/Actor.scala @@ -17,17 +17,6 @@ import scala.reflect.BeanProperty import akka.util. {ReflectiveAccess, Logging, Duration} import akka.japi.Procedure -/** - * Implements the Transactor abstraction. E.g. a transactional actor. - *

- * Equivalent to invoking the makeTransactionRequired method in the body of the ActorJonas Bonér - */ -trait Transactor extends Actor { - self.makeTransactionRequired -} - /** * Extend this abstract class to create a remote actor. *

@@ -246,7 +235,7 @@ object Actor extends Logging { * Here you find functions like: * - !, !!, !!! and forward * - link, unlink, startLink, spawnLink etc - * - makeTransactional, makeRemote etc. + * - makeRemote etc. * - start, stop * - etc. * diff --git a/akka-actor/src/main/scala/actor/ActorRef.scala b/akka-actor/src/main/scala/actor/ActorRef.scala index 09477e9c03..adf7a1cde7 100644 --- a/akka-actor/src/main/scala/actor/ActorRef.scala +++ b/akka-actor/src/main/scala/actor/ActorRef.scala @@ -7,17 +7,10 @@ package akka.actor import akka.dispatch._ import akka.config.Config._ import akka.config.Supervision._ -import akka.stm.global._ -import akka.stm.TransactionManagement._ -import akka.stm.{ TransactionManagement, TransactionSetAbortedException } import akka.AkkaException import akka.util._ import ReflectiveAccess._ -import org.multiverse.api.ThreadLocalTransaction._ -import org.multiverse.commitbarriers.CountDownCommitBarrier -import org.multiverse.api.exceptions.DeadTransactionException - import java.net.InetSocketAddress import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} import java.util.concurrent.locks.ReentrantLock @@ -39,10 +32,6 @@ private[akka] object ActorRefInternals extends Logging { object RUNNING extends StatusType object BEING_RESTARTED extends StatusType object SHUTDOWN extends StatusType - - case class TransactorConfig(factory: Option[TransactionFactory] = None, config: TransactionConfig = DefaultGlobalTransactionConfig) - val DefaultTransactorConfig = TransactorConfig() - val NoTransactionConfig = TransactorConfig() } @@ -78,7 +67,7 @@ private[akka] object ActorRefInternals extends Logging { * * @author Jonas Bonér */ -trait ActorRef extends ActorRefShared with TransactionManagement with java.lang.Comparable[ActorRef] { scalaRef: ScalaActorRef => +trait ActorRef extends ActorRefShared with java.lang.Comparable[ActorRef] { scalaRef: ScalaActorRef => //Reuse same logger import Actor.log @@ -174,23 +163,6 @@ trait ActorRef extends ActorRefShared with TransactionManagement with java.lang. @volatile protected[akka] var hotswap = Stack[PartialFunction[Any, Unit]]() - /** - * User overridable callback/setting. - *

- * Set to true if messages should have REQUIRES_NEW semantics, e.g. a new transaction should - * start if there is no one running, else it joins the existing transaction. - */ - @volatile - protected[akka] var transactorConfig = ActorRefInternals.NoTransactionConfig - - /** - * Returns true if this Actor is a Transactor - */ - def isTransactor: Boolean = { - val c = transactorConfig - (c ne ActorRefInternals.NoTransactionConfig) && (c ne null) //Could possibly be null if called before var init - } - /** * This is a reference to the message currently being processed by the actor */ @@ -412,34 +384,6 @@ trait ActorRef extends ActorRefShared with TransactionManagement with java.lang. */ def makeRemote(address: InetSocketAddress): Unit - /** - * Invoking 'makeTransactionRequired' means that the actor will **start** a new transaction if non exists. - * However, it will always participate in an existing transaction. - */ - def makeTransactionRequired(): Unit - - /** - * Sets the transaction configuration for this actor. Needs to be invoked before the actor is started. - */ - def transactionConfig_=(config: TransactionConfig): Unit - - /** - * Akka Java API - * Sets the transaction configuration for this actor. Needs to be invoked before the actor is started. - */ - def setTransactionConfig(config: TransactionConfig): Unit = transactionConfig = config - - /** - * Get the transaction configuration for this actor. - */ - def transactionConfig: TransactionConfig - - /** - * Akka Java API - * Get the transaction configuration for this actor. - */ - def getTransactionConfig(): TransactionConfig = transactionConfig - /** * Returns the home address and port for this actor. */ @@ -664,7 +608,6 @@ class LocalActorRef private[akka] ( __id: String, __hostname: String, __port: Int, - __isTransactor: Boolean, __timeout: Long, __receiveTimeout: Option[Long], __lifeCycle: LifeCycle, @@ -675,7 +618,6 @@ class LocalActorRef private[akka] ( _uuid = __uuid id = __id homeAddress = (__hostname, __port) - transactorConfig = if (__isTransactor) ActorRefInternals.DefaultTransactorConfig else ActorRefInternals.NoTransactionConfig timeout = __timeout receiveTimeout = __receiveTimeout lifeCycle = __lifeCycle @@ -737,33 +679,6 @@ class LocalActorRef private[akka] ( "Can't make a running actor remote. Make sure you call 'makeRemote' before 'start'.") } - /** - * Invoking 'makeTransactionRequired' means that the actor will **start** a new transaction if non exists. - * However, it will always participate in an existing transaction. - */ - def makeTransactionRequired() = guard.withGuard { - if (!isRunning || isBeingRestarted) { - if (transactorConfig eq ActorRefInternals.NoTransactionConfig) - transactorConfig = ActorRefInternals.DefaultTransactorConfig - } - else throw new ActorInitializationException( - "Can not make actor transaction required after it has been started") - } - - /** - * Sets the transaction configuration for this actor. Needs to be invoked before the actor is started. - */ - def transactionConfig_=(configuration: TransactionConfig) = guard.withGuard { - if (!isRunning || isBeingRestarted) transactorConfig = transactorConfig.copy(config = configuration) - else throw new ActorInitializationException( - "Cannot set transaction configuration for actor after it has been started") - } - - /** - * Get the transaction configuration for this actor. - */ - def transactionConfig: TransactionConfig = transactorConfig.config - /** * Set the contact address for this actor. This is used for replying to messages * sent asynchronously when no reply channel exists. @@ -784,8 +699,6 @@ class LocalActorRef private[akka] ( "Can't restart an actor that has been shut down with 'stop' or 'exit'") if (!isRunning) { dispatcher.attach(this) - if (isTransactor) - transactorConfig = transactorConfig.copy(factory = Some(TransactionFactory(transactorConfig.config, id))) _status = ActorRefInternals.RUNNING @@ -806,7 +719,6 @@ class LocalActorRef private[akka] ( receiveTimeout = None cancelReceiveTimeout dispatcher.detach(this) - transactorConfig = transactorConfig.copy(factory = None) _status = ActorRefInternals.SHUTDOWN actor.postStop ActorRegistry.unregister(this) @@ -948,13 +860,11 @@ class LocalActorRef private[akka] ( protected[akka] def supervisor_=(sup: Option[ActorRef]): Unit = _supervisor = sup protected[akka] def postMessageToMailbox(message: Any, senderOption: Option[ActorRef]): Unit = { - joinTransaction(message) - if (remoteAddress.isDefined && isRemotingEnabled) { RemoteClientModule.send[Any]( message, senderOption, None, remoteAddress.get, timeout, true, this, None, ActorType.ScalaActor) } else { - val invocation = new MessageInvocation(this, message, senderOption, None, transactionSet.get) + val invocation = new MessageInvocation(this, message, senderOption, None) dispatcher dispatchMessage invocation } } @@ -964,7 +874,6 @@ class LocalActorRef private[akka] ( timeout: Long, senderOption: Option[ActorRef], senderFuture: Option[CompletableFuture[T]]): CompletableFuture[T] = { - joinTransaction(message) if (remoteAddress.isDefined && isRemotingEnabled) { val future = RemoteClientModule.send[T]( @@ -974,7 +883,7 @@ class LocalActorRef private[akka] ( } else { val future = if (senderFuture.isDefined) senderFuture else Some(new DefaultCompletableFuture[T](timeout)) val invocation = new MessageInvocation( - this, message, senderOption, future.asInstanceOf[Some[CompletableFuture[Any]]], transactionSet.get) + this, message, senderOption, future.asInstanceOf[Some[CompletableFuture[Any]]]) dispatcher dispatchMessage invocation future.get } @@ -1151,61 +1060,18 @@ class LocalActorRef private[akka] ( a } - private def joinTransaction(message: Any) = if (isTransactionSetInScope) { - import org.multiverse.api.ThreadLocalTransaction - val oldTxSet = getTransactionSetInScope - val currentTxSet = if (oldTxSet.isAborted || oldTxSet.isCommitted) { - clearTransactionSet - createNewTransactionSet - } else oldTxSet - Actor.log.trace("Joining transaction set [" + currentTxSet + - "];\n\tactor " + toString + - "\n\twith message [" + message + "]") - val mtx = ThreadLocalTransaction.getThreadLocalTransaction - if ((mtx eq null) || mtx.getStatus.isDead) currentTxSet.incParties - else currentTxSet.incParties(mtx, 1) - } - private def dispatch[T](messageHandle: MessageInvocation) = { Actor.log.trace("Invoking actor with message: %s\n", messageHandle) val message = messageHandle.message //serializeMessage(messageHandle.message) - val isXactor = isTransactor - var topLevelTransaction = false - val txSet: Option[CountDownCommitBarrier] = - if (messageHandle.transactionSet.isDefined) messageHandle.transactionSet - else { - topLevelTransaction = true // FIXME create a new internal atomic block that can wait for X seconds if top level tx - if (isXactor) { - Actor.log.trace("Creating a new transaction set (top-level transaction)\n\tfor actor " + toString + - "\n\twith message " + messageHandle) - Some(createNewTransactionSet) - } else None - } - setTransactionSet(txSet) try { cancelReceiveTimeout // FIXME: leave this here? - if (isXactor) { - val txFactory = transactorConfig.factory.getOrElse(DefaultGlobalTransactionFactory) - atomic(txFactory) { - actor(message) - setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit - } - } else { - actor(message) - setTransactionSet(txSet) // restore transaction set to allow atomic block to do commit - } + actor(message) } catch { - case e: DeadTransactionException => - handleExceptionInDispatch( - new TransactionSetAbortedException("Transaction set has been aborted by another participant"), - message, topLevelTransaction) case e: InterruptedException => {} // received message while actor is shutting down, ignore - case e => handleExceptionInDispatch(e, message, topLevelTransaction) + case e => handleExceptionInDispatch(e, message) } finally { - clearTransaction - if (topLevelTransaction) clearTransactionSet checkReceiveTimeout // Reschedule receive timeout } } @@ -1226,26 +1092,14 @@ class LocalActorRef private[akka] ( true } - private def handleExceptionInDispatch(reason: Throwable, message: Any, topLevelTransaction: Boolean) = { + private def handleExceptionInDispatch(reason: Throwable, message: Any) = { Actor.log.error(reason, "Exception when invoking \n\tactor [%s] \n\twith message [%s]", this, message) //Prevent any further messages to be processed until the actor has been restarted dispatcher.suspend(this) - // abort transaction set - if (isTransactionSetInScope) { - val txSet = getTransactionSetInScope - if (!txSet.isCommitted) { - Actor.log.debug("Aborting transaction set [%s]", txSet) - txSet.abort - } - } - senderFuture.foreach(_.completeWithException(reason)) - clearTransaction - if (topLevelTransaction) clearTransactionSet - if (supervisor.isDefined) notifySupervisorWithMessage(Exit(this, reason)) else { lifeCycle match { @@ -1298,7 +1152,6 @@ class LocalActorRef private[akka] ( actor.preStart // run actor preStart Actor.log.trace("[%s] has started", toString) ActorRegistry.register(this) - clearTransactionSet // clear transaction set that might have been created if atomic block has been used within the Actor constructor body } /* @@ -1395,9 +1248,6 @@ private[akka] case class RemoteActorRef private[akka] ( def actorClass: Class[_ <: Actor] = unsupported def dispatcher_=(md: MessageDispatcher): Unit = unsupported def dispatcher: MessageDispatcher = unsupported - def makeTransactionRequired: Unit = unsupported - def transactionConfig_=(config: TransactionConfig): Unit = unsupported - def transactionConfig: TransactionConfig = unsupported def makeRemote(hostname: String, port: Int): Unit = unsupported def makeRemote(address: InetSocketAddress): Unit = unsupported def homeAddress_=(address: InetSocketAddress): Unit = unsupported diff --git a/akka-actor/src/main/scala/actor/UntypedActor.scala b/akka-actor/src/main/scala/actor/UntypedActor.scala index b31077c8f3..1c51fa47f1 100644 --- a/akka-actor/src/main/scala/actor/UntypedActor.scala +++ b/akka-actor/src/main/scala/actor/UntypedActor.scala @@ -5,7 +5,6 @@ package akka.actor import akka.dispatch._ -import akka.stm.global._ import akka.config.Supervision._ import akka.japi.Procedure @@ -84,15 +83,6 @@ abstract class UntypedActor extends Actor { def onReceive(message: Any): Unit } -/** - * Implements the Transactor abstraction. E.g. a transactional UntypedActor. - * - * @author Jonas Bonér - */ -abstract class UntypedTransactor extends UntypedActor { - self.makeTransactionRequired -} - /** * Factory closure for an UntypedActor, to be used with 'UntypedActor.actorOf(factory)'. * diff --git a/akka-actor/src/main/scala/config/SupervisionConfig.scala b/akka-actor/src/main/scala/config/SupervisionConfig.scala index 9f5af03ddf..f3d6e1ada9 100644 --- a/akka-actor/src/main/scala/config/SupervisionConfig.scala +++ b/akka-actor/src/main/scala/config/SupervisionConfig.scala @@ -105,7 +105,6 @@ object Supervision { val target: Class[_], val lifeCycle: LifeCycle, val timeout: Long, - val transactionRequired: Boolean, _dispatcher: MessageDispatcher, // optional _remoteAddress: RemoteAddress // optional ) extends Server { @@ -114,48 +113,24 @@ object Supervision { val remoteAddress: Option[RemoteAddress] = Option(_remoteAddress) def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long) = - this(null: Class[_], target, lifeCycle, timeout, false, null.asInstanceOf[MessageDispatcher], null: RemoteAddress) + this(null: Class[_], target, lifeCycle, timeout, null: MessageDispatcher, null: RemoteAddress) def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long) = - this(intf, target, lifeCycle, timeout, false, null.asInstanceOf[MessageDispatcher], null: RemoteAddress) + this(intf, target, lifeCycle, timeout, null: MessageDispatcher, null: RemoteAddress) def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, dispatcher: MessageDispatcher) = - this(intf, target, lifeCycle, timeout, false, dispatcher, null) + this(intf, target, lifeCycle, timeout, dispatcher, null: RemoteAddress) def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, dispatcher: MessageDispatcher) = - this(null: Class[_], target, lifeCycle, timeout, false, dispatcher, null:RemoteAddress) + this(null: Class[_], target, lifeCycle, timeout, dispatcher, null: RemoteAddress) def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, remoteAddress: RemoteAddress) = - this(intf, target, lifeCycle, timeout, false, null: MessageDispatcher, remoteAddress) + this(intf, target, lifeCycle, timeout, null: MessageDispatcher, remoteAddress) def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, remoteAddress: RemoteAddress) = - this(null: Class[_], target, lifeCycle, timeout, false, null, remoteAddress) - - def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, dispatcher: MessageDispatcher, remoteAddress: RemoteAddress) = - this(intf, target, lifeCycle, timeout, false, dispatcher, remoteAddress) + this(null: Class[_], target, lifeCycle, timeout, null: MessageDispatcher, remoteAddress) def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, dispatcher: MessageDispatcher, remoteAddress: RemoteAddress) = - this(null: Class[_], target, lifeCycle, timeout, false, dispatcher, remoteAddress) - - def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean) = - this(intf, target, lifeCycle, timeout, transactionRequired, null: MessageDispatcher, null: RemoteAddress) - - def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean) = - this(null: Class[_], target, lifeCycle, timeout, transactionRequired, null: MessageDispatcher, null: RemoteAddress) - - def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean, dispatcher: MessageDispatcher) = - this(intf, target, lifeCycle, timeout, transactionRequired, dispatcher, null: RemoteAddress) - - def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean, dispatcher: MessageDispatcher) = - this(null: Class[_], target, lifeCycle, timeout, transactionRequired, dispatcher, null: RemoteAddress) - - def this(intf: Class[_], target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean, remoteAddress: RemoteAddress) = - this(intf, target, lifeCycle, timeout, transactionRequired, null: MessageDispatcher, remoteAddress) - - def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean, remoteAddress: RemoteAddress) = - this(null: Class[_], target, lifeCycle, timeout, transactionRequired, null: MessageDispatcher, remoteAddress) - - def this(target: Class[_], lifeCycle: LifeCycle, timeout: Long, transactionRequired: Boolean, dispatcher: MessageDispatcher, remoteAddress: RemoteAddress) = - this(null: Class[_], target, lifeCycle, timeout, transactionRequired, dispatcher, remoteAddress) + this(null: Class[_], target, lifeCycle, timeout, dispatcher, remoteAddress) } } diff --git a/akka-actor/src/main/scala/dispatch/MessageHandling.scala b/akka-actor/src/main/scala/dispatch/MessageHandling.scala index 33a7a62af3..467bccd13e 100644 --- a/akka-actor/src/main/scala/dispatch/MessageHandling.scala +++ b/akka-actor/src/main/scala/dispatch/MessageHandling.scala @@ -4,8 +4,6 @@ package akka.dispatch -import org.multiverse.commitbarriers.CountDownCommitBarrier - import java.util.concurrent._ import atomic. {AtomicInteger, AtomicBoolean, AtomicReference, AtomicLong} import akka.util. {Switch, ReentrantGuard, Logging, HashCode} @@ -17,8 +15,7 @@ import akka.actor._ final class MessageInvocation(val receiver: ActorRef, val message: Any, val sender: Option[ActorRef], - val senderFuture: Option[CompletableFuture[Any]], - val transactionSet: Option[CountDownCommitBarrier]) { + val senderFuture: Option[CompletableFuture[Any]]) { if (receiver eq null) throw new IllegalArgumentException("Receiver can't be null") def invoke = try { @@ -47,7 +44,6 @@ final class MessageInvocation(val receiver: ActorRef, "\n\treceiver = " + receiver + "\n\tsender = " + sender + "\n\tsenderFuture = " + senderFuture + - "\n\ttransactionSet = " + transactionSet + "]" } } diff --git a/akka-actor/src/main/scala/util/ReflectiveAccess.scala b/akka-actor/src/main/scala/util/ReflectiveAccess.scala index 3320522ed6..5257d596f0 100644 --- a/akka-actor/src/main/scala/util/ReflectiveAccess.scala +++ b/akka-actor/src/main/scala/util/ReflectiveAccess.scala @@ -7,7 +7,6 @@ package akka.util import akka.actor.{ActorRef, IllegalActorStateException, ActorType, Uuid} import akka.dispatch.{Future, CompletableFuture, MessageInvocation} import akka.config.{Config, ModuleNotAvailableException} -import akka.stm.Transaction import akka.AkkaException import java.net.InetSocketAddress @@ -23,12 +22,10 @@ object ReflectiveAccess extends Logging { lazy val isRemotingEnabled = RemoteClientModule.isRemotingEnabled lazy val isTypedActorEnabled = TypedActorModule.isTypedActorEnabled - lazy val isJtaEnabled = JtaModule.isJtaEnabled lazy val isEnterpriseEnabled = EnterpriseModule.isEnterpriseEnabled def ensureRemotingEnabled = RemoteClientModule.ensureRemotingEnabled def ensureTypedActorEnabled = TypedActorModule.ensureTypedActorEnabled - def ensureJtaEnabled = JtaModule.ensureJtaEnabled def ensureEnterpriseEnabled = EnterpriseModule.ensureEnterpriseEnabled /** @@ -176,32 +173,6 @@ object ReflectiveAccess extends Logging { } } - object JtaModule { - - type TransactionContainerObject = { - def apply(): TransactionContainer - } - - type TransactionContainer = { - def beginWithStmSynchronization(transaction: Transaction): Unit - def commit: Unit - def rollback: Unit - } - - lazy val isJtaEnabled = transactionContainerObjectInstance.isDefined - - def ensureJtaEnabled = if (!isJtaEnabled) throw new ModuleNotAvailableException( - "Can't load the typed actor module, make sure that akka-jta.jar is on the classpath") - - val transactionContainerObjectInstance: Option[TransactionContainerObject] = - getObjectFor("akka.jta.TransactionContainer$") - - def createTransactionContainer: TransactionContainer = { - ensureJtaEnabled - transactionContainerObjectInstance.get.apply.asInstanceOf[TransactionContainer] - } - } - object EnterpriseModule { type Mailbox = { diff --git a/akka-camel/src/main/scala/Consumer.scala b/akka-camel/src/main/scala/Consumer.scala index 8720f34a39..a6323c3bae 100644 --- a/akka-camel/src/main/scala/Consumer.scala +++ b/akka-camel/src/main/scala/Consumer.scala @@ -55,7 +55,6 @@ trait Consumer { self: Actor => * * @see UntypedConsumerActor * @see RemoteUntypedConsumerActor - * @see UntypedConsumerTransactor * * @author Martin Krasser */ @@ -82,12 +81,6 @@ trait UntypedConsumer extends Consumer { self: UntypedActor => */ abstract class UntypedConsumerActor extends UntypedActor with UntypedConsumer -/** - * Subclass this abstract class to create an MDB-style transacted untyped consumer - * actor. This class is meant to be used from Java. - */ -abstract class UntypedConsumerTransactor extends UntypedTransactor with UntypedConsumer - /** * Subclass this abstract class to create an MDB-style remote untyped consumer * actor. This class is meant to be used from Java. diff --git a/akka-camel/src/main/scala/component/ActorComponent.scala b/akka-camel/src/main/scala/component/ActorComponent.scala index c1f5cd56a1..e84a894ee3 100644 --- a/akka-camel/src/main/scala/component/ActorComponent.scala +++ b/akka-camel/src/main/scala/component/ActorComponent.scala @@ -16,7 +16,6 @@ import akka.actor._ import akka.camel.{Failure, Message} import akka.camel.CamelMessageConversion.toExchangeAdapter import akka.dispatch.{CompletableFuture, MessageInvocation, MessageDispatcher} -import akka.stm.TransactionConfig import scala.reflect.BeanProperty @@ -274,9 +273,6 @@ private[akka] class AsyncCallbackAdapter(exchange: Exchange, callback: AsyncCall def actorClassName = unsupported def dispatcher_=(md: MessageDispatcher): Unit = unsupported def dispatcher: MessageDispatcher = unsupported - def transactionConfig_=(config: TransactionConfig): Unit = unsupported - def transactionConfig: TransactionConfig = unsupported - def makeTransactionRequired: Unit = unsupported def makeRemote(hostname: String, port: Int): Unit = unsupported def makeRemote(address: InetSocketAddress): Unit = unsupported def homeAddress_=(address: InetSocketAddress): Unit = unsupported diff --git a/akka-jta/src/test/scala/ReflectiveAccessSpec.scala b/akka-jta/src/test/scala/ReflectiveAccessSpec.scala index 1b13ef67b5..417aec6eab 100644 --- a/akka-jta/src/test/scala/ReflectiveAccessSpec.scala +++ b/akka-jta/src/test/scala/ReflectiveAccessSpec.scala @@ -6,11 +6,11 @@ package akka.jta import org.scalatest.junit.JUnitSuite import org.junit.Test -import akka.util.ReflectiveAccess +import akka.stm.JtaModule class ReflectiveAccessSpec extends JUnitSuite { @Test def ensureReflectiveAccessCanLoadTransactionContainer { - ReflectiveAccess.JtaModule.ensureJtaEnabled - assert(ReflectiveAccess.JtaModule.transactionContainerObjectInstance.isDefined) + JtaModule.ensureJtaEnabled + assert(JtaModule.transactionContainerObjectInstance.isDefined) } } diff --git a/akka-persistence/akka-persistence-cassandra/src/test/scala/CassandraPersistentActorSpec.scala b/akka-persistence/akka-persistence-cassandra/src/test/scala/CassandraPersistentActorSpec.scala index c877a06a1e..a425f0d7f5 100644 --- a/akka-persistence/akka-persistence-cassandra/src/test/scala/CassandraPersistentActorSpec.scala +++ b/akka-persistence/akka-persistence-cassandra/src/test/scala/CassandraPersistentActorSpec.scala @@ -1,7 +1,8 @@ package akka.persistence.cassandra -import akka.actor.{Actor, ActorRef, Transactor} +import akka.actor.{Actor, ActorRef} import Actor._ +import akka.stm.local._ import org.junit.Test import org.junit.Assert._ @@ -17,22 +18,24 @@ case class SetMapState(key: String, value: String) case class SetVectorState(key: String) case class SetRefState(key: String) case class Success(key: String, value: String) -case class Failure(key: String, value: String, failer: ActorRef) +case class Failure(key: String, value: String) case class SetMapStateOneWay(key: String, value: String) case class SetVectorStateOneWay(key: String) case class SetRefStateOneWay(key: String) case class SuccessOneWay(key: String, value: String) -case class FailureOneWay(key: String, value: String, failer: ActorRef) +case class FailureOneWay(key: String, value: String) -class CassandraPersistentActor extends Transactor { +class CassandraPersistentActor extends Actor { self.timeout = 100000 - private lazy val mapState = CassandraStorage.newMap - private lazy val vectorState = CassandraStorage.newVector - private lazy val refState = CassandraStorage.newRef + private val mapState = CassandraStorage.newMap + private val vectorState = CassandraStorage.newVector + private val refState = CassandraStorage.newRef - def receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { case GetMapState(key) => self.reply(mapState.get(key.getBytes("UTF-8")).get) case GetVectorSize => @@ -53,26 +56,21 @@ class CassandraPersistentActor extends Transactor { vectorState.add(msg.getBytes("UTF-8")) refState.swap(msg.getBytes("UTF-8")) self.reply(msg) - case Failure(key, msg, failer) => + case Failure(key, msg) => mapState.put(key.getBytes("UTF-8"), msg.getBytes("UTF-8")) vectorState.add(msg.getBytes("UTF-8")) refState.swap(msg.getBytes("UTF-8")) - failer !! "Failure" + fail self.reply(msg) } -} -@serializable class PersistentFailerActor extends Transactor { - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } class CassandraPersistentActorSpec extends JUnitSuite { - //@Before - //def startCassandra = EmbeddedCassandraService.start + // @Before + // def startCassandra = EmbeddedCassandraService.start @Test def testMapShouldNotRollbackStateForStatefulServerInCaseOfSuccess = { @@ -89,10 +87,8 @@ class CassandraPersistentActorSpec extends JUnitSuite { val stateful = actorOf[CassandraPersistentActor] stateful.start stateful !! SetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure", "init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch {case e: RuntimeException => {}} val result = (stateful !! GetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure")).as[Array[Byte]].get @@ -113,10 +109,8 @@ class CassandraPersistentActorSpec extends JUnitSuite { val stateful = actorOf[CassandraPersistentActor] stateful.start stateful !! SetVectorState("init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch {case e: RuntimeException => {}} assertEquals(1, (stateful !! GetVectorSize).get.asInstanceOf[java.lang.Integer].intValue) @@ -137,10 +131,8 @@ class CassandraPersistentActorSpec extends JUnitSuite { val stateful = actorOf[CassandraPersistentActor] stateful.start stateful !! SetRefState("init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch {case e: RuntimeException => {}} val result = (stateful !! GetRefState).as[Array[Byte]].get @@ -148,9 +140,10 @@ class CassandraPersistentActorSpec extends JUnitSuite { } } + /* -import org.apache.cassandra.service.CassandraDaemon -object // EmbeddedCassandraService { +object EmbeddedCassandraService { + import org.apache.cassandra.thrift.CassandraDaemon System.setProperty("storage-config", "src/test/resources"); diff --git a/akka-persistence/akka-persistence-couchdb/src/test/scala/CouchDBPersistentActorSpec.scala b/akka-persistence/akka-persistence-couchdb/src/test/scala/CouchDBPersistentActorSpec.scala index 1e8da99da5..7d418375af 100644 --- a/akka-persistence/akka-persistence-couchdb/src/test/scala/CouchDBPersistentActorSpec.scala +++ b/akka-persistence/akka-persistence-couchdb/src/test/scala/CouchDBPersistentActorSpec.scala @@ -6,18 +6,19 @@ import org.scalatest.BeforeAndAfterEach import org.scalatest.junit.JUnitRunner import org.junit.runner.RunWith -import akka.actor.{Transactor, Actor, ActorRef} +import akka.actor.{Actor, ActorRef} import Actor._ +import akka.stm.local case class Balance(accountNo: String) -case class Debit(accountNo: String, amount: Int, failer: ActorRef) -case class MultiDebit(accountNo: String, amounts: List[Int], failer: ActorRef) +case class Debit(accountNo: String, amount: Int) +case class MultiDebit(accountNo: String, amounts: List[Int]) case class Credit(accountNo: String, amount: Int) case class Log(start: Int, finish: Int) case object LogSize -class BankAccountActor extends Transactor { +class BankAccountActor extends Actor { private val accountState = CouchDBStorage.newMap private val txnLog = CouchDBStorage.newVector @@ -25,7 +26,9 @@ class BankAccountActor extends Transactor { import sjson.json.DefaultProtocol._ import sjson.json.JsonSerialization._ - def receive: Receive = { + def receive = { case message => local.atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { // check balance case Balance(accountNo) => txnLog.add(("Balance:" + accountNo).getBytes) @@ -35,20 +38,20 @@ class BankAccountActor extends Transactor { .getOrElse(0)) // debit amount: can fail - case Debit(accountNo, amount, failer) => + case Debit(accountNo, amount) => txnLog.add(("Debit:" + accountNo + " " + amount).getBytes) val m = accountState.get(accountNo.getBytes) .map(frombinary[Int](_)) .getOrElse(0) accountState.put(accountNo.getBytes, tobinary(m - amount)) - if (amount > m) failer !! "Failure" + if (amount > m) fail self.reply(m - amount) // many debits: can fail // demonstrates true rollback even if multiple puts have been done - case MultiDebit(accountNo, amounts, failer) => + case MultiDebit(accountNo, amounts) => val sum = amounts.foldRight(0)(_ + _) txnLog.add(("MultiDebit:" + accountNo + " " + sum).getBytes) @@ -61,7 +64,7 @@ class BankAccountActor extends Transactor { amount => accountState.put(accountNo.getBytes, tobinary(m - amount)) cbal = cbal - amount - if (cbal < 0) failer !! "Failure" + if (cbal < 0) fail } self.reply(m - sum) @@ -83,13 +86,8 @@ class BankAccountActor extends Transactor { case Log(start, finish) => self.reply(txnLog.slice(start, finish).map(new String(_))) } -} -@serializable class PersistentFailerActor extends Transactor { - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } @RunWith(classOf[JUnitRunner]) @@ -108,25 +106,23 @@ class CouchDBPersistentActor extends describe("successful debit") { it("should debit successfully") { - log.info("Succesful Debit starting") + Actor.log.info("Succesful Debit starting") val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) - log.info("credited") - bactor !! Debit("a-123", 3000, failer) - log.info("debited") + Actor.log.info("credited") + bactor !! Debit("a-123", 3000) + Actor.log.info("debited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(2000) - log.info("balane matched") + Actor.log.info("balane matched") bactor !! Credit("a-123", 7000) - log.info("Credited") + Actor.log.info("Credited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(9000) - log.info("Balance matched") - bactor !! Debit("a-123", 8000, failer) - log.info("Debited") + Actor.log.info("Balance matched") + bactor !! Debit("a-123", 8000) + Actor.log.info("Debited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(1000) - log.info("Balance matched") + Actor.log.info("Balance matched") (bactor !! LogSize).get.asInstanceOf[Int] should equal(7) (bactor !! Log(0, 7)).get.asInstanceOf[Iterable[String]].size should equal(7) } @@ -136,12 +132,10 @@ class CouchDBPersistentActor extends it("debit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! Debit("a-123", 7000, failer) + bactor !! Debit("a-123", 7000) } should produce [Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) @@ -152,12 +146,10 @@ class CouchDBPersistentActor extends it("multidebit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! MultiDebit("a-123", List(1000, 2000, 4000), failer) + bactor !! MultiDebit("a-123", List(1000, 2000, 4000)) } should produce [Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) diff --git a/akka-persistence/akka-persistence-hbase/src/test/scala/HbasePersistentActorSpecTestIntegration.scala b/akka-persistence/akka-persistence-hbase/src/test/scala/HbasePersistentActorSpecTestIntegration.scala index 8992ce71d7..2a8ef2c8e3 100644 --- a/akka-persistence/akka-persistence-hbase/src/test/scala/HbasePersistentActorSpecTestIntegration.scala +++ b/akka-persistence/akka-persistence-hbase/src/test/scala/HbasePersistentActorSpecTestIntegration.scala @@ -1,7 +1,8 @@ package akka.persistence.hbase -import akka.actor.{ Actor, ActorRef, Transactor } +import akka.actor.{ Actor, ActorRef } import Actor._ +import akka.stm.local._ import org.junit.Test import org.junit.Assert._ @@ -23,22 +24,24 @@ case class SetMapState(key: String, value: String) case class SetVectorState(key: String) case class SetRefState(key: String) case class Success(key: String, value: String) -case class Failure(key: String, value: String, failer: ActorRef) +case class Failure(key: String, value: String) case class SetMapStateOneWay(key: String, value: String) case class SetVectorStateOneWay(key: String) case class SetRefStateOneWay(key: String) case class SuccessOneWay(key: String, value: String) -case class FailureOneWay(key: String, value: String, failer: ActorRef) +case class FailureOneWay(key: String, value: String) -class HbasePersistentActor extends Transactor { +class HbasePersistentActor extends Actor { self.timeout = 100000 - private lazy val mapState = HbaseStorage.newMap - private lazy val vectorState = HbaseStorage.newVector - private lazy val refState = HbaseStorage.newRef + private val mapState = HbaseStorage.newMap + private val vectorState = HbaseStorage.newVector + private val refState = HbaseStorage.newRef - def receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { case GetMapState(key) => self.reply(mapState.get(key.getBytes("UTF-8")).get) case GetVectorSize => @@ -59,21 +62,15 @@ class HbasePersistentActor extends Transactor { vectorState.add(msg.getBytes("UTF-8")) refState.swap(msg.getBytes("UTF-8")) self.reply(msg) - case Failure(key, msg, failer) => + case Failure(key, msg) => mapState.put(key.getBytes("UTF-8"), msg.getBytes("UTF-8")) vectorState.add(msg.getBytes("UTF-8")) refState.swap(msg.getBytes("UTF-8")) - failer !! "Failure" + fail self.reply(msg) } -} -@serializable -class PersistentFailerActor extends Transactor { - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } class HbasePersistentActorSpecTestIntegration extends JUnitSuite with BeforeAndAfterAll { @@ -113,10 +110,8 @@ class HbasePersistentActorSpecTestIntegration extends JUnitSuite with BeforeAndA val stateful = actorOf[HbasePersistentActor] stateful.start stateful !! SetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure", "init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch { case e: RuntimeException => {} } val result = (stateful !! GetMapState("testShouldRollbackStateForStatefulServerInCaseOfFailure")).as[Array[Byte]].get @@ -137,10 +132,8 @@ class HbasePersistentActorSpecTestIntegration extends JUnitSuite with BeforeAndA val stateful = actorOf[HbasePersistentActor] stateful.start stateful !! SetVectorState("init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch { case e: RuntimeException => {} } assertEquals(1, (stateful !! GetVectorSize).get.asInstanceOf[java.lang.Integer].intValue) @@ -161,10 +154,8 @@ class HbasePersistentActorSpecTestIntegration extends JUnitSuite with BeforeAndA val stateful = actorOf[HbasePersistentActor] stateful.start stateful !! SetRefState("init") // set init state - val failer = actorOf[PersistentFailerActor] - failer.start try { - stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state", failer) // call failing transactionrequired method + stateful !! Failure("testShouldRollbackStateForStatefulServerInCaseOfFailure", "new state") // call failing transactionrequired method fail("should have thrown an exception") } catch { case e: RuntimeException => {} } val result = (stateful !! GetRefState).as[Array[Byte]].get diff --git a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala index 8708d1b45e..597a847e1b 100644 --- a/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala +++ b/akka-persistence/akka-persistence-mongo/src/test/scala/MongoPersistentActorSpec.scala @@ -6,26 +6,29 @@ import org.scalatest.BeforeAndAfterEach import org.scalatest.junit.JUnitRunner import org.junit.runner.RunWith -import akka.actor.{Transactor, Actor, ActorRef} +import akka.actor.{Actor, ActorRef} import Actor._ +import akka.stm.local._ case class Balance(accountNo: String) -case class Debit(accountNo: String, amount: Int, failer: ActorRef) -case class MultiDebit(accountNo: String, amounts: List[Int], failer: ActorRef) +case class Debit(accountNo: String, amount: Int) +case class MultiDebit(accountNo: String, amounts: List[Int]) case class Credit(accountNo: String, amount: Int) case class Log(start: Int, finish: Int) case object LogSize -class BankAccountActor extends Transactor { +class BankAccountActor extends Actor { - private lazy val accountState = MongoStorage.newMap - private lazy val txnLog = MongoStorage.newVector + private val accountState = MongoStorage.newMap + private val txnLog = MongoStorage.newVector import sjson.json.DefaultProtocol._ import sjson.json.JsonSerialization._ - def receive: Receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { // check balance case Balance(accountNo) => txnLog.add(("Balance:" + accountNo).getBytes) @@ -35,20 +38,20 @@ class BankAccountActor extends Transactor { .getOrElse(0)) // debit amount: can fail - case Debit(accountNo, amount, failer) => + case Debit(accountNo, amount) => txnLog.add(("Debit:" + accountNo + " " + amount).getBytes) val m = accountState.get(accountNo.getBytes) .map(frombinary[Int](_)) .getOrElse(0) accountState.put(accountNo.getBytes, tobinary(m - amount)) - if (amount > m) failer !! "Failure" + if (amount > m) fail self.reply(m - amount) // many debits: can fail // demonstrates true rollback even if multiple puts have been done - case MultiDebit(accountNo, amounts, failer) => + case MultiDebit(accountNo, amounts) => val sum = amounts.foldRight(0)(_ + _) txnLog.add(("MultiDebit:" + accountNo + " " + sum).getBytes) @@ -60,7 +63,7 @@ class BankAccountActor extends Transactor { amounts.foreach { amount => accountState.put(accountNo.getBytes, tobinary(m - amount)) cbal = cbal - amount - if (cbal < 0) failer !! "Failure" + if (cbal < 0) fail } self.reply(m - sum) @@ -82,13 +85,8 @@ class BankAccountActor extends Transactor { case Log(start, finish) => self.reply(txnLog.slice(start, finish).map(new String(_))) } -} -@serializable class PersistentFailerActor extends Transactor { - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } @RunWith(classOf[JUnitRunner]) @@ -109,17 +107,15 @@ class MongoPersistentActorSpec extends it("should debit successfully") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) - bactor !! Debit("a-123", 3000, failer) + bactor !! Debit("a-123", 3000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(2000) bactor !! Credit("a-123", 7000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(9000) - bactor !! Debit("a-123", 8000, failer) + bactor !! Debit("a-123", 8000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(1000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(7) @@ -131,12 +127,10 @@ class MongoPersistentActorSpec extends it("debit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! Debit("a-123", 7000, failer) + bactor !! Debit("a-123", 7000) } should produce [Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) @@ -147,12 +141,10 @@ class MongoPersistentActorSpec extends it("multidebit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! MultiDebit("a-123", List(1000, 2000, 4000), failer) + bactor !! MultiDebit("a-123", List(1000, 2000, 4000)) } should produce [Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) diff --git a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala index c487741ea6..a28a5485ed 100644 --- a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala +++ b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentActorSpec.scala @@ -5,8 +5,9 @@ import org.scalatest.junit.JUnitSuite import org.junit.{Test, Before} import org.junit.Assert._ -import akka.actor.{ActorRef, Transactor} +import akka.actor.{Actor, ActorRef} import akka.actor.Actor._ +import akka.stm.local._ /** * A persistent actor based on Redis storage. @@ -22,44 +23,47 @@ import akka.actor.Actor._ */ case class Balance(accountNo: String) -case class Debit(accountNo: String, amount: Int, failer: ActorRef) -case class MultiDebit(accountNo: String, amounts: List[Int], failer: ActorRef) +case class Debit(accountNo: String, amount: Int) +case class MultiDebit(accountNo: String, amounts: List[Int]) case class Credit(accountNo: String, amount: Int) case object LogSize -class AccountActor extends Transactor { - private lazy val accountState = RedisStorage.newMap +class AccountActor extends Actor { + private val accountState = RedisStorage.newMap private val txnLog = RedisStorage.newVector self.timeout = 100000 - def receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { // check balance case Balance(accountNo) => txnLog.add("Balance:%s".format(accountNo).getBytes) self.reply(new String(accountState.get(accountNo.getBytes).get).toInt) // debit amount: can fail - case Debit(accountNo, amount, failer) => + case Debit(accountNo, amount) => txnLog.add("Debit:%s %s".format(accountNo, amount.toString).getBytes) - val Some(m) = accountState.get(accountNo.getBytes).map(x => (new String(x)).toInt) orElse Some(0) + val Some(m) = accountState.get(accountNo.getBytes).map(x => (new String(x)).toInt) orElse Some(0) accountState.put(accountNo.getBytes, (m - amount).toString.getBytes) - if (amount > m) - failer !! "Failure" + if (amount > m) fail + self.reply(m - amount) // many debits: can fail // demonstrates true rollback even if multiple puts have been done - case MultiDebit(accountNo, amounts, failer) => + case MultiDebit(accountNo, amounts) => txnLog.add("MultiDebit:%s %s".format(accountNo, amounts.map(_.intValue).foldLeft(0)(_ + _).toString).getBytes) - val Some(m) = accountState.get(accountNo.getBytes).map(x => (new String(x)).toInt) orElse Some(0) + val Some(m) = accountState.get(accountNo.getBytes).map(x => (new String(x)).toInt) orElse Some(0) var bal = 0 amounts.foreach {amount => bal = bal + amount accountState.put(accountNo.getBytes, (m - bal).toString.getBytes) } - if (bal > m) failer !! "Failure" + if (bal > m) fail + self.reply(m - bal) // credit amount @@ -73,14 +77,8 @@ class AccountActor extends Transactor { case LogSize => self.reply(txnLog.length.asInstanceOf[AnyRef]) } -} -@serializable class PersistentFailerActor extends Transactor { - self.timeout = 5000 - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } class RedisPersistentActorSpec extends JUnitSuite { @@ -88,16 +86,14 @@ class RedisPersistentActorSpec extends JUnitSuite { def testSuccessfulDebit = { val bactor = actorOf(new AccountActor) bactor.start - val failer = actorOf(new PersistentFailerActor) - failer.start bactor !! Credit("a-123", 5000) - bactor !! Debit("a-123", 3000, failer) + bactor !! Debit("a-123", 3000) assertEquals(2000, (bactor !! Balance("a-123")).get) bactor !! Credit("a-123", 7000) assertEquals(9000, (bactor !! Balance("a-123")).get) - bactor !! Debit("a-123", 8000, failer) + bactor !! Debit("a-123", 8000) assertEquals(1000, (bactor !! Balance("a-123")).get) val c = (bactor !! LogSize).as[Int].get @@ -111,10 +107,8 @@ class RedisPersistentActorSpec extends JUnitSuite { bactor !! Credit("a-123", 5000) assertEquals(5000, (bactor !! Balance("a-123")).get) - val failer = actorOf[PersistentFailerActor] - failer.start try { - bactor !! Debit("a-123", 7000, failer) + bactor !! Debit("a-123", 7000) fail("should throw exception") } catch { case e: RuntimeException => {}} @@ -133,10 +127,8 @@ class RedisPersistentActorSpec extends JUnitSuite { assertEquals(5000, (bactor !! (Balance("a-123"), 5000)).get) - val failer = actorOf[PersistentFailerActor] - failer.start try { - bactor !! MultiDebit("a-123", List(500, 2000, 1000, 3000), failer) + bactor !! MultiDebit("a-123", List(500, 2000, 1000, 3000)) fail("should throw exception") } catch { case e: RuntimeException => {}} diff --git a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala index f21b29064f..e8f127f218 100644 --- a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala +++ b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentQSpec.scala @@ -3,8 +3,9 @@ package akka.persistence.redis import org.junit.{Test, Before} import org.junit.Assert._ -import akka.actor.{Actor, ActorRef, Transactor} +import akka.actor.{Actor, ActorRef} import Actor._ +import akka.stm.local._ /** * A persistent actor based on Redis queue storage. @@ -15,13 +16,15 @@ import Actor._ case class NQ(accountNo: String) case object DQ -case class MNDQ(accountNos: List[String], noOfDQs: Int, failer: ActorRef) +case class MNDQ(accountNos: List[String], noOfDQs: Int) case object SZ -class QueueActor extends Transactor { - private lazy val accounts = RedisStorage.newQueue +class QueueActor extends Actor { + private val accounts = RedisStorage.newQueue - def receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { // enqueue case NQ(accountNo) => accounts.enqueue(accountNo.getBytes) @@ -33,13 +36,12 @@ class QueueActor extends Transactor { self.reply(d) // multiple NQ and DQ - case MNDQ(enqs, no, failer) => + case MNDQ(enqs, no) => accounts.enqueue(enqs.map(_.getBytes): _*) try { (1 to no).foreach(e => accounts.dequeue) } catch { - case e: Exception => - failer !! "Failure" + case e: Exception => fail } self.reply(true) @@ -47,6 +49,8 @@ class QueueActor extends Transactor { case SZ => self.reply(accounts.size) } + + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } import org.scalatest.junit.JUnitSuite @@ -82,8 +86,6 @@ class RedisPersistentQSpec extends JUnitSuite { def testSuccessfulMNDQ = { val qa = actorOf[QueueActor] qa.start - val failer = actorOf[PersistentFailerActor] - failer.start qa !! NQ("a-123") qa !! NQ("a-124") @@ -93,7 +95,7 @@ class RedisPersistentQSpec extends JUnitSuite { assertEquals("a-123", (qa !! DQ).get) val s = (qa !! SZ).as[Int].get assertTrue(2 == s) - qa !! MNDQ(List("a-126", "a-127"), 2, failer) + qa !! MNDQ(List("a-126", "a-127"), 2) val u = (qa !! SZ).as[Int].get assertTrue(2 == u) } @@ -102,8 +104,6 @@ class RedisPersistentQSpec extends JUnitSuite { def testMixedMNDQ = { val qa = actorOf[QueueActor] qa.start - val failer = actorOf[PersistentFailerActor] - failer.start // 3 enqueues qa !! NQ("a-123") @@ -121,7 +121,7 @@ class RedisPersistentQSpec extends JUnitSuite { assertTrue(2 == s) // enqueue 2, dequeue 2 => size == 2 - qa !! MNDQ(List("a-126", "a-127"), 2, failer) + qa !! MNDQ(List("a-126", "a-127"), 2) val u = (qa !! SZ).as[Int].get assertTrue(2 == u) @@ -135,7 +135,7 @@ class RedisPersistentQSpec extends JUnitSuite { // dequeue 6 => fail transaction // size should remain 4 try { - qa !! MNDQ(List("a-130"), 6, failer) + qa !! MNDQ(List("a-130"), 6) } catch { case e: Exception => {} } val w = (qa !! SZ).as[Int].get diff --git a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentSortedSetSpec.scala b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentSortedSetSpec.scala index 6ad57da8bd..d8a061a64d 100644 --- a/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentSortedSetSpec.scala +++ b/akka-persistence/akka-persistence-redis/src/test/scala/RedisPersistentSortedSetSpec.scala @@ -7,8 +7,9 @@ import org.scalatest.BeforeAndAfterAll import org.scalatest.junit.JUnitRunner import org.junit.runner.RunWith -import akka.actor.{Actor, ActorRef, Transactor} +import akka.actor.{Actor, ActorRef} import Actor._ +import akka.stm.local._ /** * A persistent actor based on Redis sortedset storage. @@ -43,15 +44,17 @@ case class SCORE(h: Hacker) case class RANGE(start: Int, end: Int) // add and remove subject to the condition that there will be at least 3 hackers -case class MULTI(add: List[Hacker], rem: List[Hacker], failer: ActorRef) +case class MULTI(add: List[Hacker], rem: List[Hacker]) case class MULTIRANGE(add: List[Hacker]) -class SortedSetActor extends Transactor { +class SortedSetActor extends Actor { self.timeout = 100000 - private lazy val hackers = RedisStorage.newSortedSet + private val hackers = RedisStorage.newSortedSet - def receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { case ADD(h) => hackers.+(h.name.getBytes, h.zscore) self.reply(true) @@ -69,7 +72,7 @@ class SortedSetActor extends Transactor { case RANGE(s, e) => self.reply(hackers.zrange(s, e)) - case MULTI(a, r, failer) => + case MULTI(a, r) => a.foreach{ h: Hacker => hackers.+(h.name.getBytes, h.zscore) } @@ -80,8 +83,7 @@ class SortedSetActor extends Transactor { hackers.-(h.name.getBytes) } } catch { - case e: Exception => - failer !! "Failure" + case e: Exception => fail } self.reply((a.size, r.size)) @@ -91,6 +93,8 @@ class SortedSetActor extends Transactor { } self.reply(hackers.zrange(0, -1)) } + + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } import RedisStorageBackend._ @@ -178,13 +182,10 @@ class RedisPersistentSortedSetSpec extends val qa = actorOf[SortedSetActor] qa.start - val failer = actorOf[PersistentFailerActor] - failer.start - (qa !! SIZE).get.asInstanceOf[Int] should equal(0) val add = List(h1, h2, h3, h4) val rem = List(h2) - (qa !! MULTI(add, rem, failer)).get.asInstanceOf[Tuple2[Int, Int]] should equal((4,1)) + (qa !! MULTI(add, rem)).get.asInstanceOf[Tuple2[Int, Int]] should equal((4,1)) (qa !! SIZE).get.asInstanceOf[Int] should equal(3) // size == 3 @@ -194,7 +195,7 @@ class RedisPersistentSortedSetSpec extends // remove 3 val rem1 = List(h1, h3, h4, h5) try { - qa !! MULTI(add1, rem1, failer) + qa !! MULTI(add1, rem1) } catch { case e: RuntimeException => {} } (qa !! SIZE).get.asInstanceOf[Int] should equal(3) } diff --git a/akka-persistence/akka-persistence-voldemort/src/test/scala/VoldemortPersistentActorSuite.scala b/akka-persistence/akka-persistence-voldemort/src/test/scala/VoldemortPersistentActorSuite.scala index 7a14503335..b0b0bdb2b3 100644 --- a/akka-persistence/akka-persistence-voldemort/src/test/scala/VoldemortPersistentActorSuite.scala +++ b/akka-persistence/akka-persistence-voldemort/src/test/scala/VoldemortPersistentActorSuite.scala @@ -6,14 +6,15 @@ import org.scalatest.BeforeAndAfterEach import org.scalatest.junit.JUnitRunner import org.junit.runner.RunWith -import akka.actor.{Transactor, Actor, ActorRef} +import akka.actor.{Actor, ActorRef} import Actor._ import BankAccountActor._ +import akka.stm.local._ case class Balance(accountNo: String) -case class Debit(accountNo: String, amount: Int, failer: ActorRef) -case class MultiDebit(accountNo: String, amounts: List[Int], failer: ActorRef) +case class Debit(accountNo: String, amount: Int) +case class MultiDebit(accountNo: String, amounts: List[Int]) case class Credit(accountNo: String, amount: Int) case class Log(start: Int, finish: Int) case object LogSize @@ -23,14 +24,16 @@ object BankAccountActor { val tx = "txnLog" } -class BankAccountActor extends Transactor { +class BankAccountActor extends Actor { private val accountState = VoldemortStorage.newMap(state) private val txnLog = VoldemortStorage.newVector(tx) import sjson.json.DefaultProtocol._ import sjson.json.JsonSerialization._ - def receive: Receive = { + def receive = { case message => atomic { atomicReceive(message) } } + + def atomicReceive: Receive = { // check balance case Balance(accountNo) => txnLog.add(("Balance:" + accountNo).getBytes) @@ -40,20 +43,20 @@ class BankAccountActor extends Transactor { .getOrElse(0)) // debit amount: can fail - case Debit(accountNo, amount, failer) => + case Debit(accountNo, amount) => txnLog.add(("Debit:" + accountNo + " " + amount).getBytes) val m = accountState.get(accountNo.getBytes) .map(frombinary[Int](_)) .getOrElse(0) accountState.put(accountNo.getBytes, tobinary(m - amount)) - if (amount > m) failer !! "Failure" + if (amount > m) fail self.reply(m - amount) // many debits: can fail // demonstrates true rollback even if multiple puts have been done - case MultiDebit(accountNo, amounts, failer) => + case MultiDebit(accountNo, amounts) => val sum = amounts.foldRight(0)(_ + _) txnLog.add(("MultiDebit:" + accountNo + " " + sum).getBytes) @@ -66,7 +69,7 @@ class BankAccountActor extends Transactor { amount => accountState.put(accountNo.getBytes, tobinary(m - amount)) cbal = cbal - amount - if (cbal < 0) failer !! "Failure" + if (cbal < 0) fail } self.reply(m - sum) @@ -88,13 +91,8 @@ class BankAccountActor extends Transactor { case Log(start, finish) => self.reply(txnLog.slice(start, finish).map(new String(_))) } -} -@serializable class PersistentFailerActor extends Transactor { - def receive = { - case "Failure" => - throw new RuntimeException("Expected exception; to test fault-tolerance") - } + def fail = throw new RuntimeException("Expected exception; to test fault-tolerance") } @RunWith(classOf[JUnitRunner]) @@ -119,11 +117,9 @@ Spec with log.info("Succesful Debit starting") val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) log.info("credited") - bactor !! Debit("a-123", 3000, failer) + bactor !! Debit("a-123", 3000) log.info("debited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(2000) log.info("balane matched") @@ -131,7 +127,7 @@ Spec with log.info("Credited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(9000) log.info("Balance matched") - bactor !! Debit("a-123", 8000, failer) + bactor !! Debit("a-123", 8000) log.info("Debited") (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(1000) log.info("Balance matched") @@ -144,12 +140,10 @@ Spec with it("debit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! Debit("a-123", 7000, failer) + bactor !! Debit("a-123", 7000) } should produce[Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) @@ -160,12 +154,10 @@ Spec with it("multidebit should fail") { val bactor = actorOf[BankAccountActor] bactor.start - val failer = actorOf[PersistentFailerActor] - failer.start bactor !! Credit("a-123", 5000) (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) evaluating { - bactor !! MultiDebit("a-123", List(1000, 2000, 4000), failer) + bactor !! MultiDebit("a-123", List(1000, 2000, 4000)) } should produce[Exception] (bactor !! Balance("a-123")).get.asInstanceOf[Int] should equal(5000) (bactor !! LogSize).get.asInstanceOf[Int] should equal(3) diff --git a/akka-remote/src/main/java/akka/remote/protocol/RemoteProtocol.java b/akka-remote/src/main/java/akka/remote/protocol/RemoteProtocol.java index e4da0ea515..90e141546c 100644 --- a/akka-remote/src/main/java/akka/remote/protocol/RemoteProtocol.java +++ b/akka-remote/src/main/java/akka/remote/protocol/RemoteProtocol.java @@ -1076,50 +1076,43 @@ public final class RemoteProtocol { public boolean hasSerializerClassname() { return hasSerializerClassname; } public java.lang.String getSerializerClassname() { return serializerClassname_; } - // optional bool isTransactor = 7; - public static final int ISTRANSACTOR_FIELD_NUMBER = 7; - private boolean hasIsTransactor; - private boolean isTransactor_ = false; - public boolean hasIsTransactor() { return hasIsTransactor; } - public boolean getIsTransactor() { return isTransactor_; } - - // optional uint64 timeout = 8; - public static final int TIMEOUT_FIELD_NUMBER = 8; + // optional uint64 timeout = 7; + public static final int TIMEOUT_FIELD_NUMBER = 7; private boolean hasTimeout; private long timeout_ = 0L; public boolean hasTimeout() { return hasTimeout; } public long getTimeout() { return timeout_; } - // optional uint64 receiveTimeout = 9; - public static final int RECEIVETIMEOUT_FIELD_NUMBER = 9; + // optional uint64 receiveTimeout = 8; + public static final int RECEIVETIMEOUT_FIELD_NUMBER = 8; private boolean hasReceiveTimeout; private long receiveTimeout_ = 0L; public boolean hasReceiveTimeout() { return hasReceiveTimeout; } public long getReceiveTimeout() { return receiveTimeout_; } - // optional .LifeCycleProtocol lifeCycle = 10; - public static final int LIFECYCLE_FIELD_NUMBER = 10; + // optional .LifeCycleProtocol lifeCycle = 9; + public static final int LIFECYCLE_FIELD_NUMBER = 9; private boolean hasLifeCycle; private akka.remote.protocol.RemoteProtocol.LifeCycleProtocol lifeCycle_; public boolean hasLifeCycle() { return hasLifeCycle; } public akka.remote.protocol.RemoteProtocol.LifeCycleProtocol getLifeCycle() { return lifeCycle_; } - // optional .RemoteActorRefProtocol supervisor = 11; - public static final int SUPERVISOR_FIELD_NUMBER = 11; + // optional .RemoteActorRefProtocol supervisor = 10; + public static final int SUPERVISOR_FIELD_NUMBER = 10; private boolean hasSupervisor; private akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol supervisor_; public boolean hasSupervisor() { return hasSupervisor; } public akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol getSupervisor() { return supervisor_; } - // optional bytes hotswapStack = 12; - public static final int HOTSWAPSTACK_FIELD_NUMBER = 12; + // optional bytes hotswapStack = 11; + public static final int HOTSWAPSTACK_FIELD_NUMBER = 11; private boolean hasHotswapStack; private com.google.protobuf.ByteString hotswapStack_ = com.google.protobuf.ByteString.EMPTY; public boolean hasHotswapStack() { return hasHotswapStack; } public com.google.protobuf.ByteString getHotswapStack() { return hotswapStack_; } - // repeated .RemoteMessageProtocol messages = 13; - public static final int MESSAGES_FIELD_NUMBER = 13; + // repeated .RemoteMessageProtocol messages = 12; + public static final int MESSAGES_FIELD_NUMBER = 12; private java.util.List messages_ = java.util.Collections.emptyList(); public java.util.List getMessagesList() { @@ -1176,26 +1169,23 @@ public final class RemoteProtocol { if (hasSerializerClassname()) { output.writeString(6, getSerializerClassname()); } - if (hasIsTransactor()) { - output.writeBool(7, getIsTransactor()); - } if (hasTimeout()) { - output.writeUInt64(8, getTimeout()); + output.writeUInt64(7, getTimeout()); } if (hasReceiveTimeout()) { - output.writeUInt64(9, getReceiveTimeout()); + output.writeUInt64(8, getReceiveTimeout()); } if (hasLifeCycle()) { - output.writeMessage(10, getLifeCycle()); + output.writeMessage(9, getLifeCycle()); } if (hasSupervisor()) { - output.writeMessage(11, getSupervisor()); + output.writeMessage(10, getSupervisor()); } if (hasHotswapStack()) { - output.writeBytes(12, getHotswapStack()); + output.writeBytes(11, getHotswapStack()); } for (akka.remote.protocol.RemoteProtocol.RemoteMessageProtocol element : getMessagesList()) { - output.writeMessage(13, element); + output.writeMessage(12, element); } getUnknownFields().writeTo(output); } @@ -1230,33 +1220,29 @@ public final class RemoteProtocol { size += com.google.protobuf.CodedOutputStream .computeStringSize(6, getSerializerClassname()); } - if (hasIsTransactor()) { - size += com.google.protobuf.CodedOutputStream - .computeBoolSize(7, getIsTransactor()); - } if (hasTimeout()) { size += com.google.protobuf.CodedOutputStream - .computeUInt64Size(8, getTimeout()); + .computeUInt64Size(7, getTimeout()); } if (hasReceiveTimeout()) { size += com.google.protobuf.CodedOutputStream - .computeUInt64Size(9, getReceiveTimeout()); + .computeUInt64Size(8, getReceiveTimeout()); } if (hasLifeCycle()) { size += com.google.protobuf.CodedOutputStream - .computeMessageSize(10, getLifeCycle()); + .computeMessageSize(9, getLifeCycle()); } if (hasSupervisor()) { size += com.google.protobuf.CodedOutputStream - .computeMessageSize(11, getSupervisor()); + .computeMessageSize(10, getSupervisor()); } if (hasHotswapStack()) { size += com.google.protobuf.CodedOutputStream - .computeBytesSize(12, getHotswapStack()); + .computeBytesSize(11, getHotswapStack()); } for (akka.remote.protocol.RemoteProtocol.RemoteMessageProtocol element : getMessagesList()) { size += com.google.protobuf.CodedOutputStream - .computeMessageSize(13, element); + .computeMessageSize(12, element); } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; @@ -1438,9 +1424,6 @@ public final class RemoteProtocol { if (other.hasSerializerClassname()) { setSerializerClassname(other.getSerializerClassname()); } - if (other.hasIsTransactor()) { - setIsTransactor(other.getIsTransactor()); - } if (other.hasTimeout()) { setTimeout(other.getTimeout()); } @@ -1522,18 +1505,14 @@ public final class RemoteProtocol { break; } case 56: { - setIsTransactor(input.readBool()); - break; - } - case 64: { setTimeout(input.readUInt64()); break; } - case 72: { + case 64: { setReceiveTimeout(input.readUInt64()); break; } - case 82: { + case 74: { akka.remote.protocol.RemoteProtocol.LifeCycleProtocol.Builder subBuilder = akka.remote.protocol.RemoteProtocol.LifeCycleProtocol.newBuilder(); if (hasLifeCycle()) { subBuilder.mergeFrom(getLifeCycle()); @@ -1542,7 +1521,7 @@ public final class RemoteProtocol { setLifeCycle(subBuilder.buildPartial()); break; } - case 90: { + case 82: { akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol.Builder subBuilder = akka.remote.protocol.RemoteProtocol.RemoteActorRefProtocol.newBuilder(); if (hasSupervisor()) { subBuilder.mergeFrom(getSupervisor()); @@ -1551,11 +1530,11 @@ public final class RemoteProtocol { setSupervisor(subBuilder.buildPartial()); break; } - case 98: { + case 90: { setHotswapStack(input.readBytes()); break; } - case 106: { + case 98: { akka.remote.protocol.RemoteProtocol.RemoteMessageProtocol.Builder subBuilder = akka.remote.protocol.RemoteProtocol.RemoteMessageProtocol.newBuilder(); input.readMessage(subBuilder, extensionRegistry); addMessages(subBuilder.buildPartial()); @@ -1724,25 +1703,7 @@ public final class RemoteProtocol { return this; } - // optional bool isTransactor = 7; - public boolean hasIsTransactor() { - return result.hasIsTransactor(); - } - public boolean getIsTransactor() { - return result.getIsTransactor(); - } - public Builder setIsTransactor(boolean value) { - result.hasIsTransactor = true; - result.isTransactor_ = value; - return this; - } - public Builder clearIsTransactor() { - result.hasIsTransactor = false; - result.isTransactor_ = false; - return this; - } - - // optional uint64 timeout = 8; + // optional uint64 timeout = 7; public boolean hasTimeout() { return result.hasTimeout(); } @@ -1760,7 +1721,7 @@ public final class RemoteProtocol { return this; } - // optional uint64 receiveTimeout = 9; + // optional uint64 receiveTimeout = 8; public boolean hasReceiveTimeout() { return result.hasReceiveTimeout(); } @@ -1778,7 +1739,7 @@ public final class RemoteProtocol { return this; } - // optional .LifeCycleProtocol lifeCycle = 10; + // optional .LifeCycleProtocol lifeCycle = 9; public boolean hasLifeCycle() { return result.hasLifeCycle(); } @@ -1815,7 +1776,7 @@ public final class RemoteProtocol { return this; } - // optional .RemoteActorRefProtocol supervisor = 11; + // optional .RemoteActorRefProtocol supervisor = 10; public boolean hasSupervisor() { return result.hasSupervisor(); } @@ -1852,7 +1813,7 @@ public final class RemoteProtocol { return this; } - // optional bytes hotswapStack = 12; + // optional bytes hotswapStack = 11; public boolean hasHotswapStack() { return result.hasHotswapStack(); } @@ -1873,7 +1834,7 @@ public final class RemoteProtocol { return this; } - // repeated .RemoteMessageProtocol messages = 13; + // repeated .RemoteMessageProtocol messages = 12; public java.util.List getMessagesList() { return java.util.Collections.unmodifiableList(result.messages_); } @@ -6060,49 +6021,48 @@ public final class RemoteProtocol { "(\0132\020.AddressProtocol\022\017\n\007timeout\030\004 \001(\004\"_\n" + "\033RemoteTypedActorRefProtocol\022)\n\010actorRef" + "\030\001 \002(\0132\027.RemoteActorRefProtocol\022\025\n\rinter" + - "faceName\030\002 \002(\t\"\217\003\n\032SerializedActorRefPro" + + "faceName\030\002 \002(\t\"\371\002\n\032SerializedActorRefPro" + "tocol\022\033\n\004uuid\030\001 \002(\0132\r.UuidProtocol\022\n\n\002id" + "\030\002 \002(\t\022\026\n\016actorClassname\030\003 \002(\t\022)\n\017origin" + "alAddress\030\004 \002(\0132\020.AddressProtocol\022\025\n\ract", "orInstance\030\005 \001(\014\022\033\n\023serializerClassname\030" + - "\006 \001(\t\022\024\n\014isTransactor\030\007 \001(\010\022\017\n\007timeout\030\010" + - " \001(\004\022\026\n\016receiveTimeout\030\t \001(\004\022%\n\tlifeCycl" + - "e\030\n \001(\0132\022.LifeCycleProtocol\022+\n\nsuperviso" + - "r\030\013 \001(\0132\027.RemoteActorRefProtocol\022\024\n\014hots" + - "wapStack\030\014 \001(\014\022(\n\010messages\030\r \003(\0132\026.Remot" + - "eMessageProtocol\"g\n\037SerializedTypedActor" + - "RefProtocol\022-\n\010actorRef\030\001 \002(\0132\033.Serializ" + - "edActorRefProtocol\022\025\n\rinterfaceName\030\002 \002(" + - "\t\"r\n\017MessageProtocol\0225\n\023serializationSch", - "eme\030\001 \002(\0162\030.SerializationSchemeType\022\017\n\007m" + - "essage\030\002 \002(\014\022\027\n\017messageManifest\030\003 \001(\014\"\255\001" + - "\n\021ActorInfoProtocol\022\033\n\004uuid\030\001 \002(\0132\r.Uuid" + - "Protocol\022\016\n\006target\030\002 \002(\t\022\017\n\007timeout\030\003 \002(" + - "\004\022\035\n\tactorType\030\004 \002(\0162\n.ActorType\022/\n\016type" + - "dActorInfo\030\005 \001(\0132\027.TypedActorInfoProtoco" + - "l\022\n\n\002id\030\006 \001(\t\";\n\026TypedActorInfoProtocol\022" + - "\021\n\tinterface\030\001 \002(\t\022\016\n\006method\030\002 \002(\t\"\277\002\n\025R" + - "emoteMessageProtocol\022\033\n\004uuid\030\001 \002(\0132\r.Uui" + - "dProtocol\022%\n\tactorInfo\030\002 \002(\0132\022.ActorInfo", - "Protocol\022\016\n\006oneWay\030\003 \002(\010\022!\n\007message\030\004 \001(" + - "\0132\020.MessageProtocol\022%\n\texception\030\005 \001(\0132\022" + - ".ExceptionProtocol\022%\n\016supervisorUuid\030\006 \001" + - "(\0132\r.UuidProtocol\022\'\n\006sender\030\007 \001(\0132\027.Remo" + - "teActorRefProtocol\022(\n\010metadata\030\010 \003(\0132\026.M" + - "etadataEntryProtocol\022\016\n\006cookie\030\t \001(\t\")\n\014" + - "UuidProtocol\022\014\n\004high\030\001 \002(\004\022\013\n\003low\030\002 \002(\004\"" + - "3\n\025MetadataEntryProtocol\022\013\n\003key\030\001 \002(\t\022\r\n" + - "\005value\030\002 \002(\014\"6\n\021LifeCycleProtocol\022!\n\tlif" + - "eCycle\030\001 \002(\0162\016.LifeCycleType\"1\n\017AddressP", - "rotocol\022\020\n\010hostname\030\001 \002(\t\022\014\n\004port\030\002 \002(\r\"" + - "7\n\021ExceptionProtocol\022\021\n\tclassname\030\001 \002(\t\022" + - "\017\n\007message\030\002 \002(\t*=\n\tActorType\022\017\n\013SCALA_A" + - "CTOR\020\001\022\016\n\nJAVA_ACTOR\020\002\022\017\n\013TYPED_ACTOR\020\003*" + - "]\n\027SerializationSchemeType\022\010\n\004JAVA\020\001\022\013\n\007" + - "SBINARY\020\002\022\016\n\nSCALA_JSON\020\003\022\r\n\tJAVA_JSON\020\004" + - "\022\014\n\010PROTOBUF\020\005*-\n\rLifeCycleType\022\r\n\tPERMA" + - "NENT\020\001\022\r\n\tTEMPORARY\020\002B\030\n\024akka.remote.pro" + - "tocolH\001" + "\006 \001(\t\022\017\n\007timeout\030\007 \001(\004\022\026\n\016receiveTimeout" + + "\030\010 \001(\004\022%\n\tlifeCycle\030\t \001(\0132\022.LifeCyclePro" + + "tocol\022+\n\nsupervisor\030\n \001(\0132\027.RemoteActorR" + + "efProtocol\022\024\n\014hotswapStack\030\013 \001(\014\022(\n\010mess" + + "ages\030\014 \003(\0132\026.RemoteMessageProtocol\"g\n\037Se" + + "rializedTypedActorRefProtocol\022-\n\010actorRe" + + "f\030\001 \002(\0132\033.SerializedActorRefProtocol\022\025\n\r" + + "interfaceName\030\002 \002(\t\"r\n\017MessageProtocol\0225" + + "\n\023serializationScheme\030\001 \002(\0162\030.Serializat", + "ionSchemeType\022\017\n\007message\030\002 \002(\014\022\027\n\017messag" + + "eManifest\030\003 \001(\014\"\255\001\n\021ActorInfoProtocol\022\033\n" + + "\004uuid\030\001 \002(\0132\r.UuidProtocol\022\016\n\006target\030\002 \002" + + "(\t\022\017\n\007timeout\030\003 \002(\004\022\035\n\tactorType\030\004 \002(\0162\n" + + ".ActorType\022/\n\016typedActorInfo\030\005 \001(\0132\027.Typ" + + "edActorInfoProtocol\022\n\n\002id\030\006 \001(\t\";\n\026Typed" + + "ActorInfoProtocol\022\021\n\tinterface\030\001 \002(\t\022\016\n\006" + + "method\030\002 \002(\t\"\277\002\n\025RemoteMessageProtocol\022\033" + + "\n\004uuid\030\001 \002(\0132\r.UuidProtocol\022%\n\tactorInfo" + + "\030\002 \002(\0132\022.ActorInfoProtocol\022\016\n\006oneWay\030\003 \002", + "(\010\022!\n\007message\030\004 \001(\0132\020.MessageProtocol\022%\n" + + "\texception\030\005 \001(\0132\022.ExceptionProtocol\022%\n\016" + + "supervisorUuid\030\006 \001(\0132\r.UuidProtocol\022\'\n\006s" + + "ender\030\007 \001(\0132\027.RemoteActorRefProtocol\022(\n\010" + + "metadata\030\010 \003(\0132\026.MetadataEntryProtocol\022\016" + + "\n\006cookie\030\t \001(\t\")\n\014UuidProtocol\022\014\n\004high\030\001" + + " \002(\004\022\013\n\003low\030\002 \002(\004\"3\n\025MetadataEntryProtoc" + + "ol\022\013\n\003key\030\001 \002(\t\022\r\n\005value\030\002 \002(\014\"6\n\021LifeCy" + + "cleProtocol\022!\n\tlifeCycle\030\001 \002(\0162\016.LifeCyc" + + "leType\"1\n\017AddressProtocol\022\020\n\010hostname\030\001 ", + "\002(\t\022\014\n\004port\030\002 \002(\r\"7\n\021ExceptionProtocol\022\021" + + "\n\tclassname\030\001 \002(\t\022\017\n\007message\030\002 \002(\t*=\n\tAc" + + "torType\022\017\n\013SCALA_ACTOR\020\001\022\016\n\nJAVA_ACTOR\020\002" + + "\022\017\n\013TYPED_ACTOR\020\003*]\n\027SerializationScheme" + + "Type\022\010\n\004JAVA\020\001\022\013\n\007SBINARY\020\002\022\016\n\nSCALA_JSO" + + "N\020\003\022\r\n\tJAVA_JSON\020\004\022\014\n\010PROTOBUF\020\005*-\n\rLife" + + "CycleType\022\r\n\tPERMANENT\020\001\022\r\n\tTEMPORARY\020\002B" + + "\030\n\024akka.remote.protocolH\001" }; com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { @@ -6130,7 +6090,7 @@ public final class RemoteProtocol { internal_static_SerializedActorRefProtocol_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_SerializedActorRefProtocol_descriptor, - new java.lang.String[] { "Uuid", "Id", "ActorClassname", "OriginalAddress", "ActorInstance", "SerializerClassname", "IsTransactor", "Timeout", "ReceiveTimeout", "LifeCycle", "Supervisor", "HotswapStack", "Messages", }, + new java.lang.String[] { "Uuid", "Id", "ActorClassname", "OriginalAddress", "ActorInstance", "SerializerClassname", "Timeout", "ReceiveTimeout", "LifeCycle", "Supervisor", "HotswapStack", "Messages", }, akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.class, akka.remote.protocol.RemoteProtocol.SerializedActorRefProtocol.Builder.class); internal_static_SerializedTypedActorRefProtocol_descriptor = diff --git a/akka-remote/src/main/protocol/RemoteProtocol.proto b/akka-remote/src/main/protocol/RemoteProtocol.proto index 19282c53be..5c8b9b88f3 100644 --- a/akka-remote/src/main/protocol/RemoteProtocol.proto +++ b/akka-remote/src/main/protocol/RemoteProtocol.proto @@ -58,13 +58,12 @@ message SerializedActorRefProtocol { required AddressProtocol originalAddress = 4; optional bytes actorInstance = 5; optional string serializerClassname = 6; - optional bool isTransactor = 7; - optional uint64 timeout = 8; - optional uint64 receiveTimeout = 9; - optional LifeCycleProtocol lifeCycle = 10; - optional RemoteActorRefProtocol supervisor = 11; - optional bytes hotswapStack = 12; - repeated RemoteMessageProtocol messages = 13; + optional uint64 timeout = 7; + optional uint64 receiveTimeout = 8; + optional LifeCycleProtocol lifeCycle = 9; + optional RemoteActorRefProtocol supervisor = 10; + optional bytes hotswapStack = 11; + repeated RemoteMessageProtocol messages = 12; } /** diff --git a/akka-remote/src/main/scala/serialization/SerializationProtocol.scala b/akka-remote/src/main/scala/serialization/SerializationProtocol.scala index 21e45f6681..98f2161173 100644 --- a/akka-remote/src/main/scala/serialization/SerializationProtocol.scala +++ b/akka-remote/src/main/scala/serialization/SerializationProtocol.scala @@ -4,9 +4,6 @@ package akka.serialization -import akka.stm.global._ -import akka.stm.TransactionManagement._ -import akka.stm.TransactionManagement import akka.dispatch.MessageInvocation import akka.remote.{RemoteServer, RemoteClient, MessageSerializer} import akka.remote.protocol.RemoteProtocol.{ActorType => ActorTypeProtocol, _} @@ -111,7 +108,6 @@ object ActorSerialization { .setId(actorRef.id) .setActorClassname(actorRef.actorClass.getName) .setOriginalAddress(originalAddress) - .setIsTransactor(actorRef.isTransactor) .setTimeout(actorRef.timeout) @@ -197,7 +193,6 @@ object ActorSerialization { protocol.getId, protocol.getOriginalAddress.getHostname, protocol.getOriginalAddress.getPort, - if (protocol.hasIsTransactor) protocol.getIsTransactor else false, if (protocol.hasTimeout) protocol.getTimeout else Actor.TIMEOUT, if (protocol.hasReceiveTimeout) Some(protocol.getReceiveTimeout) else None, lifeCycle, diff --git a/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/PersistentSimpleServiceImpl.java b/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/PersistentSimpleServiceImpl.java index f59262d693..33136d6a2e 100644 --- a/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/PersistentSimpleServiceImpl.java +++ b/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/PersistentSimpleServiceImpl.java @@ -4,28 +4,38 @@ package sample.rest.java; -import akka.actor.TypedTransactor; +import akka.actor.TypedActor; +import akka.stm.local.Atomic; import akka.persistence.common.PersistentMap; import akka.persistence.cassandra.CassandraStorage; import java.nio.ByteBuffer; -public class PersistentSimpleServiceImpl extends TypedTransactor implements PersistentSimpleService { +public class PersistentSimpleServiceImpl extends TypedActor implements PersistentSimpleService { private String KEY = "COUNTER"; private boolean hasStartedTicking = false; - private PersistentMap storage; + private final PersistentMap storage = CassandraStorage.newMap(); public String count() { - if (storage == null) storage = CassandraStorage.newMap(); if (!hasStartedTicking) { - storage.put(KEY.getBytes(), ByteBuffer.allocate(4).putInt(0).array()); + new Atomic() { + public Object atomically() { + storage.put(KEY.getBytes(), ByteBuffer.allocate(4).putInt(0).array()); + return null; + } + }.execute(); hasStartedTicking = true; return "Tick: 0\n"; } else { - byte[] bytes = (byte[])storage.get(KEY.getBytes()).get(); - int counter = ByteBuffer.wrap(bytes).getInt(); - storage.put(KEY.getBytes(), ByteBuffer.allocate(4).putInt(counter + 1).array()); + int counter = new Atomic() { + public Integer atomically() { + byte[] bytes = (byte[])storage.get(KEY.getBytes()).get(); + int count = ByteBuffer.wrap(bytes).getInt() + 1; + storage.put(KEY.getBytes(), ByteBuffer.allocate(4).putInt(count).array()); + return count; + } + }.execute(); return "Tick: " + counter + "\n"; } } diff --git a/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/SimpleServiceImpl.java b/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/SimpleServiceImpl.java index 0eb7be096a..effcda5607 100644 --- a/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/SimpleServiceImpl.java +++ b/akka-samples/akka-sample-rest-java/src/main/java/sample/rest/java/SimpleServiceImpl.java @@ -5,27 +5,36 @@ package sample.rest.java; import akka.actor.TypedActor; -import akka.actor.TypedTransactor; +import akka.stm.local.Atomic; import akka.stm.TransactionalMap; -public class SimpleServiceImpl extends TypedTransactor implements SimpleService { +public class SimpleServiceImpl extends TypedActor implements SimpleService { private String KEY = "COUNTER"; private boolean hasStartedTicking = false; - private TransactionalMap storage; + private final TransactionalMap storage = new TransactionalMap(); private Receiver receiver = TypedActor.newInstance(Receiver.class, ReceiverImpl.class); public String count() { - if (storage == null) storage = new TransactionalMap(); if (!hasStartedTicking) { - storage.put(KEY, 0); + new Atomic() { + public Object atomically() { + storage.put(KEY, 0); + return null; + } + }.execute(); hasStartedTicking = true; return "Tick: 0\n"; } else { // Grabs the sender address and returns it //SimpleService sender = receiver.receive(); - int counter = (Integer)storage.get(KEY).get() + 1; - storage.put(KEY, counter); + int counter = new Atomic() { + public Integer atomically() { + int count = (Integer) storage.get(KEY).get() + 1; + storage.put(KEY, count); + return count; + } + }.execute(); return "Tick: " + counter + "\n"; } } diff --git a/akka-samples/akka-sample-rest-scala/src/main/scala/SimpleService.scala b/akka-samples/akka-sample-rest-scala/src/main/scala/SimpleService.scala index e6245cc7c6..fe743f0278 100644 --- a/akka-samples/akka-sample-rest-scala/src/main/scala/SimpleService.scala +++ b/akka-samples/akka-sample-rest-scala/src/main/scala/SimpleService.scala @@ -4,8 +4,9 @@ package sample.rest.scala -import akka.actor.{Transactor, SupervisorFactory, Actor} +import akka.actor.{SupervisorFactory, Actor} import akka.actor.Actor._ +import akka.stm.local._ import akka.stm.TransactionalMap import akka.persistence.cassandra.CassandraStorage import akka.config.Supervision._ @@ -59,18 +60,24 @@ class SimpleService { } } -class SimpleServiceActor extends Transactor { +class SimpleServiceActor extends Actor { private val KEY = "COUNTER" private var hasStartedTicking = false - private lazy val storage = TransactionalMap[String, Integer]() + private val storage = TransactionalMap[String, Integer]() def receive = { case "Tick" => if (hasStartedTicking) { - val counter = storage.get(KEY).get.asInstanceOf[Integer].intValue - storage.put(KEY, new Integer(counter + 1)) - self.reply(Tick:{counter + 1}) + val count = atomic { + val current = storage.get(KEY).get.asInstanceOf[Integer].intValue + val updated = current + 1 + storage.put(KEY, new Integer(updated)) + updated + } + self.reply(Tick:{count}) } else { - storage.put(KEY, new Integer(0)) + atomic { + storage.put(KEY, new Integer(0)) + } hasStartedTicking = true self.reply(Tick: 0) } @@ -113,22 +120,28 @@ class PersistentSimpleService { } } -class PersistentSimpleServiceActor extends Transactor { +class PersistentSimpleServiceActor extends Actor { private val KEY = "COUNTER" private var hasStartedTicking = false private lazy val storage = CassandraStorage.newMap def receive = { case "Tick" => if (hasStartedTicking) { - val bytes = storage.get(KEY.getBytes).get - val counter = Integer.parseInt(new String(bytes, "UTF8")) - storage.put(KEY.getBytes, (counter + 1).toString.getBytes ) + val count = atomic { + val bytes = storage.get(KEY.getBytes).get + val current = Integer.parseInt(new String(bytes, "UTF8")) + val updated = current + 1 + storage.put(KEY.getBytes, (updated).toString.getBytes) + updated + } // val bytes = storage.get(KEY.getBytes).get // val counter = ByteBuffer.wrap(bytes).getInt // storage.put(KEY.getBytes, ByteBuffer.allocate(4).putInt(counter + 1).array) - self.reply(Tick:{counter + 1}) + self.reply(Tick:{count}) } else { - storage.put(KEY.getBytes, "0".getBytes) + atomic { + storage.put(KEY.getBytes, "0".getBytes) + } // storage.put(KEY.getBytes, Array(0.toByte)) hasStartedTicking = true self.reply(Tick: 0) diff --git a/akka-samples/akka-sample-security/src/main/scala/SimpleService.scala b/akka-samples/akka-sample-security/src/main/scala/SimpleService.scala index d42ae12f2e..003979511e 100644 --- a/akka-samples/akka-sample-security/src/main/scala/SimpleService.scala +++ b/akka-samples/akka-sample-security/src/main/scala/SimpleService.scala @@ -4,11 +4,12 @@ package sample.security -import akka.actor.{SupervisorFactory, Transactor, Actor} +import akka.actor.{SupervisorFactory, Actor} import akka.actor.Actor._ import akka.config.Supervision._ import akka.util.Logging import akka.security.{BasicAuthenticationActor,BasicCredentials,SpnegoAuthenticationActor,DigestAuthenticationActor, UserInfo} +import akka.stm.local._ import akka.stm.TransactionalMap import akka.actor.ActorRegistry.actorFor @@ -21,14 +22,14 @@ class Boot { Supervise( actorOf[BasicAuthenticationService], Permanent) :: - /** - Supervise( - actorOf[DigestAuthenticationService], - Permanent) :: - Supervise( - actorOf[SpnegoAuthenticationService], - Permanent) :: - **/ + + // Supervise( + // actorOf[DigestAuthenticationService], + // Permanent) :: + // Supervise( + // actorOf[SpnegoAuthenticationService], + // Permanent) :: + Supervise( actorOf[SecureTickActor], Permanent):: Nil)) @@ -132,17 +133,23 @@ class SecureTickService { } } -class SecureTickActor extends Transactor with Logging { +class SecureTickActor extends Actor with Logging { private val KEY = "COUNTER" private var hasStartedTicking = false - private lazy val storage = TransactionalMap[String, Integer]() + private val storage = TransactionalMap[String, Integer]() def receive = { case "Tick" => if (hasStartedTicking) { - val counter = storage.get(KEY).get.intValue - storage.put(KEY, counter + 1) - self.reply(new Integer(counter + 1)) + val count = atomic { + val current = storage.get(KEY).get.intValue + val updated = current + 1 + storage.put(KEY, updated) + updated + } + self.reply(new Integer(count)) } else { - storage.put(KEY, 0) + atomic { + storage.put(KEY, 0) + } hasStartedTicking = true self.reply(new Integer(0)) } diff --git a/akka-spring/src/main/resources/akka/spring/akka-1.0-SNAPSHOT.xsd b/akka-spring/src/main/resources/akka/spring/akka-1.0-SNAPSHOT.xsd index ea438ebc88..16e3bde6f3 100644 --- a/akka-spring/src/main/resources/akka/spring/akka-1.0-SNAPSHOT.xsd +++ b/akka-spring/src/main/resources/akka/spring/akka-1.0-SNAPSHOT.xsd @@ -166,13 +166,6 @@ - - - - Set this to true if messages should have REQUIRES_NEW semantics. - - - @@ -219,13 +212,6 @@ - - - - Set this to true if messages should have REQUIRES_NEW semantics. - - - diff --git a/akka-spring/src/main/scala/ActorFactoryBean.scala b/akka-spring/src/main/scala/ActorFactoryBean.scala index c5014d6e18..0d3d407475 100644 --- a/akka-spring/src/main/scala/ActorFactoryBean.scala +++ b/akka-spring/src/main/scala/ActorFactoryBean.scala @@ -42,7 +42,6 @@ class ActorFactoryBean extends AbstractFactoryBean[AnyRef] with Logging with App @BeanProperty var implementation: String = "" @BeanProperty var beanRef: String = null @BeanProperty var timeoutStr: String = "" - @BeanProperty var transactional: Boolean = false @BeanProperty var host: String = "" @BeanProperty var port: String = "" @BeanProperty var serverManaged: Boolean = false @@ -140,9 +139,6 @@ class ActorFactoryBean extends AbstractFactoryBean[AnyRef] with Logging with App if (timeout > 0) { actorRef.setTimeout(timeout) } - if (transactional) { - actorRef.makeTransactionRequired - } if (isRemote) { if (serverManaged) { val server = RemoteServer.getOrCreateServer(new InetSocketAddress(host, port.toInt)) @@ -200,7 +196,6 @@ class ActorFactoryBean extends AbstractFactoryBean[AnyRef] with Logging with App private[akka] def createConfig: TypedActorConfiguration = { val config = new TypedActorConfiguration().timeout(Duration(timeout, "millis")) - if (transactional) config.makeTransactionRequired if (isRemote && !serverManaged) config.makeRemote(host, port.toInt) if (hasDispatcher) { if (dispatcher.dispatcherType != THREAD_BASED) { diff --git a/akka-spring/src/main/scala/ActorParser.scala b/akka-spring/src/main/scala/ActorParser.scala index 8466995dc2..da059d2f8e 100644 --- a/akka-spring/src/main/scala/ActorParser.scala +++ b/akka-spring/src/main/scala/ActorParser.scala @@ -60,9 +60,6 @@ trait ActorParser extends BeanParser with DispatcherParser { throw new IllegalArgumentException("Mandatory attribute missing, you need to provide either implementation or ref ") } - - objectProperties.transactional = if (element.getAttribute(TRANSACTIONAL).isEmpty) false else element.getAttribute(TRANSACTIONAL).toBoolean - if (element.hasAttribute(INTERFACE)) { objectProperties.interface = element.getAttribute(INTERFACE) } diff --git a/akka-spring/src/main/scala/ActorProperties.scala b/akka-spring/src/main/scala/ActorProperties.scala index c403de8da4..d0e7c49392 100644 --- a/akka-spring/src/main/scala/ActorProperties.scala +++ b/akka-spring/src/main/scala/ActorProperties.scala @@ -18,7 +18,6 @@ class ActorProperties { var beanRef: String = "" var timeoutStr: String = "" var interface: String = "" - var transactional: Boolean = false var host: String = "" var port: String = "" var serverManaged: Boolean = false @@ -43,7 +42,6 @@ class ActorProperties { builder.addPropertyValue(IMPLEMENTATION, target) builder.addPropertyValue("beanRef", beanRef) builder.addPropertyValue(INTERFACE, interface) - builder.addPropertyValue(TRANSACTIONAL, transactional) builder.addPropertyValue(LIFECYCLE, lifecycle) builder.addPropertyValue(SCOPE, scope) builder.addPropertyValue(DISPATCHER_TAG, dispatcher) diff --git a/akka-spring/src/main/scala/AkkaSpringConfigurationTags.scala b/akka-spring/src/main/scala/AkkaSpringConfigurationTags.scala index 67ca06d8fb..6548495070 100644 --- a/akka-spring/src/main/scala/AkkaSpringConfigurationTags.scala +++ b/akka-spring/src/main/scala/AkkaSpringConfigurationTags.scala @@ -45,7 +45,6 @@ object AkkaSpringConfigurationTags { val IMPLEMENTATION = "implementation" val BEANREF = "ref" val INTERFACE = "interface" - val TRANSACTIONAL = "transactional" val HOST = "host" val PORT = "port" val MANAGED_BY = "managed-by" diff --git a/akka-spring/src/main/scala/SupervisionFactoryBean.scala b/akka-spring/src/main/scala/SupervisionFactoryBean.scala index 4fd842f755..5e88374f18 100644 --- a/akka-spring/src/main/scala/SupervisionFactoryBean.scala +++ b/akka-spring/src/main/scala/SupervisionFactoryBean.scala @@ -61,15 +61,15 @@ class SupervisionFactoryBean extends AbstractFactoryBean[AnyRef] { //val remote = new RemoteAddress(props.host, props.port) val remote = new RemoteAddress(props.host, props.port.toInt) if (withInterface) { - new SuperviseTypedActor(props.interface.toClass, props.target.toClass, lifeCycle, props.timeout, props.transactional, remote) + new SuperviseTypedActor(props.interface.toClass, props.target.toClass, lifeCycle, props.timeout, remote) } else { - new SuperviseTypedActor(props.target.toClass, lifeCycle, props.timeout, props.transactional, remote) + new SuperviseTypedActor(props.target.toClass, lifeCycle, props.timeout, remote) } } else { if (withInterface) { - new SuperviseTypedActor(props.interface.toClass, props.target.toClass, lifeCycle, props.timeout, props.transactional) + new SuperviseTypedActor(props.interface.toClass, props.target.toClass, lifeCycle, props.timeout) } else { - new SuperviseTypedActor(props.target.toClass, lifeCycle, props.timeout, props.transactional) + new SuperviseTypedActor(props.target.toClass, lifeCycle, props.timeout) } } } @@ -85,9 +85,6 @@ class SupervisionFactoryBean extends AbstractFactoryBean[AnyRef] { if (props.timeout > 0) { actorRef.setTimeout(props.timeout) } - if (props.transactional) { - actorRef.makeTransactionRequired - } val supervise = if (isRemote) { val remote = new RemoteAddress(props.host, props.port.toInt) diff --git a/akka-spring/src/test/java/akka/spring/foo/StatefulPojo.java b/akka-spring/src/test/java/akka/spring/foo/StatefulPojo.java index 62f876a169..9f53371b5f 100644 --- a/akka-spring/src/test/java/akka/spring/foo/StatefulPojo.java +++ b/akka-spring/src/test/java/akka/spring/foo/StatefulPojo.java @@ -1,6 +1,6 @@ package akka.spring.foo; - +/* import akka.stm.TransactionalMap; import akka.stm.TransactionalVector; import akka.stm.Ref; @@ -27,7 +27,6 @@ public class StatefulPojo extends TypedActor { } } - /* public String getMapState(String key) { return (String)mapState.get(key).get(); } @@ -55,6 +54,5 @@ public class StatefulPojo extends TypedActor { public boolean isInitialized() { return isInitialized; } - */ - } +*/ diff --git a/akka-spring/src/test/resources/failing-appContext.xml b/akka-spring/src/test/resources/failing-appContext.xml index 2fff521e28..28187fe4ef 100644 --- a/akka-spring/src/test/resources/failing-appContext.xml +++ b/akka-spring/src/test/resources/failing-appContext.xml @@ -9,7 +9,6 @@ diff --git a/akka-spring/src/test/resources/supervisor-config.xml b/akka-spring/src/test/resources/supervisor-config.xml index 23917b4e50..8dcdc25c56 100644 --- a/akka-spring/src/test/resources/supervisor-config.xml +++ b/akka-spring/src/test/resources/supervisor-config.xml @@ -59,7 +59,7 @@ http://scalablesolutions.se/akka/akka-1.0-SNAPSHOT.xsd"> + timeout="10000"/> --> diff --git a/akka-spring/src/test/resources/typed-actor-config.xml b/akka-spring/src/test/resources/typed-actor-config.xml index 695be2f31d..7b994fca28 100644 --- a/akka-spring/src/test/resources/typed-actor-config.xml +++ b/akka-spring/src/test/resources/typed-actor-config.xml @@ -27,24 +27,16 @@ http://scalablesolutions.se/akka/akka-1.0-SNAPSHOT.xsd"> implementation="akka.spring.foo.MyPojo" timeout="10000"/> - - + timeout="2000"> + timeout="2000"> diff --git a/akka-spring/src/test/resources/untyped-actor-config.xml b/akka-spring/src/test/resources/untyped-actor-config.xml index dca630a028..e827f4c1de 100644 --- a/akka-spring/src/test/resources/untyped-actor-config.xml +++ b/akka-spring/src/test/resources/untyped-actor-config.xml @@ -21,11 +21,6 @@ http://scalablesolutions.se/akka/akka-1.0-SNAPSHOT.xsd"> implementation="akka.spring.foo.PingActor" timeout="10000"/> - - diff --git a/akka-spring/src/test/scala/SupervisionBeanDefinitionParserTest.scala b/akka-spring/src/test/scala/SupervisionBeanDefinitionParserTest.scala index f033930ce1..e5b4fc2c70 100644 --- a/akka-spring/src/test/scala/SupervisionBeanDefinitionParserTest.scala +++ b/akka-spring/src/test/scala/SupervisionBeanDefinitionParserTest.scala @@ -30,7 +30,6 @@ class SupervisionBeanDefinitionParserTest extends Spec with ShouldMatchers { assert(props ne null) assert(props.timeout == 1000) assert(props.target == "foo.bar.MyPojo") - assert(props.transactional) } it("should parse the supervisor restart strategy") { @@ -72,8 +71,7 @@ class SupervisionBeanDefinitionParserTest extends Spec with ShouldMatchers { private def createTypedActorElement : Element = { val xml = + timeout="1000"/> dom(xml).getDocumentElement } diff --git a/akka-spring/src/test/scala/TypedActorBeanDefinitionParserTest.scala b/akka-spring/src/test/scala/TypedActorBeanDefinitionParserTest.scala index 36b8588861..f665784355 100644 --- a/akka-spring/src/test/scala/TypedActorBeanDefinitionParserTest.scala +++ b/akka-spring/src/test/scala/TypedActorBeanDefinitionParserTest.scala @@ -25,7 +25,6 @@ class TypedActorBeanDefinitionParserTest extends Spec with ShouldMatchers { val xml = @@ -34,15 +33,12 @@ class TypedActorBeanDefinitionParserTest extends Spec with ShouldMatchers { assert(props ne null) assert(props.timeout === 1000) assert(props.target === "foo.bar.MyPojo") - assert(props.transactional) assert(props.scope === "prototype") assert(props.propertyEntries.entryList.size === 1) } it("should throw IllegalArgumentException on missing mandatory attributes") { - val xml = + val xml = evaluating { parser.parseActor(dom(xml).getDocumentElement) } should produce [IllegalArgumentException] } diff --git a/akka-spring/src/test/scala/TypedActorSpringFeatureTest.scala b/akka-spring/src/test/scala/TypedActorSpringFeatureTest.scala index d9f9526b95..36ff8d9f06 100644 --- a/akka-spring/src/test/scala/TypedActorSpringFeatureTest.scala +++ b/akka-spring/src/test/scala/TypedActorSpringFeatureTest.scala @@ -99,14 +99,6 @@ class TypedActorSpringFeatureTest extends FeatureSpec with ShouldMatchers with B assert(myPojo.longRunning() === "this took long"); } - scenario("transactional typed-actor") { - val myPojo = getTypedActorFromContext("/typed-actor-config.xml", "transactional-typed-actor") - assert(myPojo.getFoo() === "foo") - myPojo.oneWay("hello 2") - MyPojo.latch.await - assert(MyPojo.lastOneWayMessage === "hello 2") - } - scenario("get a remote typed-actor") { val myPojo = getTypedActorFromContext("/typed-actor-config.xml", "remote-typed-actor") assert(myPojo.getFoo() === "foo") diff --git a/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala b/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala index b2aaed7c1f..aded68a559 100644 --- a/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala +++ b/akka-spring/src/test/scala/UntypedActorSpringFeatureTest.scala @@ -83,13 +83,6 @@ class UntypedActorSpringFeatureTest extends FeatureSpec with ShouldMatchers with assert(PingActor.lastMessage === "Hello 2") } - scenario("transactional untyped-actor") { - val myactor = getPingActorFromContext("/untyped-actor-config.xml", "transactional-untyped-actor") - myactor.sendOneWay("Hello 3") - PingActor.latch.await - assert(PingActor.lastMessage === "Hello 3") - } - scenario("get a remote typed-actor") { val myactor = getPingActorFromContext("/untyped-actor-config.xml", "remote-untyped-actor") myactor.sendOneWay("Hello 4") diff --git a/akka-actor/src/main/scala/actor/Agent.scala b/akka-stm/disabled-src/main/scala/actor/Agent.scala similarity index 100% rename from akka-actor/src/main/scala/actor/Agent.scala rename to akka-stm/disabled-src/main/scala/actor/Agent.scala diff --git a/akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/NestedTransactionalTypedActor.java b/akka-stm/disabled-src/test/java/akka/actor/NestedTransactionalTypedActor.java similarity index 100% rename from akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/NestedTransactionalTypedActor.java rename to akka-stm/disabled-src/test/java/akka/actor/NestedTransactionalTypedActor.java diff --git a/akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/NestedTransactionalTypedActorImpl.java b/akka-stm/disabled-src/test/java/akka/actor/NestedTransactionalTypedActorImpl.java similarity index 100% rename from akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/NestedTransactionalTypedActorImpl.java rename to akka-stm/disabled-src/test/java/akka/actor/NestedTransactionalTypedActorImpl.java diff --git a/akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/TransactionalTypedActor.java b/akka-stm/disabled-src/test/java/akka/actor/TransactionalTypedActor.java similarity index 100% rename from akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/TransactionalTypedActor.java rename to akka-stm/disabled-src/test/java/akka/actor/TransactionalTypedActor.java diff --git a/akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/TransactionalTypedActorImpl.java b/akka-stm/disabled-src/test/java/akka/actor/TransactionalTypedActorImpl.java similarity index 100% rename from akka-typed-actor/src/test/java/se/scalablesolutions/akka/actor/TransactionalTypedActorImpl.java rename to akka-stm/disabled-src/test/java/akka/actor/TransactionalTypedActorImpl.java diff --git a/akka-actor/src/test/scala/actor/actor/AgentSpec.scala b/akka-stm/disabled-src/test/scala/actor/AgentSpec.scala similarity index 100% rename from akka-actor/src/test/scala/actor/actor/AgentSpec.scala rename to akka-stm/disabled-src/test/scala/actor/AgentSpec.scala diff --git a/akka-actor/src/test/scala/actor/actor/TransactorSpec.scala b/akka-stm/disabled-src/test/scala/actor/TransactorSpec.scala similarity index 100% rename from akka-actor/src/test/scala/actor/actor/TransactorSpec.scala rename to akka-stm/disabled-src/test/scala/actor/TransactorSpec.scala diff --git a/akka-remote/src/test/scala/remote/RemoteAgentSpec.scala b/akka-stm/disabled-src/test/scala/remote/RemoteAgentSpec.scala similarity index 100% rename from akka-remote/src/test/scala/remote/RemoteAgentSpec.scala rename to akka-stm/disabled-src/test/scala/remote/RemoteAgentSpec.scala diff --git a/akka-remote/src/test/scala/remote/RemoteTransactionalTypedActorSpec.scala b/akka-stm/disabled-src/test/scala/remote/RemoteTransactionalTypedActorSpec.scala similarity index 100% rename from akka-remote/src/test/scala/remote/RemoteTransactionalTypedActorSpec.scala rename to akka-stm/disabled-src/test/scala/remote/RemoteTransactionalTypedActorSpec.scala diff --git a/akka-typed-actor/src/test/scala/actor/typed-actor/NestedTransactionalTypedActorSpec.scala b/akka-stm/disabled-src/test/scala/typed-actor/NestedTransactionalTypedActorSpec.scala similarity index 100% rename from akka-typed-actor/src/test/scala/actor/typed-actor/NestedTransactionalTypedActorSpec.scala rename to akka-stm/disabled-src/test/scala/typed-actor/NestedTransactionalTypedActorSpec.scala diff --git a/akka-typed-actor/src/test/scala/actor/typed-actor/RestartNestedTransactionalTypedActorSpec.scala b/akka-stm/disabled-src/test/scala/typed-actor/RestartNestedTransactionalTypedActorSpec.scala similarity index 100% rename from akka-typed-actor/src/test/scala/actor/typed-actor/RestartNestedTransactionalTypedActorSpec.scala rename to akka-stm/disabled-src/test/scala/typed-actor/RestartNestedTransactionalTypedActorSpec.scala diff --git a/akka-typed-actor/src/test/scala/actor/typed-actor/RestartTransactionalTypedActorSpec.scala b/akka-stm/disabled-src/test/scala/typed-actor/RestartTransactionalTypedActorSpec.scala similarity index 100% rename from akka-typed-actor/src/test/scala/actor/typed-actor/RestartTransactionalTypedActorSpec.scala rename to akka-stm/disabled-src/test/scala/typed-actor/RestartTransactionalTypedActorSpec.scala diff --git a/akka-typed-actor/src/test/scala/actor/typed-actor/TransactionalTypedActorSpec.scala b/akka-stm/disabled-src/test/scala/typed-actor/TransactionalTypedActorSpec.scala similarity index 100% rename from akka-typed-actor/src/test/scala/actor/typed-actor/TransactionalTypedActorSpec.scala rename to akka-stm/disabled-src/test/scala/typed-actor/TransactionalTypedActorSpec.scala diff --git a/akka-actor/src/main/scala/stm/Ref.scala b/akka-stm/src/main/scala/stm/Ref.scala similarity index 100% rename from akka-actor/src/main/scala/stm/Ref.scala rename to akka-stm/src/main/scala/stm/Ref.scala diff --git a/akka-actor/src/main/scala/stm/Transaction.scala b/akka-stm/src/main/scala/stm/Transaction.scala similarity index 88% rename from akka-actor/src/main/scala/stm/Transaction.scala rename to akka-stm/src/main/scala/stm/Transaction.scala index 59a9d921e7..eba7356ba6 100644 --- a/akka-actor/src/main/scala/stm/Transaction.scala +++ b/akka-stm/src/main/scala/stm/Transaction.scala @@ -9,10 +9,9 @@ import java.util.concurrent.atomic.AtomicInteger import scala.collection.mutable.HashMap -import akka.util.ReflectiveAccess.JtaModule - -import akka.util.Logging +import akka.util.{Logging, ReflectiveAccess} import akka.config.Config._ +import akka.config.ModuleNotAvailableException import akka.AkkaException import org.multiverse.api.{Transaction => MultiverseTransaction} @@ -211,3 +210,27 @@ trait Abortable { def abort(): Unit } +object JtaModule { + type TransactionContainerObject = { + def apply(): TransactionContainer + } + + type TransactionContainer = { + def beginWithStmSynchronization(transaction: Transaction): Unit + def commit: Unit + def rollback: Unit + } + + lazy val isJtaEnabled = transactionContainerObjectInstance.isDefined + + def ensureJtaEnabled = if (!isJtaEnabled) throw new ModuleNotAvailableException( + "Can't load the typed actor module, make sure that akka-jta.jar is on the classpath") + + val transactionContainerObjectInstance: Option[TransactionContainerObject] = + ReflectiveAccess.getObjectFor("akka.jta.TransactionContainer$") + + def createTransactionContainer: TransactionContainer = { + ensureJtaEnabled + transactionContainerObjectInstance.get.apply.asInstanceOf[TransactionContainer] + } +} diff --git a/akka-actor/src/main/scala/stm/TransactionFactory.scala b/akka-stm/src/main/scala/stm/TransactionFactory.scala similarity index 100% rename from akka-actor/src/main/scala/stm/TransactionFactory.scala rename to akka-stm/src/main/scala/stm/TransactionFactory.scala diff --git a/akka-actor/src/main/scala/stm/TransactionFactoryBuilder.scala b/akka-stm/src/main/scala/stm/TransactionFactoryBuilder.scala similarity index 100% rename from akka-actor/src/main/scala/stm/TransactionFactoryBuilder.scala rename to akka-stm/src/main/scala/stm/TransactionFactoryBuilder.scala diff --git a/akka-actor/src/main/scala/stm/TransactionManagement.scala b/akka-stm/src/main/scala/stm/TransactionManagement.scala similarity index 100% rename from akka-actor/src/main/scala/stm/TransactionManagement.scala rename to akka-stm/src/main/scala/stm/TransactionManagement.scala diff --git a/akka-actor/src/main/scala/stm/TransactionalMap.scala b/akka-stm/src/main/scala/stm/TransactionalMap.scala similarity index 100% rename from akka-actor/src/main/scala/stm/TransactionalMap.scala rename to akka-stm/src/main/scala/stm/TransactionalMap.scala diff --git a/akka-actor/src/main/scala/stm/TransactionalVector.scala b/akka-stm/src/main/scala/stm/TransactionalVector.scala similarity index 100% rename from akka-actor/src/main/scala/stm/TransactionalVector.scala rename to akka-stm/src/main/scala/stm/TransactionalVector.scala diff --git a/akka-actor/src/main/scala/stm/global/Atomic.scala b/akka-stm/src/main/scala/stm/global/Atomic.scala similarity index 100% rename from akka-actor/src/main/scala/stm/global/Atomic.scala rename to akka-stm/src/main/scala/stm/global/Atomic.scala diff --git a/akka-actor/src/main/scala/stm/global/GlobalStm.scala b/akka-stm/src/main/scala/stm/global/GlobalStm.scala similarity index 100% rename from akka-actor/src/main/scala/stm/global/GlobalStm.scala rename to akka-stm/src/main/scala/stm/global/GlobalStm.scala diff --git a/akka-actor/src/main/scala/stm/global/package.scala b/akka-stm/src/main/scala/stm/global/package.scala similarity index 100% rename from akka-actor/src/main/scala/stm/global/package.scala rename to akka-stm/src/main/scala/stm/global/package.scala diff --git a/akka-actor/src/main/scala/stm/local/Atomic.scala b/akka-stm/src/main/scala/stm/local/Atomic.scala similarity index 100% rename from akka-actor/src/main/scala/stm/local/Atomic.scala rename to akka-stm/src/main/scala/stm/local/Atomic.scala diff --git a/akka-actor/src/main/scala/stm/local/LocalStm.scala b/akka-stm/src/main/scala/stm/local/LocalStm.scala similarity index 100% rename from akka-actor/src/main/scala/stm/local/LocalStm.scala rename to akka-stm/src/main/scala/stm/local/LocalStm.scala diff --git a/akka-actor/src/main/scala/stm/local/package.scala b/akka-stm/src/main/scala/stm/local/package.scala similarity index 100% rename from akka-actor/src/main/scala/stm/local/package.scala rename to akka-stm/src/main/scala/stm/local/package.scala diff --git a/akka-actor/src/main/scala/stm/transactional.scala b/akka-stm/src/main/scala/stm/transactional.scala similarity index 100% rename from akka-actor/src/main/scala/stm/transactional.scala rename to akka-stm/src/main/scala/stm/transactional.scala diff --git a/akka-actor/src/test/java/akka/stm/Address.java b/akka-stm/src/test/java/akka/stm/Address.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/Address.java rename to akka-stm/src/test/java/akka/stm/Address.java diff --git a/akka-actor/src/test/java/akka/stm/CounterExample.java b/akka-stm/src/test/java/akka/stm/CounterExample.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/CounterExample.java rename to akka-stm/src/test/java/akka/stm/CounterExample.java diff --git a/akka-actor/src/test/java/akka/stm/JavaStmTests.java b/akka-stm/src/test/java/akka/stm/JavaStmTests.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/JavaStmTests.java rename to akka-stm/src/test/java/akka/stm/JavaStmTests.java diff --git a/akka-actor/src/test/java/akka/stm/RefExample.java b/akka-stm/src/test/java/akka/stm/RefExample.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/RefExample.java rename to akka-stm/src/test/java/akka/stm/RefExample.java diff --git a/akka-actor/src/test/java/akka/stm/StmExamples.java b/akka-stm/src/test/java/akka/stm/StmExamples.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/StmExamples.java rename to akka-stm/src/test/java/akka/stm/StmExamples.java diff --git a/akka-actor/src/test/java/akka/stm/TransactionFactoryExample.java b/akka-stm/src/test/java/akka/stm/TransactionFactoryExample.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/TransactionFactoryExample.java rename to akka-stm/src/test/java/akka/stm/TransactionFactoryExample.java diff --git a/akka-actor/src/test/java/akka/stm/TransactionalMapExample.java b/akka-stm/src/test/java/akka/stm/TransactionalMapExample.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/TransactionalMapExample.java rename to akka-stm/src/test/java/akka/stm/TransactionalMapExample.java diff --git a/akka-actor/src/test/java/akka/stm/TransactionalVectorExample.java b/akka-stm/src/test/java/akka/stm/TransactionalVectorExample.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/TransactionalVectorExample.java rename to akka-stm/src/test/java/akka/stm/TransactionalVectorExample.java diff --git a/akka-actor/src/test/java/akka/stm/User.java b/akka-stm/src/test/java/akka/stm/User.java similarity index 100% rename from akka-actor/src/test/java/akka/stm/User.java rename to akka-stm/src/test/java/akka/stm/User.java diff --git a/akka-actor/src/test/scala/stm/JavaStmSpec.scala b/akka-stm/src/test/scala/stm/JavaStmSpec.scala similarity index 100% rename from akka-actor/src/test/scala/stm/JavaStmSpec.scala rename to akka-stm/src/test/scala/stm/JavaStmSpec.scala diff --git a/akka-actor/src/test/scala/stm/RefSpec.scala b/akka-stm/src/test/scala/stm/RefSpec.scala similarity index 100% rename from akka-actor/src/test/scala/stm/RefSpec.scala rename to akka-stm/src/test/scala/stm/RefSpec.scala diff --git a/akka-actor/src/test/scala/stm/StmSpec.scala b/akka-stm/src/test/scala/stm/StmSpec.scala similarity index 99% rename from akka-actor/src/test/scala/stm/StmSpec.scala rename to akka-stm/src/test/scala/stm/StmSpec.scala index a13615f6f2..ed91d0720b 100644 --- a/akka-actor/src/test/scala/stm/StmSpec.scala +++ b/akka-stm/src/test/scala/stm/StmSpec.scala @@ -1,6 +1,6 @@ package akka.stm -import akka.actor.{Actor, Transactor} +import akka.actor.Actor import Actor._ import org.multiverse.api.exceptions.ReadonlyException @@ -126,7 +126,9 @@ class StmSpec extends WordSpec with MustMatchers { readonlyOuter must be (2) } } +} +/* "Global STM" should { "be able to initialize with atomic {..} block inside actor constructor" in { import GlobalTransactionVectorTestActor._ @@ -146,7 +148,6 @@ class StmSpec extends WordSpec with MustMatchers { } } -/* "Transactor" should { "be able receive message sent with !! and pass it along to nested transactor with !! and receive reply; multiple times in a row" in { import GlobalTransactionVectorTestActor._ @@ -167,7 +168,6 @@ class StmSpec extends WordSpec with MustMatchers { size4 must be (3) } } - */ } object GlobalTransactionVectorTestActor { @@ -228,3 +228,4 @@ class NestedTransactorLevelTwoActor extends Transactor { case "HiLevelTwo" => println("HiLevelTwo") } } +*/ diff --git a/akka-typed-actor/src/main/scala/actor/TypedActor.scala b/akka-typed-actor/src/main/scala/actor/TypedActor.scala index 3ad5468369..8ca5f7094f 100644 --- a/akka-typed-actor/src/main/scala/actor/TypedActor.scala +++ b/akka-typed-actor/src/main/scala/actor/TypedActor.scala @@ -233,16 +233,6 @@ abstract class TypedActor extends Actor with Proxyable { } } -/** - * Transactional TypedActor. All messages send to this actor as sent in a transaction. If an enclosing transaction - * exists it will be joined, if not then a new transaction will be created. - * - * @author Jonas Bonér - */ -abstract class TypedTransactor extends TypedActor { - self.makeTransactionRequired -} - /** * Holds RTTI (runtime type information) for the TypedActor, f.e. current 'sender' * reference, the 'senderFuture' reference etc. @@ -327,12 +317,6 @@ object TypedActorConfiguration { def apply(host: String, port: Int, timeout: Long) : TypedActorConfiguration = { new TypedActorConfiguration().makeRemote(host, port).timeout(Duration(timeout, "millis")) } - - def apply(transactionRequired: Boolean) : TypedActorConfiguration = { - if (transactionRequired) { - new TypedActorConfiguration().makeTransactionRequired - } else new TypedActorConfiguration() - } } /** @@ -342,7 +326,6 @@ object TypedActorConfiguration { */ final class TypedActorConfiguration { private[akka] var _timeout: Long = Actor.TIMEOUT - private[akka] var _transactionRequired = false private[akka] var _host: Option[InetSocketAddress] = None private[akka] var _messageDispatcher: Option[MessageDispatcher] = None private[akka] var _threadBasedDispatcher: Option[Boolean] = None @@ -353,11 +336,6 @@ final class TypedActorConfiguration { this } - def makeTransactionRequired() : TypedActorConfiguration = { - _transactionRequired = true; - this - } - def makeRemote(hostname: String, port: Int): TypedActorConfiguration = makeRemote(new InetSocketAddress(hostname, port)) def makeRemote(remoteAddress: InetSocketAddress): TypedActorConfiguration = { @@ -700,12 +678,6 @@ object TypedActor extends Logging { this } - def isTransactional(clazz: Class[_]): Boolean = { - if (clazz eq null) false - else if (clazz.isAssignableFrom(classOf[TypedTransactor])) true - else isTransactional(clazz.getSuperclass) - } - private[akka] def newTypedActor(targetClass: Class[_]): TypedActor = { val instance = targetClass.newInstance val typedActor = diff --git a/akka-typed-actor/src/main/scala/config/TypedActorGuiceConfigurator.scala b/akka-typed-actor/src/main/scala/config/TypedActorGuiceConfigurator.scala index 0f7532ea0f..dc2e010923 100644 --- a/akka-typed-actor/src/main/scala/config/TypedActorGuiceConfigurator.scala +++ b/akka-typed-actor/src/main/scala/config/TypedActorGuiceConfigurator.scala @@ -88,7 +88,7 @@ private[akka] class TypedActorGuiceConfigurator extends TypedActorConfiguratorBa val targetClass = if (component.target.isInstanceOf[Class[_ <: TypedActor]]) component.target.asInstanceOf[Class[_ <: TypedActor]] else throw new IllegalArgumentException("TypedActor [" + component.target.getName + "] must be a subclass of TypedActor") - val actorRef = Actor.actorOf(new Dispatcher(component.transactionRequired)) + val actorRef = Actor.actorOf(new Dispatcher()) if (component.dispatcher.isDefined) actorRef.dispatcher = component.dispatcher.get val remoteAddress = if (component.remoteAddress.isDefined) diff --git a/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.jar b/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.jar deleted file mode 100644 index 0929a5bc75..0000000000 Binary files a/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.jar and /dev/null differ diff --git a/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.pom b/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.pom deleted file mode 100644 index 32e32fc857..0000000000 --- a/embedded-repo/org/multiverse/multiverse-alpha/0.6-2010-07-15/multiverse-alpha-0.6-2010-07-15.pom +++ /dev/null @@ -1,261 +0,0 @@ - - - 4.0.0 - - multiverse-alpha - Alpha Multiverse STM engine - - Contains an all in one jar that that contains the AlphaStm including the Multiverse - Javaagent and the Multiverse Compiler. This is the JAR you want to include in your - projects, if you do, you don't need to worry about any Multiverse dependency - at all. - - jar - 0.6-2010-07-15 - - - org.multiverse - multiverse - 0.6-2010-07-15 - - - - org.multiverse.javaagent.MultiverseJavaAgent - org.multiverse.stms.alpha.instrumentation.AlphaStmInstrumentor - - - - - - - maven-antrun-plugin - - - - create-main-jar - compile - - run - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - Compiles the tests - test-compile - - run - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ant - ant - 1.7.0 - - - junit - junit - ${junit.version} - - - ant - optional - 1.5.4 - - - com.tonicsystems.jarjar - jarjar - 1.0-rc8 - - - - - - org.apache.maven.plugins - maven-jar-plugin - - - - ${multiverse.agentclass} - - ${project.build.outputDirectory}/META-INF/MANIFEST.MF - - - - - - - - - - ${project.groupId} - multiverse-alpha-unborn - ${project.version} - - - ${project.groupId} - multiverse-core - ${project.version} - - - ${project.groupId} - multiverse-instrumentation - ${project.version} - - - ${project.groupId} - multiverse-alpha-unborn - ${project.version} - test - - - - args4j - args4j - ${args4j.version} - provided - - - - com.tonicsystems.jarjar - jarjar - 1.0-rc8 - provided - - - - asm - asm-all - ${asm.version} - - - diff --git a/embedded-repo/org/multiverse/multiverse/0.6-2010-07-15/multiverse-0.6-2010-07-15.pom b/embedded-repo/org/multiverse/multiverse/0.6-2010-07-15/multiverse-0.6-2010-07-15.pom deleted file mode 100644 index 5d89a92a8c..0000000000 --- a/embedded-repo/org/multiverse/multiverse/0.6-2010-07-15/multiverse-0.6-2010-07-15.pom +++ /dev/null @@ -1,488 +0,0 @@ - - - 4.0.0 - - Multiverse Software Transactional Memory - - Multiverse is a Software Transactional Memory implementation that can be used in Java - but also in other languages running on the JVM like Scala or Groovy. Essentially it is a framework that allows - different STM implementation (with different featuresets or performance characteristics) to be used - under the hood. The main STM implementation is multiverse-alpha.. - - org.multiverse - multiverse - 0.6-2010-07-15 - 2008 - pom - - - UTF-8 - 1.6 - 3.2 - 2.0.16 - 4.8.1 - 1.8.2 - - - - - The Apache License, ASL Version 2.0 - http://www.apache.org/licenses/LICENSE-2.0 - - - - - Multiverse - http://multiverse.codehaus.org - - - - - pveentjer - Peter Veentjer - +1 - alarmnummer AT gmail DOTCOM - - Founder - - - - aphillips - Andrew Phillips - +1 - aphillips AT qrmedia DOTCOM - - Committer - - - - - - - maven.atlassian.com - Atlassian Maven Proxy - https://maven.atlassian.com/content/groups/public - - - repo1.maven - Maven Main Repository - http://repo1.maven.org/maven2 - - - maven2-repository.dev.java.net - Java.net Repository for Maven - http://download.java.net/maven/2 - - - java.net - Java.net Legacy Repository for Maven - http://download.java.net/maven/1 - legacy - - - google-maven-repository - Google Maven Repository - http://google-maven-repository.googlecode.com/svn/repository/ - - - repository.codehaus.org - Codehaus Maven Repository - http://repository.codehaus.org - - - ibiblio - http://www.ibiblio.org/maven - - - sourceforge - http://maven-plugins.sourceforge.net/repository - - - - mandubian-mvn - http://mandubian-mvn.googlecode.com/svn/trunk/mandubian-mvn/repository - - - - - - snapshots - http://snapshots.maven.codehaus.org/maven2 - - - - - multiverse-benchy - - multiverse-core - multiverse-core-tests - multiverse-instrumentation - multiverse-alpha-unborn - multiverse-alpha - - multiverse-site - - multiverse-performance-tool - - - - - - - maven-compiler-plugin - - ${sourceEncoding} - ${targetJdk} - ${targetJdk} - - - - maven-resources-plugin - - ${sourceEncoding} - - - - maven-surefire-plugin - - - **/*LongTest.java - **/*longTest.java - **/*StressTest.java - **/*stressTest.java - **/*PerformanceTest.java - **/*performanceTest.java - - - **/*Test.java - - once - - - - - - - maven-enforcer-plugin - - - enforce-java - - enforce - - - - - ${targetJdk} - - - - - - - - - maven-source-plugin - - - attach-sources - - jar - - - - - - - - - - org.apache.maven.wagon - wagon-webdav - 1.0-beta-2 - - - org.apache.maven.wagon - wagon-ftp - 1.0-beta-6 - - - - - - - junit - junit - ${junit.version} - test - - - org.mockito - mockito-all - ${mockito.version} - test - - - - - scm:git:git://git.codehaus.org/multiverse.git - scm:git:ssh://git@git.codehaus.org/multiverse.git - http://git.codehaus.org/gitweb.cgi?p=multiverse.git - - - - Jira - http://jira.codehaus.org/browse/MULTIVERSE - - - - - Development List - dev-subscribe@multiverse.codehaus.org - dev-unsubscribe@multiverse.codehaus.org - dev@multiverse.codehaus.org - http://archive.multiverse.codehaus.org/dev - - - - User List - user-subscribe@multiverse.codehaus.org - user-unsubscribe@multiverse.codehaus.org - user@multiverse.codehaus.org - http://archive.multiverse.codehaus.org/user - - - - Commits List - scm-subscribe@multiverse.codehaus.org - scm-unsubscribe@multiverse.codehaus.org - http://archive.multiverse.codehaus.org/scm - - - - - - - maven-javadoc-plugin - 2.6.1 - - ${sourceEncoding} - true - - - - default - - aggregate - - - - - - org.codehaus.mojo - jxr-maven-plugin - - - org.codehaus.mojo - taglist-maven-plugin - 2.3 - - ${sourceEncoding} - - FIXME - Fixme - fixme - TODO - todo - Todo - @todo - @deprecated - - - - - maven-project-info-reports-plugin - - - maven-changes-plugin - 2.0-beta-3 - - - - changes-report - - - - - ${basedir}/changes.xml - - - - maven-surefire-report-plugin - - false - - - - - report-only - - - - - - org.codehaus.mojo - findbugs-maven-plugin - 2.0.1 - - - maven-pmd-plugin - 2.3 - - ${sourceEncoding} - ${targetJdk} - - - - - - - - - multiverse-releases - Multiverse Central Repository - dav:https://dav.codehaus.org/repository/multiverse/ - - - multiverse-snapshots - Multiverse Central Development Repository - dav:https://dav.codehaus.org/snapshots.repository/multiverse/ - - - http://dist.codehaus.org/multiverse/ - - - - - release - - - - - maven-javadoc-plugin - 2.6.1 - false - - - generate-resources - - aggregate - - - - - ${sourceEncoding} - true - - - - - maven-assembly-plugin - 2.2-beta-2 - false - - - distribution - package - - single - - - - distribution.xml - - - - - - - - - - - stress - - - - org.apache.maven.plugins - maven-surefire-plugin - - - **/*LongTest.java - - -Xmx256m - once - - - - - - - diff --git a/project/build/AkkaProject.scala b/project/build/AkkaProject.scala index 398d502e3c..cf76facc7c 100644 --- a/project/build/AkkaProject.scala +++ b/project/build/AkkaProject.scala @@ -1,6 +1,6 @@ - /*---------------------------------------------------------------------------\ -| Copyright (C) 2009-2010 Scalable Solutions AB | -\---------------------------------------------------------------------------*/ +/** + * Copyright (C) 2009-2010 Scalable Solutions AB + */ import com.weiglewilczek.bnd4sbt.BNDPlugin import java.io.File @@ -203,6 +203,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { lazy val casbah = "com.novus" % "casbah_2.8.0" % "1.0.8.5" % "compile" //ApacheV2 lazy val multiverse = "org.multiverse" % "multiverse-alpha" % MULTIVERSE_VERSION % "compile" intransitive //ApacheV2 + lazy val multiverse_test = "org.multiverse" % "multiverse-alpha" % MULTIVERSE_VERSION % "test" intransitive //ApacheV2 lazy val netty = "org.jboss.netty" % "netty" % "3.2.3.Final" % "compile" //ApacheV2 @@ -288,6 +289,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { // ------------------------------------------------------------------------------------------------------------------- lazy val akka_actor = project("akka-actor", "akka-actor", new AkkaActorProject(_)) + lazy val akka_stm = project("akka-stm", "akka-stm", new AkkaStmProject(_), akka_actor) lazy val akka_typed_actor = project("akka-typed-actor", "akka-typed-actor", new AkkaTypedActorProject(_), akka_actor) lazy val akka_remote = project("akka-remote", "akka-remote", new AkkaRemoteProject(_), akka_typed_actor) lazy val akka_amqp = project("akka-amqp", "akka-amqp", new AkkaAMQPProject(_), akka_remote) @@ -295,7 +297,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { lazy val akka_camel = project("akka-camel", "akka-camel", new AkkaCamelProject(_), akka_remote) lazy val akka_persistence = project("akka-persistence", "akka-persistence", new AkkaPersistenceParentProject(_)) lazy val akka_spring = project("akka-spring", "akka-spring", new AkkaSpringProject(_), akka_remote, akka_camel) - lazy val akka_jta = project("akka-jta", "akka-jta", new AkkaJTAProject(_), akka_remote) + lazy val akka_jta = project("akka-jta", "akka-jta", new AkkaJTAProject(_), akka_stm, akka_remote) lazy val akka_kernel = project("akka-kernel", "akka-kernel", new AkkaKernelProject(_), akka_remote, akka_jta, akka_http, akka_spring, akka_camel, akka_persistence, akka_amqp) lazy val akka_osgi = project("akka-osgi", "akka-osgi", new AkkaOSGiParentProject(_)) @@ -326,6 +328,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { " config/" + " scala-library.jar" + " dist/akka-actor_%s-%s.jar".format(buildScalaVersion, version) + + " dist/akka-stm_%s-%s.jar".format(buildScalaVersion, version) + " dist/akka-typed-actor_%s-%s.jar".format(buildScalaVersion, version) + " dist/akka-remote_%s-%s.jar".format(buildScalaVersion, version) + " dist/akka-http_%s-%s.jar".format(buildScalaVersion, version) + @@ -411,15 +414,27 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { val uuid = Dependencies.uuid val configgy = Dependencies.configgy val hawtdispatch = Dependencies.hawtdispatch - val multiverse = Dependencies.multiverse val jsr166x = Dependencies.jsr166x val slf4j = Dependencies.slf4j val logback = Dependencies.logback val logback_core = Dependencies.logback_core // testing - val junit = Dependencies.junit - val scalatest = Dependencies.scalatest + val junit = Dependencies.junit + val scalatest = Dependencies.scalatest + val multiverse_test = Dependencies.multiverse_test // StandardLatch + } + + // ------------------------------------------------------------------------------------------------------------------- + // akka-stm subproject + // ------------------------------------------------------------------------------------------------------------------- + + class AkkaStmProject(info: ProjectInfo) extends AkkaDefaultProject(info, distPath) { + val multiverse = Dependencies.multiverse + + // testing + val junit = Dependencies.junit + val scalatest = Dependencies.scalatest } // ------------------------------------------------------------------------------------------------------------------- @@ -526,7 +541,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { class AkkaPersistenceParentProject(info: ProjectInfo) extends ParentProject(info) { lazy val akka_persistence_common = project("akka-persistence-common", "akka-persistence-common", - new AkkaPersistenceCommonProject(_), akka_remote) + new AkkaPersistenceCommonProject(_), akka_remote, akka_stm) lazy val akka_persistence_redis = project("akka-persistence-redis", "akka-persistence-redis", new AkkaRedisProject(_), akka_persistence_common) lazy val akka_persistence_mongo = project("akka-persistence-mongo", "akka-persistence-mongo", @@ -625,6 +640,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { override def testOptions = createTestFilter( _.endsWith("Test") ) } + // ------------------------------------------------------------------------------------------------------------------- // akka-persistence-voldemort subproject // ------------------------------------------------------------------------------------------------------------------- @@ -645,7 +661,8 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { override def testOptions = createTestFilter({ s:String=> s.endsWith("Suite") || s.endsWith("Test")}) } -// akka-persistence-riak subproject + // ------------------------------------------------------------------------------------------------------------------- + // akka-persistence-riak subproject // ------------------------------------------------------------------------------------------------------------------- class AkkaRiakProject(info: ProjectInfo) extends AkkaDefaultProject(info, distPath) { @@ -658,6 +675,10 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { override def testOptions = createTestFilter(_.endsWith("Test")) } + // ------------------------------------------------------------------------------------------------------------------- + // akka-persistence-couchdb subproject + // ------------------------------------------------------------------------------------------------------------------- + class AkkaCouchDBProject(info: ProjectInfo) extends AkkaDefaultProject(info, distPath) { val couch = Dependencies.commonsHttpClient val spec = Dependencies.specs @@ -870,7 +891,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) { class AkkaSamplesParentProject(info: ProjectInfo) extends ParentProject(info) { lazy val akka_sample_ants = project("akka-sample-ants", "akka-sample-ants", - new AkkaSampleAntsProject(_), akka_remote) + new AkkaSampleAntsProject(_), akka_stm) lazy val akka_sample_chat = project("akka-sample-chat", "akka-sample-chat", new AkkaSampleChatProject(_), akka_kernel) lazy val akka_sample_pubsub = project("akka-sample-pubsub", "akka-sample-pubsub",