f with each listener as argument.
*/
diff --git a/akka-actor/src/main/scala/util/LockUtil.scala b/akka-actor/src/main/scala/util/LockUtil.scala
index 885e11def7..3d1261e468 100644
--- a/akka-actor/src/main/scala/util/LockUtil.scala
+++ b/akka-actor/src/main/scala/util/LockUtil.scala
@@ -5,6 +5,7 @@
package se.scalablesolutions.akka.util
import java.util.concurrent.locks.{ReentrantReadWriteLock, ReentrantLock}
+import java.util.concurrent.atomic.AtomicBoolean
/**
* @author Jonas Bonér
@@ -58,3 +59,56 @@ class ReadWriteGuard {
}
}
+/**
+ * A very simple lock that uses CCAS (Compare Compare-And-Swap)
+ * Does not keep track of the owner and isn't Reentrant, so don't nest and try to stick to the if*-methods
+ */
+class SimpleLock {
+ val acquired = new AtomicBoolean(false)
+
+ def ifPossible(perform: () => Unit): Boolean = {
+ if (tryLock()) {
+ try {
+ perform
+ } finally {
+ unlock()
+ }
+ true
+ } else false
+ }
+
+ def ifPossibleYield[T](perform: () => T): Option[T] = {
+ if (tryLock()) {
+ try {
+ Some(perform())
+ } finally {
+ unlock()
+ }
+ } else None
+ }
+
+ def ifPossibleApply[T,R](value: T)(function: (T) => R): Option[R] = {
+ if (tryLock()) {
+ try {
+ Some(function(value))
+ } finally {
+ unlock()
+ }
+ } else None
+ }
+
+ def tryLock() = {
+ if (acquired.get) false
+ else acquired.compareAndSet(false,true)
+ }
+
+ def tryUnlock() = {
+ acquired.compareAndSet(true,false)
+ }
+
+ def locked = acquired.get
+
+ def unlock() {
+ acquired.set(false)
+ }
+}
\ No newline at end of file
diff --git a/akka-actor/src/main/scala/util/Logging.scala b/akka-actor/src/main/scala/util/Logging.scala
index f13c229653..b6ddaaa16a 100644
--- a/akka-actor/src/main/scala/util/Logging.scala
+++ b/akka-actor/src/main/scala/util/Logging.scala
@@ -99,18 +99,26 @@ class Logger(val logger: SLFLogger) {
warning(t,message(fmt,arg,argN:_*))
}
+ def warn(t: Throwable, fmt: => String, arg: Any, argN: Any*) = warning(t, fmt, arg, argN)
+
def warning(t: Throwable, msg: => String) {
if (warning_?) logger.warn(msg,t)
}
+ def warn(t: Throwable, msg: => String) = warning(t, msg)
+
def warning(fmt: => String, arg: Any, argN: Any*) {
warning(message(fmt,arg,argN:_*))
}
+ def warn(fmt: => String, arg: Any, argN: Any*) = warning(fmt, arg, argN)
+
def warning(msg: => String) {
if (warning_?) logger warn msg
}
+ def warn(msg: => String) = warning(msg)
+
//Error
def error(t: Throwable, fmt: => String, arg: Any, argN: Any*) {
error(t,message(fmt,arg,argN:_*))
diff --git a/akka-actor/src/main/scala/util/ReflectiveAccess.scala b/akka-actor/src/main/scala/util/ReflectiveAccess.scala
index 4582304188..abccd5d9b0 100644
--- a/akka-actor/src/main/scala/util/ReflectiveAccess.scala
+++ b/akka-actor/src/main/scala/util/ReflectiveAccess.scala
@@ -62,14 +62,8 @@ object ReflectiveAccess {
def ensureRemotingEnabled = if (!isRemotingEnabled) throw new ModuleNotAvailableException(
"Can't load the remoting module, make sure that akka-remote.jar is on the classpath")
- val remoteClientObjectInstance: Option[RemoteClientObject] = {
- try {
- val clazz = loader.loadClass("se.scalablesolutions.akka.remote.RemoteClient$")
- val ctor = clazz.getDeclaredConstructor(Array[Class[_]](): _*)
- ctor.setAccessible(true)
- Some(ctor.newInstance(Array[AnyRef](): _*).asInstanceOf[RemoteClientObject])
- } catch { case e: Exception => None }
- }
+ val remoteClientObjectInstance: Option[RemoteClientObject] =
+ getObject("se.scalablesolutions.akka.remote.RemoteClient$")
def register(address: InetSocketAddress, uuid: String) = {
ensureRemotingEnabled
@@ -126,23 +120,11 @@ object ReflectiveAccess {
def unregister(actorRef: ActorRef): Unit
}
- val remoteServerObjectInstance: Option[RemoteServerObject] = {
- try {
- val clazz = loader.loadClass("se.scalablesolutions.akka.remote.RemoteServer$")
- val ctor = clazz.getDeclaredConstructor(Array[Class[_]](): _*)
- ctor.setAccessible(true)
- Some(ctor.newInstance(Array[AnyRef](): _*).asInstanceOf[RemoteServerObject])
- } catch { case e: Exception => None }
- }
+ val remoteServerObjectInstance: Option[RemoteServerObject] =
+ getObject("se.scalablesolutions.akka.remote.RemoteServer$")
- val remoteNodeObjectInstance: Option[RemoteNodeObject] = {
- try {
- val clazz = loader.loadClass("se.scalablesolutions.akka.remote.RemoteNode$")
- val ctor = clazz.getDeclaredConstructor(Array[Class[_]](): _*)
- ctor.setAccessible(true)
- Some(ctor.newInstance(Array[AnyRef](): _*).asInstanceOf[RemoteNodeObject])
- } catch { case e: Exception => None }
- }
+ val remoteNodeObjectInstance: Option[RemoteNodeObject] =
+ getObject("se.scalablesolutions.akka.remote.RemoteNode$")
def registerActor(address: InetSocketAddress, uuid: String, actorRef: ActorRef) = {
ensureRemotingEnabled
@@ -177,14 +159,8 @@ object ReflectiveAccess {
def ensureTypedActorEnabled = if (!isTypedActorEnabled) throw new ModuleNotAvailableException(
"Can't load the typed actor module, make sure that akka-typed-actor.jar is on the classpath")
- val typedActorObjectInstance: Option[TypedActorObject] = {
- try {
- val clazz = loader.loadClass("se.scalablesolutions.akka.actor.TypedActor$")
- val ctor = clazz.getDeclaredConstructor(Array[Class[_]](): _*)
- ctor.setAccessible(true)
- Some(ctor.newInstance(Array[AnyRef](): _*).asInstanceOf[TypedActorObject])
- } catch { case e: Exception => None }
- }
+ val typedActorObjectInstance: Option[TypedActorObject] =
+ getObject("se.scalablesolutions.akka.actor.TypedActor$")
def resolveFutureIfMessageIsJoinPoint(message: Any, future: Future[_]): Boolean = {
ensureTypedActorEnabled
@@ -212,18 +188,46 @@ object ReflectiveAccess {
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] = {
- try {
- val clazz = loader.loadClass("se.scalablesolutions.akka.actor.TransactionContainer$")
- val ctor = clazz.getDeclaredConstructor(Array[Class[_]](): _*)
- ctor.setAccessible(true)
- Some(ctor.newInstance(Array[AnyRef](): _*).asInstanceOf[TransactionContainerObject])
- } catch { case e: Exception => None }
- }
+ val transactionContainerObjectInstance: Option[TransactionContainerObject] =
+ getObject("se.scalablesolutions.akka.actor.TransactionContainer$")
def createTransactionContainer: TransactionContainer = {
ensureJtaEnabled
transactionContainerObjectInstance.get.apply.asInstanceOf[TransactionContainer]
}
}
+
+ val noParams = Array[Class[_]]()
+ val noArgs = Array[AnyRef]()
+
+ def createInstance[T](clazz: Class[_],
+ params: Array[Class[_]],
+ args: Array[AnyRef]): Option[T] = try {
+ val ctor = clazz.getDeclaredConstructor(params: _*)
+ ctor.setAccessible(true)
+ Some(ctor.newInstance(args: _*).asInstanceOf[T])
+ } catch {
+ case e: Exception => None
+ }
+
+ def createInstance[T](fqn: String,
+ params: Array[Class[_]],
+ args: Array[AnyRef],
+ classloader: ClassLoader = loader): Option[T] = try {
+ val clazz = classloader.loadClass(fqn)
+ val ctor = clazz.getDeclaredConstructor(params: _*)
+ ctor.setAccessible(true)
+ Some(ctor.newInstance(args: _*).asInstanceOf[T])
+ } catch {
+ case e: Exception => None
+ }
+
+ def getObject[T](fqn: String, classloader: ClassLoader = loader): Option[T] = try {//Obtains a reference to $MODULE$
+ val clazz = classloader.loadClass(fqn)
+ val instance = clazz.getDeclaredField("MODULE$")
+ instance.setAccessible(true)
+ Option(instance.get(null).asInstanceOf[T])
+ } catch {
+ case e: Exception => None
+ }
}
diff --git a/akka-actor/src/test/resources/logback-test.xml b/akka-actor/src/test/resources/logback-test.xml
deleted file mode 100644
index 78eae40ec4..0000000000
--- a/akka-actor/src/test/resources/logback-test.xml
+++ /dev/null
@@ -1,21 +0,0 @@
-
-
-
-
-
-
-
-
-
-Actor.shutdown for freeing resources needed
+ * Default implementation of Actor.postStop for freeing resources needed
* to actually send messages to endpointUri.
*/
- override def shutdown {
+ override def postStop {
processor.stop
}
diff --git a/akka-camel/src/main/scala/component/ActorComponent.scala b/akka-camel/src/main/scala/component/ActorComponent.scala
index a5d56dd9dc..6c1c5902fa 100644
--- a/akka-camel/src/main/scala/component/ActorComponent.scala
+++ b/akka-camel/src/main/scala/component/ActorComponent.scala
@@ -247,8 +247,8 @@ private[akka] class AsyncCallbackAdapter(exchange: Exchange, callback: AsyncCall
protected[akka] def postMessageToMailboxAndCreateFutureResultWithTimeout[T](message: Any, timeout: Long, senderOption: Option[ActorRef], senderFuture: Option[CompletableFuture[T]]) = unsupported
protected[akka] def mailbox: AnyRef = unsupported
protected[akka] def mailbox_=(msg: AnyRef):AnyRef = unsupported
- protected[akka] def restart(reason: Throwable, maxNrOfRetries: Int, withinTimeRange: Int): Unit = unsupported
- protected[akka] def restartLinkedActors(reason: Throwable, maxNrOfRetries: Int, withinTimeRange: Int): Unit = unsupported
+ protected[akka] def restart(reason: Throwable, maxNrOfRetries: Option[Int], withinTimeRange: Option[Int]): Unit = unsupported
+ protected[akka] def restartLinkedActors(reason: Throwable, maxNrOfRetries: Option[Int], withinTimeRange: Option[Int]): Unit = unsupported
protected[akka] def handleTrapExit(dead: ActorRef, reason: Throwable): Unit = unsupported
protected[akka] def linkedActors: JavaMap[String, ActorRef] = unsupported
protected[akka] def linkedActorsAsList: List[ActorRef] = unsupported
diff --git a/akka-core/src/test/scala/actor/supervisor/SupervisorMiscSpec.scala b/akka-core/src/test/scala/actor/supervisor/SupervisorMiscSpec.scala
index 3d048684cd..26fdb6e1ef 100644
--- a/akka-core/src/test/scala/actor/supervisor/SupervisorMiscSpec.scala
+++ b/akka-core/src/test/scala/actor/supervisor/SupervisorMiscSpec.scala
@@ -26,7 +26,7 @@ class SupervisorMiscSpec extends WordSpec with MustMatchers {
}).start
val actor2 = Actor.actorOf(new Actor {
- self.dispatcher = Dispatchers.newReactorBasedSingleThreadEventDrivenDispatcher("test")
+ self.dispatcher = Dispatchers.newThreadBasedDispatcher(self)
override def postRestart(cause: Throwable) {countDownLatch.countDown}
protected def receive = {
diff --git a/akka-http/src/main/scala/AkkaBroadcaster.scala b/akka-http/src/main/scala/AkkaBroadcaster.scala
index 8f724ff445..ca5abc6f1d 100644
--- a/akka-http/src/main/scala/AkkaBroadcaster.scala
+++ b/akka-http/src/main/scala/AkkaBroadcaster.scala
@@ -9,10 +9,13 @@ import se.scalablesolutions.akka.actor.Actor._
import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.dispatch.Dispatchers
-class AkkaBroadcaster extends org.atmosphere.jersey.JerseyBroadcaster {
- name = classOf[AkkaBroadcaster].getName
-
+object AkkaBroadcaster {
val broadcasterDispatcher = Dispatchers.fromConfig("akka.rest.comet-dispatcher")
+}
+
+class AkkaBroadcaster extends org.atmosphere.jersey.JerseyBroadcaster {
+ import AkkaBroadcaster._
+ name = classOf[AkkaBroadcaster].getName
//FIXME should be supervised
val caster = actorOf(new Actor {
diff --git a/akka-http/src/main/scala/AkkaClusterBroadcastFilter.scala b/akka-http/src/main/scala/AkkaClusterBroadcastFilter.scala
index 775c8b554d..7ea963872f 100644
--- a/akka-http/src/main/scala/AkkaClusterBroadcastFilter.scala
+++ b/akka-http/src/main/scala/AkkaClusterBroadcastFilter.scala
@@ -24,6 +24,11 @@ class AkkaClusterBroadcastFilter extends Actor with ClusterBroadcastFilter {
@BeanProperty var clusterName = ""
@BeanProperty var broadcaster : Broadcaster = null
+ def init() {
+ //Since this class is instantiated by Atmosphere, we need to make sure it's started
+ self.start
+ }
+
/**
* Stops the actor
*/
@@ -48,7 +53,4 @@ class AkkaClusterBroadcastFilter extends Actor with ClusterBroadcastFilter {
case b @ ClusterCometBroadcast(c, _) if (c == clusterName) && (broadcaster ne null) => broadcaster broadcast b
case _ =>
}
-
- //Since this class is instantiated by Atmosphere, we need to make sure it's started
- self.start
}
diff --git a/akka-http/src/main/scala/Initializer.scala b/akka-http/src/main/scala/Initializer.scala
index da95a39b77..c1cd8bfc87 100644
--- a/akka-http/src/main/scala/Initializer.scala
+++ b/akka-http/src/main/scala/Initializer.scala
@@ -13,7 +13,7 @@ import se.scalablesolutions.akka.util.{Logging, Bootable}
import javax.servlet.{ServletContextListener, ServletContextEvent}
/**
- * This class can be added to web.xml mappings as a listener to start and shutdown Akka.
+ * This class can be added to web.xml mappings as a listener to start and postStop Akka.
*
*- * Map( - * "debasish.address" -> "kolkata, India", - * "debasish.company" -> "anshinsoft", - * "debasish.programming_language" -> "scala", - * )- * will be stored as the following key-value pair in Redis: - * - * - * base64(T1):base64("debasish.address") -> "kolkata, India" - * base64(T1):base64("debasish.company") -> "anshinsoft" - * base64(T1):base64("debasish.programming_language") -> "scala" - * + * Maps are stored as key/value pairs in redis. */ def insertMapStorageEntryFor(name: String, key: Array[Byte], value: Array[Byte]): Unit = withErrorHandling { insertMapStorageEntriesFor(name, List((key, value))) @@ -134,12 +96,12 @@ private [akka] object RedisStorageBackend extends *
- * curl http://localhost:8080/liftcount - *- * Or browse to the URL from a web browser. - */ - -object SimpleRestService extends RestHelper { - serve { - case Get("liftcount" :: _, req) => - //Fetch the first actor of type SimpleServiceActor - //Send it the "Tick" message and expect a Node back - val result = for( a <- ActorRegistry.actorFor[SimpleServiceActor]; - r <- (a !! "Tick").as[Node] ) yield r - - //Return either the resulting NodeSeq or a default one - (result getOrElse
- * curl http://localhost:8080/persistentliftcount - *- * Or browse to the URL from a web browser. - */ - object PersistentRestService extends RestHelper { - serve { - case Get("persistentliftcount" :: _, req) => - //Fetch the first actor of type SimpleServiceActor - //Send it the "Tick" message and expect a Node back - val result = for( a <- ActorRegistry.actorFor[PersistentServiceActor]; - r <- (a !! "Tick").as[Node] ) yield r - - //Return either the resulting NodeSeq or a default one - (result getOrElse
This page is served by Lift, and Lift alone. In order to demonstrate how AkkaServlet and
- Lift can work in harmony we have supplied a sample JAX-RS service that is secured using
- Lift's HTTP Basic Authentication.
To access the Akka service, visit this url and enter the - following access credentials:
- -user: someuser
- password: 1234