raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Some Java friendliness for STM

Browse source
This commit is contained in:
Peter Vlugter 2010-08-16 11:38:39 +12:00
parent 8e48ace976
commit bea8f16264
23 changed files with 611 additions and 122 deletions
Download patch file Download diff file Expand all files Collapse all files

95
akka-core/src/main/scala/stm/TransactionManagement.scala
View file

@ -4,16 +4,11 @@
package se.scalablesolutions.akka.stm
import se.scalablesolutions.akka.util.Logging
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.TimeUnit
import org.multiverse.api.{StmUtils => MultiverseStmUtils}
import org.multiverse.api.ThreadLocalTransaction._
import org.multiverse.api.{Transaction => MultiverseTransaction}
import org.multiverse.commitbarriers.CountDownCommitBarrier
import org.multiverse.templates.{TransactionalCallable, OrElseTemplate}
import org.multiverse.templates.OrElseTemplate
class TransactionSetAbortedException(msg: String) extends RuntimeException(msg)
@ -93,84 +88,7 @@ trait TransactionManagement {
}
}
/**
* Local transaction management, local in the context of threads.
* Use this if you do <b>not</b> need to have one transaction span
* multiple threads (or Actors).
* <p/>
* Example of atomic transaction management using the atomic block.
* <p/>
* <pre>
* import se.scalablesolutions.akka.stm.local._
*
* atomic {
* // do something within a transaction
* }
* </pre>
*/
class LocalStm extends TransactionManagement with Logging {
val DefaultLocalTransactionConfig = TransactionConfig()
val DefaultLocalTransactionFactory = TransactionFactory(DefaultLocalTransactionConfig, "DefaultLocalTransaction")
def atomic[T](body: => T)(implicit factory: TransactionFactory = DefaultLocalTransactionFactory): T = atomic(factory)(body)
def atomic[T](factory: TransactionFactory)(body: => T): T = {
factory.boilerplate.execute(new TransactionalCallable[T]() {
def call(mtx: MultiverseTransaction): T = {
factory.addHooks
val result = body
log.trace("Committing local transaction [" + mtx + "]")
result
}
})
}
}
/**
* Global transaction management, global in the context of multiple threads.
* Use this if you need to have one transaction span multiple threads (or Actors).
* <p/>
* Example of atomic transaction management using the atomic block:
* <p/>
* <pre>
* import se.scalablesolutions.akka.stm.global._
*
* atomic {
* // do something within a transaction
* }
* </pre>
*/
class GlobalStm extends TransactionManagement with Logging {
val DefaultGlobalTransactionConfig = TransactionConfig()
val DefaultGlobalTransactionFactory = TransactionFactory(DefaultGlobalTransactionConfig, "DefaultGlobalTransaction")
def atomic[T](body: => T)(implicit factory: TransactionFactory = DefaultGlobalTransactionFactory): T = atomic(factory)(body)
def atomic[T](factory: TransactionFactory)(body: => T): T = {
factory.boilerplate.execute(new TransactionalCallable[T]() {
def call(mtx: MultiverseTransaction): T = {
if (!isTransactionSetInScope) createNewTransactionSet
factory.addHooks
val result = body
val txSet = getTransactionSetInScope
log.trace("Committing global transaction [" + mtx + "]\n\tand joining transaction set [" + txSet + "]")
try {
txSet.tryJoinCommit(
mtx,
TransactionConfig.DefaultTimeout.length,
TransactionConfig.DefaultTimeout.unit)
// Need to catch IllegalStateException until we have fix in Multiverse, since it throws it by mistake
} catch { case e: IllegalStateException => {} }
result
}
})
}
}
trait StmUtil {
/**
* Schedule a deferred task on the thread local transaction (use within an atomic).
* This is executed when the transaction commits.
@ -209,3 +127,14 @@ trait StmUtil {
}.execute()
}
}
trait StmCommon {
type TransactionConfig = se.scalablesolutions.akka.stm.TransactionConfig
val TransactionConfig = se.scalablesolutions.akka.stm.TransactionConfig
type TransactionFactory = se.scalablesolutions.akka.stm.TransactionFactory
val TransactionFactory = se.scalablesolutions.akka.stm.TransactionFactory
type Ref[T] = se.scalablesolutions.akka.stm.Ref[T]
val Ref = se.scalablesolutions.akka.stm.Ref
}