From 18260a3b7bcda1055572eef472cb654d62c25604 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Fri, 8 Jun 2012 13:56:53 +0200 Subject: [PATCH 1/9] #2203 - publish failed message deliveries to DeadLetters when bounded or durable mailbox enqueues fail --- .../actor/ActorWithBoundedStashSpec.scala | 51 ++++++------------- .../akka/dispatch/MailboxConfigSpec.scala | 10 ++-- .../main/scala/akka/dispatch/Mailbox.scala | 35 +++++-------- 3 files changed, 32 insertions(+), 64 deletions(-) diff --git a/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala index 9d411268aa..33283b18cf 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/ActorWithBoundedStashSpec.scala @@ -6,7 +6,7 @@ package akka.actor import akka.testkit._ import akka.testkit.DefaultTimeout import akka.testkit.TestEvent._ -import akka.dispatch.{ Await, MessageQueueAppendFailedException, BoundedDequeBasedMailbox } +import akka.dispatch.{ Await, BoundedDequeBasedMailbox } import akka.pattern.ask import akka.util.duration._ import akka.actor.ActorSystem.Settings @@ -17,16 +17,8 @@ object ActorWithBoundedStashSpec { class StashingActor(implicit sys: ActorSystem) extends Actor with Stash { def receive = { - case "hello" ⇒ - stash() - sender ! "OK" - case "world" ⇒ - try { - unstashAll() - } catch { - case e: MessageQueueAppendFailedException ⇒ - expectedException.open() - } + case "hello" ⇒ stash() + case "world" ⇒ unstashAll() } } @@ -36,18 +28,10 @@ object ActorWithBoundedStashSpec { def receive = { case "hello" ⇒ numStashed += 1 - try { - stash() - } catch { - case e: StashOverflowException ⇒ - if (numStashed == 21) stashOverflow.open() - } + try stash() catch { case e: StashOverflowException ⇒ if (numStashed == 21) sender ! "STASHOVERFLOW" } } } - @volatile var expectedException: TestLatch = null - @volatile var stashOverflow: TestLatch = null - val testConf: Config = ConfigFactory.parseString(""" my-dispatcher { mailbox-type = "akka.actor.ActorWithBoundedStashSpec$Bounded" @@ -56,47 +40,42 @@ object ActorWithBoundedStashSpec { """) // bounded deque-based mailbox with capacity 10 - class Bounded(settings: Settings, config: Config) extends BoundedDequeBasedMailbox(10, 5 seconds) - + class Bounded(settings: Settings, config: Config) extends BoundedDequeBasedMailbox(10, 1 seconds) } @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner]) -class ActorWithBoundedStashSpec extends AkkaSpec(ActorWithBoundedStashSpec.testConf) with DefaultTimeout with BeforeAndAfterEach { +class ActorWithBoundedStashSpec extends AkkaSpec(ActorWithBoundedStashSpec.testConf) with DefaultTimeout with BeforeAndAfterEach with ImplicitSender { import ActorWithBoundedStashSpec._ implicit val sys = system - override def atStartup { - system.eventStream.publish(Mute(EventFilter[Exception]("Crashing..."))) - } + override def atStartup { system.eventStream.publish(Mute(EventFilter[Exception]("Crashing..."))) } def myProps(creator: ⇒ Actor): Props = Props(creator).withDispatcher("my-dispatcher") "An Actor with Stash and BoundedDequeBasedMailbox" must { - "throw a MessageQueueAppendFailedException in case of a capacity violation" in { - ActorWithBoundedStashSpec.expectedException = new TestLatch + "end up in DeadLetters in case of a capacity violation" in { + system.eventStream.subscribe(testActor, classOf[DeadLetter]) + val stasher = system.actorOf(myProps(new StashingActor)) // fill up stash - val futures = for (_ ← 1 to 11) yield { stasher ? "hello" } - futures foreach { Await.ready(_, 10 seconds) } + (1 to 11) foreach { _ ⇒ stasher ! "hello" } // cause unstashAll with capacity violation stasher ! "world" - Await.ready(ActorWithBoundedStashSpec.expectedException, 10 seconds) + expectMsg(DeadLetter("hello", testActor, stasher)) + system.eventStream.unsubscribe(testActor, classOf[DeadLetter]) } - } "An Actor with bounded Stash" must { "throw a StashOverflowException in case of a stash capacity violation" in { - ActorWithBoundedStashSpec.stashOverflow = new TestLatch val stasher = system.actorOf(myProps(new StashingActorWithOverflow)) // fill up stash - for (_ ← 1 to 21) { stasher ! "hello" } - Await.ready(ActorWithBoundedStashSpec.stashOverflow, 10 seconds) + (1 to 21) foreach { _ ⇒ stasher ! "hello" } + expectMsg("STASHOVERFLOW") } - } } diff --git a/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala b/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala index 8759f1aad9..4f2d61de65 100644 --- a/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/dispatch/MailboxConfigSpec.scala @@ -6,9 +6,8 @@ import java.util.concurrent.ConcurrentLinkedQueue import akka.util._ import akka.util.duration._ import akka.testkit.AkkaSpec -import akka.actor.{ ActorRef, ActorContext, Props, LocalActorRef } import com.typesafe.config.Config -import akka.actor.ActorSystem +import akka.actor._ @org.junit.runner.RunWith(classOf[org.scalatest.junit.JUnitRunner]) abstract class MailboxSpec extends AkkaSpec with BeforeAndAfterAll with BeforeAndAfterEach { @@ -39,9 +38,10 @@ abstract class MailboxSpec extends AkkaSpec with BeforeAndAfterAll with BeforeAn q.numberOfMessages must be === config.capacity q.hasMessages must be === true - intercept[MessageQueueAppendFailedException] { - q.enqueue(null, exampleMessage) - } + system.eventStream.subscribe(testActor, classOf[DeadLetter]) + q.enqueue(testActor, exampleMessage) + expectMsg(DeadLetter(exampleMessage.message, system.deadLetters, testActor)) + system.eventStream.unsubscribe(testActor, classOf[DeadLetter]) q.dequeue must be === exampleMessage q.numberOfMessages must be(config.capacity - 1) diff --git a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala index b6af478ac7..25fc0250af 100644 --- a/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala +++ b/akka-actor/src/main/scala/akka/dispatch/Mailbox.scala @@ -6,18 +6,11 @@ package akka.dispatch import akka.AkkaException import java.util.{ Comparator, PriorityQueue, Queue, Deque } import akka.util._ -import akka.actor.{ ActorCell, ActorRef } import java.util.concurrent._ import annotation.tailrec import akka.event.Logging.Error -import akka.actor.ActorContext import com.typesafe.config.Config -import akka.actor.ActorSystem - -/** - * This exception normally is thrown when a bounded mailbox is over capacity - */ -class MessageQueueAppendFailedException(message: String, cause: Throwable = null) extends AkkaException(message, cause) +import akka.actor._ /** * INTERNAL API @@ -401,13 +394,11 @@ trait BoundedMessageQueueSemantics extends QueueBasedMessageQueue { def pushTimeOut: Duration override def queue: BlockingQueue[Envelope] - def enqueue(receiver: ActorRef, handle: Envelope) { + def enqueue(receiver: ActorRef, handle: Envelope): Unit = if (pushTimeOut.length > 0) { - queue.offer(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } + if (!queue.offer(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) } else queue put handle - } def dequeue(): Envelope = queue.poll() } @@ -439,18 +430,16 @@ trait BoundedDequeBasedMessageQueueSemantics extends DequeBasedMessageQueue { override def queue: BlockingDeque[Envelope] def enqueue(receiver: ActorRef, handle: Envelope): Unit = - if (pushTimeOut.length > 0) - queue.offer(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } - else queue put handle + if (pushTimeOut.length > 0) { + if (!queue.offer(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) + } else queue put handle def enqueueFirst(receiver: ActorRef, handle: Envelope): Unit = - if (pushTimeOut.length > 0) - queue.offerFirst(handle, pushTimeOut.length, pushTimeOut.unit) || { - throw new MessageQueueAppendFailedException("Couldn't enqueue message " + handle + " to " + receiver) - } - else queue putFirst handle + if (pushTimeOut.length > 0) { + if (!queue.offerFirst(handle, pushTimeOut.length, pushTimeOut.unit)) + receiver.asInstanceOf[InternalActorRef].provider.deadLetters ! DeadLetter(handle.message, handle.sender, receiver) + } else queue putFirst handle def dequeue(): Envelope = queue.poll() } From fb62311f49f4e0155d080fcbcf788e932bb75757 Mon Sep 17 00:00:00 2001 From: Patrik Nordwall Date: Fri, 8 Jun 2012 15:03:11 +0200 Subject: [PATCH 2/9] Rename NodeShutdownSpec to SingletonClusterSpec, see #2182 --- ...deShutdownSpec.scala => SingletonClusterSpec.scala} | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) rename akka-cluster/src/multi-jvm/scala/akka/cluster/{NodeShutdownSpec.scala => SingletonClusterSpec.scala} (78%) diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeShutdownSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/SingletonClusterSpec.scala similarity index 78% rename from akka-cluster/src/multi-jvm/scala/akka/cluster/NodeShutdownSpec.scala rename to akka-cluster/src/multi-jvm/scala/akka/cluster/SingletonClusterSpec.scala index 4dc90a5b89..68d20012f5 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeShutdownSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/SingletonClusterSpec.scala @@ -9,7 +9,7 @@ import akka.remote.testkit.MultiNodeSpec import akka.testkit._ import akka.util.duration._ -object NodeShutdownMultiJvmSpec extends MultiNodeConfig { +object SingletonClusterMultiJvmSpec extends MultiNodeConfig { val first = role("first") val second = role("second") @@ -24,11 +24,11 @@ object NodeShutdownMultiJvmSpec extends MultiNodeConfig { } -class NodeShutdownMultiJvmNode1 extends NodeShutdownSpec -class NodeShutdownMultiJvmNode2 extends NodeShutdownSpec +class SingletonClusterMultiJvmNode1 extends SingletonClusterSpec +class SingletonClusterMultiJvmNode2 extends SingletonClusterSpec -abstract class NodeShutdownSpec extends MultiNodeSpec(NodeShutdownMultiJvmSpec) with MultiNodeClusterSpec { - import NodeShutdownMultiJvmSpec._ +abstract class SingletonClusterSpec extends MultiNodeSpec(SingletonClusterMultiJvmSpec) with MultiNodeClusterSpec { + import SingletonClusterMultiJvmSpec._ "A cluster of 2 nodes" must { From a1dd4bc23560778fc48cb7440cb3c4f36753a83a Mon Sep 17 00:00:00 2001 From: Patrik Nordwall Date: Fri, 8 Jun 2012 15:28:03 +0200 Subject: [PATCH 3/9] Remove jenkins color codes in LogRoleReplace script --- .../scala/akka/remote/testkit/LogRoleReplace.scala | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/akka-remote-tests/src/test/scala/akka/remote/testkit/LogRoleReplace.scala b/akka-remote-tests/src/test/scala/akka/remote/testkit/LogRoleReplace.scala index 3b3527240e..1e5a53d82e 100644 --- a/akka-remote-tests/src/test/scala/akka/remote/testkit/LogRoleReplace.scala +++ b/akka-remote-tests/src/test/scala/akka/remote/testkit/LogRoleReplace.scala @@ -90,6 +90,7 @@ class LogRoleReplace { private val RoleStarted = """\[([\w\-]+)\].*Role \[([\w]+)\] started""".r private val RemoteServerStarted = """\[([\w\-]+)\].*RemoteServerStarted@akka://.*@([\w\-\.]+):([0-9]+)""".r + private val ColorCode = """\[[0-9]+m""" private var replacements: Map[String, String] = Map.empty private var jvmToAddress: Map[String, String] = Map.empty @@ -106,12 +107,16 @@ class LogRoleReplace { } def processLine(line: String): String = { - if (updateReplacements(line)) - replaceLine(line) + val cleanLine = removeColorCodes(line) + if (updateReplacements(cleanLine)) + replaceLine(cleanLine) else - line + cleanLine } + private def removeColorCodes(line: String): String = + line.replaceAll(ColorCode, "") + private def updateReplacements(line: String): Boolean = { if (line.startsWith("[info] * ")) { // reset when new test begins From 8479db7e75dea006658869addc02d22cbe644c52 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Fri, 8 Jun 2012 15:30:17 +0200 Subject: [PATCH 4/9] #2186 - Adding test to verify that parent is restarted if decider throws an exception --- .../scala/akka/actor/SupervisorSpec.scala | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/akka-actor-tests/src/test/scala/akka/actor/SupervisorSpec.scala b/akka-actor-tests/src/test/scala/akka/actor/SupervisorSpec.scala index 9e14a510e2..3db5b5b5dc 100644 --- a/akka-actor-tests/src/test/scala/akka/actor/SupervisorSpec.scala +++ b/akka-actor-tests/src/test/scala/akka/actor/SupervisorSpec.scala @@ -364,5 +364,39 @@ class SupervisorSpec extends AkkaSpec with BeforeAndAfterEach with ImplicitSende system.stop(supervisor) } + + "must not lose system messages when a NonFatal exception occurs when processing a system message" in { + val parent = system.actorOf(Props(new Actor { + override val supervisorStrategy = OneForOneStrategy()({ + case e: IllegalStateException if e.getMessage == "OHNOES" ⇒ throw e + case _ ⇒ SupervisorStrategy.Restart + }) + val child = context.watch(context.actorOf(Props(new Actor { + override def postRestart(reason: Throwable): Unit = testActor ! "child restarted" + def receive = { + case "die" ⇒ throw new IllegalStateException("OHNOES") + case "test" ⇒ sender ! "child green" + } + }), "child")) + + override def postRestart(reason: Throwable): Unit = testActor ! "parent restarted" + + def receive = { + case t @ Terminated(`child`) ⇒ testActor ! "child terminated" + case "die" ⇒ child ! "die" + case "test" ⇒ sender ! "green" + case "testchild" ⇒ child forward "test" + } + })) + + parent ! "die" + parent ! "testchild" + expectMsg("parent restarted") + expectMsg("child terminated") + parent ! "test" + expectMsg("green") + parent ! "testchild" + expectMsg("child green") + } } } From c9e259d56992a785de70361734c62a3b35b591d0 Mon Sep 17 00:00:00 2001 From: Patrik Nordwall Date: Fri, 8 Jun 2012 16:51:27 +0200 Subject: [PATCH 5/9] Turn on debug logging due to failures --- .../src/multi-jvm/scala/akka/cluster/JoinTwoClustersSpec.scala | 2 +- .../src/multi-jvm/scala/akka/cluster/SunnyWeatherSpec.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinTwoClustersSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinTwoClustersSpec.scala index e86602949f..2e27f4c3bd 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinTwoClustersSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/JoinTwoClustersSpec.scala @@ -17,7 +17,7 @@ object JoinTwoClustersMultiJvmSpec extends MultiNodeConfig { val c1 = role("c1") val c2 = role("c2") - commonConfig(debugConfig(on = false).withFallback(MultiNodeClusterSpec.clusterConfig)) + commonConfig(debugConfig(on = true).withFallback(MultiNodeClusterSpec.clusterConfig)) } class JoinTwoClustersMultiJvmNode1 extends JoinTwoClustersSpec diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/SunnyWeatherSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/SunnyWeatherSpec.scala index fcb1393f8a..e36980d859 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/SunnyWeatherSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/SunnyWeatherSpec.scala @@ -24,7 +24,7 @@ object SunnyWeatherMultiJvmSpec extends MultiNodeConfig { gossip-interval = 400 ms nr-of-deputy-nodes = 0 } - akka.loglevel = INFO + akka.loglevel = DEBUG """)) } From 2b69f67777fa3b7d73a97a3afc23ac45182e14b4 Mon Sep 17 00:00:00 2001 From: Patrik Nordwall Date: Fri, 8 Jun 2012 16:54:40 +0200 Subject: [PATCH 6/9] Ignore due to failures, see #2180 --- .../cluster/NodeLeavingAndExitingAndBeingRemovedSpec.scala | 3 ++- .../scala/akka/cluster/NodeLeavingAndExitingSpec.scala | 3 ++- .../src/multi-jvm/scala/akka/cluster/NodeLeavingSpec.scala | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingAndBeingRemovedSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingAndBeingRemovedSpec.scala index 8e274be311..d85016c714 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingAndBeingRemovedSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingAndBeingRemovedSpec.scala @@ -36,7 +36,8 @@ abstract class NodeLeavingAndExitingAndBeingRemovedSpec "A node that is LEAVING a non-singleton cluster" must { - "be moved to EXITING and then to REMOVED by the reaper" taggedAs LongRunningTest in { + // FIXME make it work and remove ignore + "be moved to EXITING and then to REMOVED by the reaper" taggedAs LongRunningTest ignore { awaitClusterUp(first, second, third) diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingSpec.scala index 79fff4770f..2909362fa7 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingAndExitingSpec.scala @@ -42,7 +42,8 @@ abstract class NodeLeavingAndExitingSpec "A node that is LEAVING a non-singleton cluster" must { - "be moved to EXITING by the leader" taggedAs LongRunningTest in { + // FIXME make it work and remove ignore + "be moved to EXITING by the leader" taggedAs LongRunningTest ignore { awaitClusterUp(first, second, third) diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingSpec.scala index b834492045..27bc36a3bf 100644 --- a/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingSpec.scala +++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/NodeLeavingSpec.scala @@ -36,7 +36,8 @@ abstract class NodeLeavingSpec extends MultiNodeSpec(NodeLeavingMultiJvmSpec) "A node that is LEAVING a non-singleton cluster" must { - "be marked as LEAVING in the converged membership table" taggedAs LongRunningTest in { + // FIXME make it work and remove ignore + "be marked as LEAVING in the converged membership table" taggedAs LongRunningTest ignore { awaitClusterUp(first, second, third) From fd455d14bff465d51a5abc1cb98a781cd65147c3 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Fri, 8 Jun 2012 21:57:39 +0200 Subject: [PATCH 7/9] #2208 - Upgrading to Netty 3.5.0 - remove StaticChannelPipeline since it's deprecated. --- .../akka/remote/testconductor/RemoteConnection.scala | 7 +++++-- .../src/main/scala/akka/remote/netty/Client.scala | 2 +- .../scala/akka/remote/netty/NettyRemoteSupport.scala | 9 ++++++--- project/AkkaBuild.scala | 2 +- 4 files changed, 13 insertions(+), 7 deletions(-) diff --git a/akka-remote-tests/src/main/scala/akka/remote/testconductor/RemoteConnection.scala b/akka-remote-tests/src/main/scala/akka/remote/testconductor/RemoteConnection.scala index 5aeb484c42..1979857bf0 100644 --- a/akka-remote-tests/src/main/scala/akka/remote/testconductor/RemoteConnection.scala +++ b/akka-remote-tests/src/main/scala/akka/remote/testconductor/RemoteConnection.scala @@ -3,7 +3,7 @@ */ package akka.remote.testconductor -import org.jboss.netty.channel.{ Channel, ChannelPipeline, ChannelPipelineFactory, ChannelUpstreamHandler, SimpleChannelUpstreamHandler, StaticChannelPipeline } +import org.jboss.netty.channel.{ Channel, ChannelPipeline, ChannelPipelineFactory, ChannelUpstreamHandler, SimpleChannelUpstreamHandler, DefaultChannelPipeline } import org.jboss.netty.channel.socket.nio.{ NioClientSocketChannelFactory, NioServerSocketChannelFactory } import org.jboss.netty.bootstrap.{ ClientBootstrap, ServerBootstrap } import org.jboss.netty.handler.codec.frame.{ LengthFieldBasedFrameDecoder, LengthFieldPrepender } @@ -12,6 +12,7 @@ import org.jboss.netty.handler.codec.protobuf.{ ProtobufDecoder, ProtobufEncoder import org.jboss.netty.handler.timeout.{ ReadTimeoutHandler, ReadTimeoutException } import java.net.InetSocketAddress import java.util.concurrent.Executors +import akka.event.Logging /** * INTERNAL API. @@ -21,7 +22,9 @@ private[akka] class TestConductorPipelineFactory(handler: ChannelUpstreamHandler val encap = List(new LengthFieldPrepender(4), new LengthFieldBasedFrameDecoder(10000, 0, 4, 0, 4)) val proto = List(new ProtobufEncoder, new ProtobufDecoder(TestConductorProtocol.Wrapper.getDefaultInstance)) val msg = List(new MsgEncoder, new MsgDecoder) - new StaticChannelPipeline(encap ::: proto ::: msg ::: handler :: Nil: _*) + (encap ::: proto ::: msg ::: handler :: Nil).foldLeft(new DefaultChannelPipeline) { + (pipe, handler) ⇒ pipe.addLast(Logging.simpleName(handler.getClass), handler); pipe + } } } diff --git a/akka-remote/src/main/scala/akka/remote/netty/Client.scala b/akka-remote/src/main/scala/akka/remote/netty/Client.scala index 86c534c418..e3a2cea9a7 100644 --- a/akka-remote/src/main/scala/akka/remote/netty/Client.scala +++ b/akka-remote/src/main/scala/akka/remote/netty/Client.scala @@ -8,7 +8,7 @@ import java.net.{ InetAddress, InetSocketAddress } import org.jboss.netty.util.{ Timeout, TimerTask, HashedWheelTimer } import org.jboss.netty.bootstrap.ClientBootstrap import org.jboss.netty.channel.group.DefaultChannelGroup -import org.jboss.netty.channel.{ ChannelFutureListener, ChannelHandler, StaticChannelPipeline, MessageEvent, ExceptionEvent, ChannelStateEvent, ChannelPipelineFactory, ChannelPipeline, ChannelHandlerContext, ChannelFuture, Channel } +import org.jboss.netty.channel.{ ChannelFutureListener, ChannelHandler, DefaultChannelPipeline, MessageEvent, ExceptionEvent, ChannelStateEvent, ChannelPipelineFactory, ChannelPipeline, ChannelHandlerContext, ChannelFuture, Channel } import org.jboss.netty.handler.codec.frame.{ LengthFieldPrepender, LengthFieldBasedFrameDecoder } import org.jboss.netty.handler.execution.ExecutionHandler import org.jboss.netty.handler.timeout.{ IdleState, IdleStateEvent, IdleStateAwareChannelHandler, IdleStateHandler } diff --git a/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala b/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala index b42239f470..61124cfecb 100644 --- a/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala +++ b/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala @@ -12,7 +12,7 @@ import java.util.concurrent.Executors import scala.collection.mutable.HashMap import org.jboss.netty.channel.group.{ DefaultChannelGroup, ChannelGroupFuture } import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory -import org.jboss.netty.channel.{ ChannelHandlerContext, Channel, StaticChannelPipeline, ChannelHandler, ChannelPipelineFactory, ChannelLocal } +import org.jboss.netty.channel.{ ChannelHandlerContext, Channel, DefaultChannelPipeline, ChannelHandler, ChannelPipelineFactory, ChannelLocal } import org.jboss.netty.handler.codec.frame.{ LengthFieldPrepender, LengthFieldBasedFrameDecoder } import org.jboss.netty.handler.codec.protobuf.{ ProtobufEncoder, ProtobufDecoder } import org.jboss.netty.handler.execution.{ ExecutionHandler, OrderedMemoryAwareThreadPoolExecutor } @@ -50,10 +50,13 @@ private[akka] class NettyRemoteTransport(_system: ExtendedActorSystem, _provider */ object PipelineFactory { /** - * Construct a StaticChannelPipeline from a sequence of handlers; to be used + * Construct a DefaultChannelPipeline from a sequence of handlers; to be used * in implementations of ChannelPipelineFactory. */ - def apply(handlers: Seq[ChannelHandler]): StaticChannelPipeline = new StaticChannelPipeline(handlers: _*) + def apply(handlers: Seq[ChannelHandler]): DefaultChannelPipeline = + handlers.foldLeft(new DefaultChannelPipeline) { + (pipe, handler) ⇒ pipe.addLast(Logging.simpleName(handler.getClass), handler); pipe + } /** * Constructs the NettyRemoteTransport default pipeline with the give “head” handler, which diff --git a/project/AkkaBuild.scala b/project/AkkaBuild.scala index 736927e7c2..d6d23eb56b 100644 --- a/project/AkkaBuild.scala +++ b/project/AkkaBuild.scala @@ -485,7 +485,7 @@ object Dependency { object V { val Camel = "2.8.0" val Logback = "1.0.4" - val Netty = "3.3.0.Final" + val Netty = "3.5.0.Final" val Protobuf = "2.4.1" val ScalaStm = "0.5" val Scalatest = "1.6.1" From 44fefb9b55dc682fac79fb0337bd0045b37ec728 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Mon, 11 Jun 2012 11:05:19 +0200 Subject: [PATCH 8/9] #2187 - Making Warning sections in docs yellow so they aren't overlooked --- akka-docs/_sphinx/themes/akka/static/docs.css | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/akka-docs/_sphinx/themes/akka/static/docs.css b/akka-docs/_sphinx/themes/akka/static/docs.css index 7b6d3dbf52..3d37718c68 100644 --- a/akka-docs/_sphinx/themes/akka/static/docs.css +++ b/akka-docs/_sphinx/themes/akka/static/docs.css @@ -90,6 +90,42 @@ strong {color: #1d3c52; } box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.25); } +.warning { + background-image: none; + background-color: #fdf5d9; + filter: progid:DXImageTransform.Microsoft.gradient(enabled = false); + padding: 14px; + border-color: #ffffc4; + -webkit-box-shadow: none; + -moz-box-shadow: none; + box-shadow: none; + margin-bottom: 18px; + position: relative; + padding: 7px 15px; + color: #404040; + background-repeat: repeat-x; + background-image: -khtml-gradient(linear, left top, left bottom, from(#ffffc4), to(#ffff00)); + background-image: -moz-linear-gradient(top, #ffffc4, #ffff00); + background-image: -ms-linear-gradient(top, #ffffc4, #ffff00); + background-image: -webkit-gradient(linear, left top, left bottom, color-stop(0%, #ffffc4), color-stop(100%, #ffff00)); + background-image: -webkit-linear-gradient(top, #ffffc4, #ffff00); + background-image: -o-linear-gradient(top, #ffffc4, #ffff00); + background-image: linear-gradient(top, #ffffc4, #ffff00); + filter: progid:DXImageTransform.Microsoft.gradient(startColorstr='#ffffc4', endColorstr='#ffff00', GradientType=0); + text-shadow: 0 -1px 0 rgba(0, 0, 0, 0.25); + border-color: #dff69a #ffff00 #E4C652; + border-color: rgba(0, 0, 0, 0.1) rgba(0, 0, 0, 0.1) rgba(0, 0, 0, 0.25); + text-shadow: 0 1px 0 rgba(255, 255, 255, 0.5); + border-width: 1px; + border-style: solid; + -webkit-border-radius: 4px; + -moz-border-radius: 4px; + border-radius: 4px; + -webkit-box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.25); + -moz-box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.25); + box-shadow: inset 0 1px 0 rgba(255, 255, 255, 0.25); +} + .admonition p.admonition-title { color: rgba(0, 0, 0, 0.6); text-shadow: 0 1px 0 rgba(255, 255, 255, .7); From edc0c0d888d88adf90cc9c3201f9707d6fa1c4b5 Mon Sep 17 00:00:00 2001 From: Viktor Klang Date: Mon, 11 Jun 2012 11:26:28 +0200 Subject: [PATCH 9/9] #2119 - enforce Java6 for releases --- project/scripts/release | 2 ++ 1 file changed, 2 insertions(+) diff --git a/project/scripts/release b/project/scripts/release index 886e6629b1..13795b3d53 100755 --- a/project/scripts/release +++ b/project/scripts/release @@ -93,6 +93,8 @@ fi declare -r version=$1 declare -r publish_path="${release_server}:${release_path}" +[[ `java -version 2>&1 | grep "java version" | awk '{print $3}' | tr -d \" | awk '{split($0, array, ".")} END{print array[2]}'` -eq 6 ]] || fail "Java version is not 1.6" + # check for a git command type -P git &> /dev/null || fail "git command not found"