diff --git a/akka-actor/src/main/scala/actor/Actor.scala b/akka-actor/src/main/scala/actor/Actor.scala index 943854fc61..0fe1e9766b 100644 --- a/akka-actor/src/main/scala/actor/Actor.scala +++ b/akka-actor/src/main/scala/actor/Actor.scala @@ -82,6 +82,41 @@ class ActorTimeoutException private[akka](message: String) extends AkkaException * @author Jonas Bonér */ object Actor extends Logging { + + /** + * Add shutdown cleanups + */ + private[akka] lazy val shutdownHook = { + val hook = new Runnable { + override def run { + // Shutdown HawtDispatch GlobalQueue + log.info("Shutting down Hawt Dispatch global queue") + org.fusesource.hawtdispatch.ScalaDispatch.globalQueue.asInstanceOf[org.fusesource.hawtdispatch.internal.GlobalDispatchQueue].shutdown + + // Clear Thread.subclassAudits + log.info("Clearing subclass audits") + val tf = classOf[java.lang.Thread].getDeclaredField("subclassAudits") + tf.setAccessible(true) + val subclassAudits = tf.get(null).asInstanceOf[java.util.Map[_,_]] + subclassAudits.synchronized {subclassAudits.clear} + + // Clear and reset j.u.l.Level.known (due to Configgy) + log.info("Removing Configgy-installed log levels") + import java.util.logging.Level + val lf = classOf[Level].getDeclaredField("known") + lf.setAccessible(true) + val known = lf.get(null).asInstanceOf[java.util.ArrayList[Level]] + known.synchronized { + known.clear + List(Level.OFF,Level.SEVERE,Level.WARNING,Level.INFO,Level.CONFIG, + Level.FINE,Level.FINER,Level.FINEST,Level.ALL) foreach known.add + } + } + } + Runtime.getRuntime.addShutdownHook(new Thread(hook)) + hook + } + val TIMEOUT = Duration(config.getInt("akka.actor.timeout", 5), TIME_UNIT).toMillis val SERIALIZE_MESSAGES = config.getBool("akka.actor.serialize-messages", false) @@ -110,7 +145,6 @@ object Actor extends Logging { def actorOf[T <: Actor : Manifest]: ActorRef = actorOf(manifest[T].erasure.asInstanceOf[Class[_ <: Actor]]) /** -<<<<<<< HEAD:akka-actor/src/main/scala/actor/Actor.scala * Creates an ActorRef out of the Actor with type T. *
* import Actor._
diff --git a/akka-camel/src/main/scala/CamelService.scala b/akka-camel/src/main/scala/CamelService.scala
index cf2183da0f..b546636610 100644
--- a/akka-camel/src/main/scala/CamelService.scala
+++ b/akka-camel/src/main/scala/CamelService.scala
@@ -4,13 +4,14 @@
package akka.camel
import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
import org.apache.camel.CamelContext
import akka.actor.Actor._
import akka.actor.{AspectInitRegistry, ActorRegistry}
import akka.config.Config._
-import akka.japi.{Option => JOption}
+import akka.japi.{SideEffect, Option => JOption}
import akka.util.{Logging, Bootable}
/**
@@ -92,12 +93,76 @@ trait CamelService extends Bootable with Logging {
CamelContextManager.stop
}
+ /**
+ * Waits for an expected number (count) of endpoints to be activated
+ * during execution of f. The wait-timeout is by default 10 seconds.
+ * Other timeout values can be set via the timeout and timeUnit
+ * parameters.
+ */
+ def awaitEndpointActivation(count: Int, timeout: Long = 10, timeUnit: TimeUnit = TimeUnit.SECONDS)(f: => Unit): Boolean = {
+ val activation = expectEndpointActivationCount(count)
+ f; activation.await(timeout, timeUnit)
+ }
+
+ /**
+ * Waits for an expected number (count) of endpoints to be de-activated
+ * during execution of f. The wait-timeout is by default 10 seconds.
+ * Other timeout values can be set via the timeout and timeUnit
+ * parameters.
+ */
+ def awaitEndpointDeactivation(count: Int, timeout: Long = 10, timeUnit: TimeUnit = TimeUnit.SECONDS)(f: => Unit): Boolean = {
+ val activation = expectEndpointDeactivationCount(count)
+ f; activation.await(timeout, timeUnit)
+ }
+
+ /**
+ * Waits for an expected number (count) of endpoints to be activated
+ * during execution of p. The wait timeout is 10 seconds.
+ *
+ * Java API
+ */
+ def awaitEndpointActivation(count: Int, p: SideEffect): Boolean = {
+ awaitEndpointActivation(count, 10, TimeUnit.SECONDS, p)
+ }
+
+ /**
+ * Waits for an expected number (count) of endpoints to be activated
+ * during execution of p. Timeout values can be set via the
+ * timeout and timeUnit parameters.
+ *
+ * Java API
+ */
+ def awaitEndpointActivation(count: Int, timeout: Long, timeUnit: TimeUnit, p: SideEffect): Boolean = {
+ awaitEndpointActivation(count, timeout, timeUnit) { p.apply }
+ }
+
+ /**
+ * Waits for an expected number (count) of endpoints to be de-activated
+ * during execution of p. The wait timeout is 10 seconds.
+ *
+ * Java API
+ */
+ def awaitEndpointDeactivation(count: Int, p: SideEffect): Boolean = {
+ awaitEndpointDeactivation(count, 10, TimeUnit.SECONDS, p)
+ }
+
+ /**
+ * Waits for an expected number (count) of endpoints to be de-activated
+ * during execution of p. Timeout values can be set via the
+ * timeout and timeUnit parameters.
+ *
+ * Java API
+ */
+ def awaitEndpointDeactivation(count: Int, timeout: Long, timeUnit: TimeUnit, p: SideEffect): Boolean = {
+ awaitEndpointDeactivation(count, timeout, timeUnit) { p.apply }
+ }
+
/**
* Sets an expectation on the number of upcoming endpoint activations and returns
- * a CountDownLatch that can be used to wait for the activations to occur. Endpoint
+ * a CountDownLatch that can be used to wait for the activations to occur. Endpoint
* activations that occurred in the past are not considered.
*/
- def expectEndpointActivationCount(count: Int): CountDownLatch =
+ private def expectEndpointActivationCount(count: Int): CountDownLatch =
(consumerPublisher !! SetExpectedRegistrationCount(count)).as[CountDownLatch].get
/**
@@ -105,7 +170,7 @@ trait CamelService extends Bootable with Logging {
* a CountDownLatch that can be used to wait for the de-activations to occur. Endpoint
* de-activations that occurred in the past are not considered.
*/
- def expectEndpointDeactivationCount(count: Int): CountDownLatch =
+ private def expectEndpointDeactivationCount(count: Int): CountDownLatch =
(consumerPublisher !! SetExpectedUnregistrationCount(count)).as[CountDownLatch].get
private[camel] def publishRequestorRegistered: Boolean = {
diff --git a/akka-camel/src/test/scala/ConsumerTest.scala b/akka-camel/src/test/scala/ConsumerTest.scala
index 181e59ed52..29a2697ff5 100644
--- a/akka-camel/src/test/scala/ConsumerTest.scala
+++ b/akka-camel/src/test/scala/ConsumerTest.scala
@@ -30,12 +30,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
// start consumer publisher, otherwise we cannot set message
// count expectations in the next step (needed for testing only).
service.consumerPublisher.start
- // set expectations on publish count
- val latch = service.expectEndpointActivationCount(1)
- // start the CamelService
- service.start
- // await publication of first test consumer
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(1) {
+ service.start
+ } must be (true)
}
override protected def afterAll = {
@@ -57,9 +54,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
}
"started" must {
"support an in-out message exchange via its endpoint" in {
- val latch = service.expectEndpointActivationCount(1)
- consumer.start
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(1) {
+ consumer.start
+ } must be (true)
mandatoryTemplate.requestBody("direct:publish-test-2", "msg2") must equal ("received msg2")
}
"have an associated endpoint in the CamelContext" in {
@@ -68,9 +65,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
}
"stopped" must {
"not support an in-out message exchange via its endpoint" in {
- val latch = service.expectEndpointDeactivationCount(1)
- consumer.stop
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointDeactivation(1) {
+ consumer.stop
+ } must be (true)
intercept[CamelExecutionException] {
mandatoryTemplate.requestBody("direct:publish-test-2", "msg2")
}
@@ -82,9 +79,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
var actor: SampleTypedConsumer = null
"started" must {
"support in-out message exchanges via its endpoints" in {
- val latch = service.expectEndpointActivationCount(3)
- actor = TypedActor.newInstance(classOf[SampleTypedConsumer], classOf[SampleTypedConsumerImpl])
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(3) {
+ actor = TypedActor.newInstance(classOf[SampleTypedConsumer], classOf[SampleTypedConsumerImpl])
+ } must be (true)
mandatoryTemplate.requestBodyAndHeader("direct:m2", "x", "test", "y") must equal ("m2: x y")
mandatoryTemplate.requestBodyAndHeader("direct:m3", "x", "test", "y") must equal ("m3: x y")
mandatoryTemplate.requestBodyAndHeader("direct:m4", "x", "test", "y") must equal ("m4: x y")
@@ -92,9 +89,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
}
"stopped" must {
"not support in-out message exchanges via its endpoints" in {
- val latch = service.expectEndpointDeactivationCount(3)
- TypedActor.stop(actor)
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointDeactivation(3) {
+ TypedActor.stop(actor)
+ } must be (true)
intercept[CamelExecutionException] {
mandatoryTemplate.requestBodyAndHeader("direct:m2", "x", "test", "y")
}
@@ -112,18 +109,18 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
var actor: TestTypedConsumer = null
"started" must {
"support in-out message exchanges via its endpoints" in {
- val latch = service.expectEndpointActivationCount(2)
- actor = TypedActor.newInstance(classOf[TestTypedConsumer], classOf[TestTypedConsumerImpl])
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(2) {
+ actor = TypedActor.newInstance(classOf[TestTypedConsumer], classOf[TestTypedConsumerImpl])
+ } must be (true)
mandatoryTemplate.requestBody("direct:publish-test-3", "x") must equal ("foo: x")
mandatoryTemplate.requestBody("direct:publish-test-4", "x") must equal ("bar: x")
}
}
"stopped" must {
"not support in-out message exchanges via its endpoints" in {
- val latch = service.expectEndpointDeactivationCount(2)
- TypedActor.stop(actor)
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointDeactivation(2) {
+ TypedActor.stop(actor)
+ } must be (true)
intercept[CamelExecutionException] {
mandatoryTemplate.requestBody("direct:publish-test-3", "x")
}
@@ -138,17 +135,17 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
val consumer = UntypedActor.actorOf(classOf[SampleUntypedConsumer])
"started" must {
"support an in-out message exchange via its endpoint" in {
- val latch = service.expectEndpointActivationCount(1)
- consumer.start
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(1) {
+ consumer.start
+ } must be (true)
mandatoryTemplate.requestBodyAndHeader("direct:test-untyped-consumer", "x", "test", "y") must equal ("x y")
}
}
"stopped" must {
"not support an in-out message exchange via its endpoint" in {
- val latch = service.expectEndpointDeactivationCount(1)
- consumer.stop
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointDeactivation(1) {
+ consumer.stop
+ } must be (true)
intercept[CamelExecutionException] {
mandatoryTemplate.sendBodyAndHeader("direct:test-untyped-consumer", "blah", "test", "blub")
}
@@ -159,9 +156,9 @@ class ConsumerTest extends WordSpec with BeforeAndAfterAll with MustMatchers {
"A non-responding, blocking consumer" when {
"receiving an in-out message exchange" must {
"lead to a TimeoutException" in {
- val latch = service.expectEndpointActivationCount(1)
- actorOf(new TestBlocker("direct:publish-test-5")).start
- latch.await(5000, TimeUnit.MILLISECONDS) must be (true)
+ service.awaitEndpointActivation(1) {
+ actorOf(new TestBlocker("direct:publish-test-5")).start
+ } must be (true)
try {
mandatoryTemplate.requestBody("direct:publish-test-5", "msg3")
diff --git a/akka-camel/src/test/scala/RemoteConsumerTest.scala b/akka-camel/src/test/scala/RemoteConsumerTest.scala
index 77a1d9e757..957080c2ec 100644
--- a/akka-camel/src/test/scala/RemoteConsumerTest.scala
+++ b/akka-camel/src/test/scala/RemoteConsumerTest.scala
@@ -45,9 +45,9 @@ class RemoteConsumerTest extends FeatureSpec with BeforeAndAfterAll with GivenWh
val consumer = actorOf[RemoteConsumer].start
when("remote consumer publication is triggered")
- var latch = mandatoryService.expectEndpointActivationCount(1)
- consumer !! "init"
- assert(latch.await(5000, TimeUnit.MILLISECONDS))
+ assert(mandatoryService.awaitEndpointActivation(1) {
+ consumer !! "init"
+ })
then("the published consumer is accessible via its endpoint URI")
val response = CamelContextManager.mandatoryTemplate.requestBody("direct:remote-consumer", "test")
@@ -61,10 +61,9 @@ class RemoteConsumerTest extends FeatureSpec with BeforeAndAfterAll with GivenWh
val consumer = TypedActor.newRemoteInstance(classOf[SampleRemoteTypedConsumer], classOf[SampleRemoteTypedConsumerImpl], host, port)
when("remote typed consumer publication is triggered")
- var latch = mandatoryService.expectEndpointActivationCount(1)
- consumer.foo("init")
- assert(latch.await(5000, TimeUnit.MILLISECONDS))
-
+ assert(mandatoryService.awaitEndpointActivation(1) {
+ consumer.foo("init")
+ })
then("the published method is accessible via its endpoint URI")
val response = CamelContextManager.mandatoryTemplate.requestBody("direct:remote-typed-consumer", "test")
assert(response === "remote typed actor: test")
@@ -77,10 +76,9 @@ class RemoteConsumerTest extends FeatureSpec with BeforeAndAfterAll with GivenWh
val consumer = UntypedActor.actorOf(classOf[SampleRemoteUntypedConsumer]).start
when("remote untyped consumer publication is triggered")
- var latch = mandatoryService.expectEndpointActivationCount(1)
- consumer.sendRequestReply(Message("init", Map("test" -> "init")))
- assert(latch.await(5000, TimeUnit.MILLISECONDS))
-
+ assert(mandatoryService.awaitEndpointActivation(1) {
+ consumer.sendRequestReply(Message("init", Map("test" -> "init")))
+ })
then("the published untyped consumer is accessible via its endpoint URI")
val response = CamelContextManager.mandatoryTemplate.requestBodyAndHeader("direct:remote-untyped-consumer", "a", "test", "b")
assert(response === "a b")
diff --git a/akka-http/src/main/scala/AkkaLoader.scala b/akka-http/src/main/scala/AkkaLoader.scala
index d8afac67bc..22ff40016c 100644
--- a/akka-http/src/main/scala/AkkaLoader.scala
+++ b/akka-http/src/main/scala/AkkaLoader.scala
@@ -6,6 +6,7 @@ package akka.servlet
import akka.config.Config
import akka.util.{Logging, Bootable}
+import akka.actor.Actor
/*
* This class is responsible for booting up a stack of bundles and then shutting them down
@@ -40,6 +41,7 @@ class AkkaLoader extends Logging {
log.info("Shutting down Akka...")
_bundles.foreach(_.onUnload)
_bundles = None
+ Actor.shutdownHook.run
log.info("Akka succesfully shut down")
}
}
diff --git a/akka-samples/akka-sample-camel/src/main/resources/context-jms.xml b/akka-samples/akka-sample-camel/src/main/resources/context-jms.xml
index b3d811d8de..12e4541be3 100644
--- a/akka-samples/akka-sample-camel/src/main/resources/context-jms.xml
+++ b/akka-samples/akka-sample-camel/src/main/resources/context-jms.xml
@@ -13,7 +13,11 @@ http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
-
+
+
+
+
+
diff --git a/akka-samples/akka-sample-camel/src/main/scala/Actors.scala b/akka-samples/akka-sample-camel/src/main/scala/Actors.scala
index 15010c4ed6..f19144bbbf 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/Actors.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/Actors.scala
@@ -41,7 +41,7 @@ class Consumer1 extends Actor with Consumer with Logging {
}
}
-class Consumer2 extends Actor {
+class Consumer2 extends Actor with Consumer {
def endpointUri = "jetty:http://0.0.0.0:8877/camel/default"
def receive = {
diff --git a/akka-samples/akka-sample-camel/src/main/scala/StandaloneApplication.scala b/akka-samples/akka-sample-camel/src/main/scala/StandaloneApplication.scala
index b083509216..13717e17d4 100644
--- a/akka-samples/akka-sample-camel/src/main/scala/StandaloneApplication.scala
+++ b/akka-samples/akka-sample-camel/src/main/scala/StandaloneApplication.scala
@@ -28,14 +28,10 @@ object StandaloneApplication extends Application {
// access 'externally' registered typed actors
assert("hello msg1" == mandatoryContext.createProducerTemplate.requestBody("direct:test", "msg1"))
- // set expectations on upcoming endpoint activation
- val activation = mandatoryService.expectEndpointActivationCount(1)
-
- // 'internally' register typed actor (requires CamelService)
- TypedActor.newInstance(classOf[TypedConsumer2], classOf[TypedConsumer2Impl])
-
- // internal registration is done in background. Wait a bit ...
- activation.await
+ mandatoryService.awaitEndpointActivation(1) {
+ // 'internally' register typed actor (requires CamelService)
+ TypedActor.newInstance(classOf[TypedConsumer2], classOf[TypedConsumer2Impl])
+ }
// access 'internally' (automatically) registered typed-actors
// (see @consume annotation value at TypedConsumer2.foo method)
@@ -85,17 +81,13 @@ object StandaloneJmsApplication extends Application {
startCamelService
- // Expect two consumer endpoints to be activated
- val completion = mandatoryService.expectEndpointActivationCount(2)
-
val jmsUri = "jms:topic:test"
- // Wire publisher and consumer using a JMS topic
- val jmsSubscriber1 = Actor.actorOf(new Subscriber("jms-subscriber-1", jmsUri)).start
- val jmsSubscriber2 = Actor.actorOf(new Subscriber("jms-subscriber-2", jmsUri)).start
val jmsPublisher = Actor.actorOf(new Publisher("jms-publisher", jmsUri)).start
- // wait for the consumer (subscriber) endpoint being activated
- completion.await
+ mandatoryService.awaitEndpointActivation(2) {
+ Actor.actorOf(new Subscriber("jms-subscriber-1", jmsUri)).start
+ Actor.actorOf(new Subscriber("jms-subscriber-2", jmsUri)).start
+ }
// Send 10 messages to via publisher actor
for(i <- 1 to 10) {
@@ -107,7 +99,9 @@ object StandaloneJmsApplication extends Application {
CamelContextManager.mandatoryTemplate.sendBody(jmsUri, "Camel rocks (%d)" format i)
}
- stopCamelService
+ // Wait a bit for subscribes to receive messages
+ Thread.sleep(1000)
+ stopCamelService
ActorRegistry.shutdownAll
}
diff --git a/akka-samples/akka-sample-camel/src/test/scala/HttpConcurrencyTestStress.scala b/akka-samples/akka-sample-camel/src/test/scala/HttpConcurrencyTestStress.scala
index bca3fed321..27e8ec1800 100644
--- a/akka-samples/akka-sample-camel/src/test/scala/HttpConcurrencyTestStress.scala
+++ b/akka-samples/akka-sample-camel/src/test/scala/HttpConcurrencyTestStress.scala
@@ -44,15 +44,15 @@ class HttpConcurrencyTestStress extends JUnitSuite {
object HttpConcurrencyTestStress {
@BeforeClass
- def beforeClass = {
+ def beforeClass: Unit = {
startCamelService
val workers = for (i <- 1 to 8) yield actorOf[HttpServerWorker].start
val balancer = loadBalancerActor(new CyclicIterator(workers.toList))
- val completion = service.get.expectEndpointActivationCount(1)
- val server = actorOf(new HttpServerActor(balancer)).start
- completion.await
+ service.get.awaitEndpointActivation(1) {
+ actorOf(new HttpServerActor(balancer)).start
+ }
}
@AfterClass
diff --git a/project/build/AkkaProject.scala b/project/build/AkkaProject.scala
index b579f391f5..d8a570a7a7 100644
--- a/project/build/AkkaProject.scala
+++ b/project/build/AkkaProject.scala
@@ -118,7 +118,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) {
// -------------------------------------------------------------------------------------------------------------------
lazy val ATMO_VERSION = "0.6.2"
- lazy val CAMEL_VERSION = "2.4.0"
+ lazy val CAMEL_VERSION = "2.5.0"
lazy val CASSANDRA_VERSION = "0.6.1"
lazy val DISPATCH_VERSION = "0.7.4"
lazy val HAWT_DISPATCH_VERSION = "1.0"
@@ -128,9 +128,9 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) {
lazy val SCALATEST_VERSION = "1.2"
lazy val LOGBACK_VERSION = "0.9.24"
lazy val SLF4J_VERSION = "1.6.0"
- lazy val SPRING_VERSION = "3.0.3.RELEASE"
+ lazy val SPRING_VERSION = "3.0.4.RELEASE"
lazy val ASPECTWERKZ_VERSION = "2.2.2"
- lazy val JETTY_VERSION = "7.1.4.v20100610"
+ lazy val JETTY_VERSION = "7.1.6.v20100715"
// -------------------------------------------------------------------------------------------------------------------
// Dependencies
@@ -820,7 +820,7 @@ class AkkaParentProject(info: ProjectInfo) extends DefaultProject(info) {
-
+