Revert "Propagate statusPromise down to the actual change in throttler, see #2877"

This reverts commit 9c1a00d020.

akka-remote/src/main/scala/akka/remote/Endpoint.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote
 
 import akka.{ OnlyCauseStackTrace, AkkaException }

akka-remote/src/main/scala/akka/remote/FailureDetector.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote
 
 import java.util.concurrent.TimeUnit._

akka-remote/src/main/scala/akka/remote/PhiAccrualFailureDetector.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote
 
 import akka.remote.FailureDetector.Clock

akka-remote/src/main/scala/akka/remote/Remoting.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote
 
 import scala.language.postfixOps
@@ -403,12 +400,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
 
   val accepting: Receive = {
     case ManagementCommand(cmd, statusPromise) ⇒
-      val allStatuses = transportMapping.values map { transport ⇒
+      transportMapping.values foreach { _.managementCommand(cmd, statusPromise) }
-        val p = Promise[Boolean]()
-        transport.managementCommand(cmd, p)
-        p.future
-      }
-      statusPromise completeWith Future.fold(allStatuses)(true)(_ && _)
 
     case s @ Send(message, senderOption, recipientRef) ⇒
       val recipientAddress = recipientRef.path.address

akka-remote/src/main/scala/akka/remote/RemotingLifecycle.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote
 
 import akka.event.{ LoggingAdapter, Logging }

akka-remote/src/main/scala/akka/remote/transport/AbstractTransportAdapter.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import scala.language.postfixOps

akka-remote/src/main/scala/akka/remote/transport/AkkaPduCodec.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import akka.AkkaException

akka-remote/src/main/scala/akka/remote/transport/AkkaProtocolTransport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import akka.{ OnlyCauseStackTrace, AkkaException }
@@ -378,8 +375,7 @@ private[transport] class ProtocolStateActor(initialData: InitialProtocolStateDat
         case Disassociate ⇒
           stop()
 
-        case Heartbeat ⇒
+        case Heartbeat ⇒ failureDetector.heartbeat(); stay()
-          failureDetector.heartbeat(); stay()
 
         case Payload(payload) ⇒ stateData match {
           case AssociatedWaitHandler(handlerFuture, wrappedHandle, queue) ⇒

akka-remote/src/main/scala/akka/remote/transport/FailureInjectorTransportAdapter.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import FailureInjectorTransportAdapter._

akka-remote/src/main/scala/akka/remote/transport/TestTransport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import TestTransport._

akka-remote/src/main/scala/akka/remote/transport/ThrottlerTransportAdapter.scala

@@ -1,12 +1,8 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import ThrottlerTransportAdapter._
 import akka.actor._
 import akka.pattern.pipe
-import akka.remote.EndpointManager.ManagementCommand
 import akka.remote.transport.ActorTransportAdapter.AssociateUnderlying
 import akka.remote.transport.ActorTransportAdapter.ListenUnderlying
 import akka.remote.transport.ActorTransportAdapter.ListenerRegistered
@@ -21,7 +17,6 @@ import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
 import scala.annotation.tailrec
 import scala.collection.immutable.Queue
-import scala.concurrent.Future
 import scala.concurrent.Promise
 import scala.math.min
 import scala.util.{ Success, Failure }
@@ -112,7 +107,9 @@ class ThrottlerTransportAdapter(_wrappedTransport: Transport, _system: ExtendedA
   protected def managerProps = Props(new ThrottlerManager(wrappedTransport))
 
   override def managementCommand(cmd: Any, statusPromise: Promise[Boolean]): Unit = cmd match {
-    case s: SetThrottle ⇒ manager ! ManagementCommand(s, statusPromise)
+    case s @ SetThrottle(_, _, _) ⇒
+      manager ! s
+      statusPromise.success(true)
     case _ ⇒ wrappedTransport.managementCommand(cmd, statusPromise)
   }
 }
@@ -167,22 +164,17 @@ private[transport] class ThrottlerManager(wrappedTransport: Transport) extends A
       wrappedHandle.readHandlerPromise.future.map { (_, inMode) } pipeTo wrappedHandle.throttlerActor
       handleTable ::= nakedAddress(naked) -> wrappedHandle
       statusPromise.success(wrappedHandle)
-    case ManagementCommand(SetThrottle(address, direction, mode), statusPromise) ⇒
+    case s @ SetThrottle(address, direction, mode) ⇒
       val naked = nakedAddress(address)
       throttlingModes += naked -> (mode, direction)
-      val allStatuses = handleTable.map {
+      handleTable.foreach {
-        case (`naked`, handle) ⇒
+        case (addr, handle) ⇒
-          val p = Promise[Boolean]()
+          if (addr == naked) setMode(handle, mode, direction)
-          setMode(handle, mode, direction, p)
-          p.future
-        case _ ⇒ Future.successful(true)
       }
-      statusPromise completeWith Future.fold(allStatuses)(true)(_ && _)
-
     case Checkin(origin, handle) ⇒
       val naked: Address = nakedAddress(origin)
       handleTable ::= naked -> handle
-      setMode(naked, handle, Promise[Boolean]())
+      setMode(naked, handle)
 
   }
 
@@ -200,19 +192,16 @@ private[transport] class ThrottlerManager(wrappedTransport: Transport) extends A
     }
   }
 
-  private def setMode(nakedAddress: Address, handle: ThrottlerHandle, statusPromise: Promise[Boolean]): Unit = {
+  private def setMode(nakedAddress: Address, handle: ThrottlerHandle): Unit = {
     throttlingModes.get(nakedAddress) match {
-      case Some((mode, direction)) ⇒ setMode(handle, mode, direction, statusPromise)
+      case Some((mode, direction)) ⇒ setMode(handle, mode, direction)
-      case None                    ⇒ setMode(handle, Unthrottled, Direction.Both, statusPromise)
+      case None                    ⇒ setMode(handle, Unthrottled, Direction.Both)
     }
   }
 
-  private def setMode(handle: ThrottlerHandle, mode: ThrottleMode, direction: Direction,
+  private def setMode(handle: ThrottlerHandle, mode: ThrottleMode, direction: Direction): Unit = {
-                      statusPromise: Promise[Boolean]): Unit = {
+    if (direction.includes(Direction.Receive)) handle.throttlerActor ! mode
-    if (direction.includes(Direction.Send))
+    if (direction.includes(Direction.Send)) handle.outboundThrottleMode.set(mode)
-      handle.outboundThrottleMode.set(mode)
-    if (direction.includes(Direction.Receive))
-      handle.throttlerActor ! ManagementCommand(mode, statusPromise)
   }
 
   private def wrapHandle(originalHandle: AssociationHandle, listener: AssociationEventListener, inbound: Boolean): ThrottlerHandle = {
@@ -297,9 +286,9 @@ private[transport] class ThrottledAssociation(
     case Event(InboundPayload(p), _) ⇒
       throttledMessages = throttledMessages enqueue p
       stay()
-    case Event(ManagementCommand(mode: ThrottleMode, statusPromise), ExposedHandle(exposedHandle)) ⇒
+    case Event(mode: ThrottleMode, ExposedHandle(exposedHandle)) ⇒
       inboundThrottleMode = mode
-      try if (mode == Blackhole) {
+      if (inboundThrottleMode == Blackhole) {
         throttledMessages = Queue.empty[ByteString]
         exposedHandle.disassociate()
         stop()
@@ -307,7 +296,7 @@ private[transport] class ThrottledAssociation(
         associationHandler notify InboundAssociation(exposedHandle)
         exposedHandle.readHandlerPromise.future pipeTo self
         goto(WaitUpstreamListener)
-      } finally statusPromise.success(true)
+      }
   }
 
   when(WaitUpstreamListener) {
@@ -332,13 +321,12 @@ private[transport] class ThrottledAssociation(
   }
 
   when(Throttling) {
-    case Event(ManagementCommand(mode: ThrottleMode, statusPromise), _) ⇒
+    case Event(mode: ThrottleMode, _) ⇒
       inboundThrottleMode = mode
-      if (mode == Blackhole) throttledMessages = Queue.empty[ByteString]
+      if (inboundThrottleMode == Blackhole) throttledMessages = Queue.empty[ByteString]
       cancelTimer(DequeueTimerName)
       if (throttledMessages.nonEmpty)
         scheduleDequeue(inboundThrottleMode.timeToAvailable(System.nanoTime(), throttledMessages.head.length))
-      statusPromise.success(true)
       stay()
     case Event(InboundPayload(p), _) ⇒
       forwardOrDelay(p)

akka-remote/src/main/scala/akka/remote/transport/Transport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport
 
 import scala.concurrent.{ Promise, Future }

akka-remote/src/main/scala/akka/remote/transport/netty/NettyHelpers.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport.netty
 
 import akka.AkkaException

akka-remote/src/main/scala/akka/remote/transport/netty/NettyTransport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport.netty
 
 import akka.{ OnlyCauseStackTrace, ConfigurationException }

akka-remote/src/main/scala/akka/remote/transport/netty/TcpSupport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport.netty
 
 import akka.actor.Address

akka-remote/src/main/scala/akka/remote/transport/netty/UdpSupport.scala

@@ -1,6 +1,3 @@
-/**
- * Copyright (C) 2009-2012 Typesafe Inc. <http://www.typesafe.com>
- */
 package akka.remote.transport.netty
 
 import akka.actor.Address