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

Fix compiler warnings in akka-remote, akka-actor-typed. (#26139)

Browse source 
Fix compiler warnings in akka-remote, akka-actor-typed. (#26139)
This commit is contained in:
Helena Edelson 2018-12-20 09:10:54 -08:00 • committed by GitHub
parent 15e2527629
commit 408fab3313
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
40 changed files with 144 additions and 151 deletions
Download patch file Download diff file Expand all files Collapse all files

13
akka-actor-testkit-typed/src/main/scala/akka/actor/testkit/typed/Effect.scala
View file

@ -7,6 +7,7 @@ package akka.actor.testkit.typed
import akka.actor.typed.{ ActorRef, Behavior, Props }
import akka.annotation.{ DoNotInherit, InternalApi }
import akka.util.JavaDurationConverters._
import akka.util.unused
import scala.compat.java8.FunctionConverters._
import scala.concurrent.duration.FiniteDuration
@ -114,18 +115,18 @@ object Effect {
private[akka] final class SpawnedAnonymousAdapter[T](val ref: ActorRef[T])
extends Effect with Product with Serializable {
override def equals(other: Any) = other match {
override def equals(other: Any): Boolean = other match {
case _: SpawnedAnonymousAdapter[_] ⇒ true
case _ ⇒ false
}
override def hashCode: Int = Nil.##
override def toString: String = "SpawnedAnonymousAdapter"
override def productPrefix = "SpawnedAnonymousAdapter"
override def productIterator = Iterator.empty
override def productArity = 0
override def productPrefix: String = "SpawnedAnonymousAdapter"
override def productIterator: Iterator[_] = Iterator.empty
override def productArity: Int = 0
override def productElement(n: Int) = throw new NoSuchElementException
override def canEqual(o: Any) = o.isInstanceOf[SpawnedAnonymousAdapter[_]]
override def canEqual(o: Any): Boolean = o.isInstanceOf[SpawnedAnonymousAdapter[_]]
}
/**
@ -134,7 +135,7 @@ object Effect {
@InternalApi
private[akka] object SpawnedAnonymousAdapter {
def apply[T]() = new SpawnedAnonymousAdapter[T](null)
def unapply[T](s: SpawnedAnonymousAdapter[T]): Boolean = true
def unapply[T](@unused s: SpawnedAnonymousAdapter[T]): Boolean = true
}
/**

9
akka-actor-testkit-typed/src/main/scala/akka/actor/testkit/typed/javadsl/TestProbe.scala
View file

@ -10,16 +10,17 @@ import java.util.{ List ⇒ JList }
import akka.actor.typed.{ ActorRef, ActorSystem }
import akka.annotation.DoNotInherit
import akka.annotation.InternalApi
import akka.actor.testkit.typed.internal.TestProbeImpl
import akka.actor.testkit.typed.{ FishingOutcome, TestKitSettings }
import akka.actor.testkit.typed.scaladsl.TestDuration
import akka.util.JavaDurationConverters._
import akka.util.unused
import scala.collection.immutable
import scala.collection.JavaConverters._
import scala.concurrent.duration.FiniteDuration
import akka.annotation.InternalApi
object FishingOutcomes {
/**
* Consume this message and continue with the next
@ -47,13 +48,13 @@ object TestProbe {
def create[M](system: ActorSystem[_]): TestProbe[M] =
create(name = "testProbe", system)
def create[M](clazz: Class[M], system: ActorSystem[_]): TestProbe[M] =
def create[M](@unused clazz: Class[M], system: ActorSystem[_]): TestProbe[M] =
create(system)
def create[M](name: String, system: ActorSystem[_]): TestProbe[M] =
new TestProbeImpl[M](name, system)
def create[M](name: String, clazz: Class[M], system: ActorSystem[_]): TestProbe[M] =
def create[M](name: String, @unused clazz: Class[M], system: ActorSystem[_]): TestProbe[M] =
new TestProbeImpl[M](name, system)
}

6
akka-actor-tests/src/test/scala/akka/actor/HotSwapSpec.scala
View file

@ -19,7 +19,7 @@ class HotSwapSpec extends AkkaSpec with ImplicitSender {
"be able to become in its constructor" in {
val a = system.actorOf(Props(new Becomer {
context.become { case always ⇒ sender() ! always }
def receive = { case always ⇒ sender() ! "FAILURE" }
def receive = { case _ ⇒ sender() ! "FAILURE" }
}))
a ! "pigdog"
expectMsg("pigdog")
@ -28,7 +28,7 @@ class HotSwapSpec extends AkkaSpec with ImplicitSender {
"be able to become multiple times in its constructor" in {
val a = system.actorOf(Props(new Becomer {
for (i ← 1 to 4) context.become({ case always ⇒ sender() ! i + ":" + always })
def receive = { case always ⇒ sender() ! "FAILURE" }
def receive = { case _ ⇒ sender() ! "FAILURE" }
}))
a ! "pigdog"
expectMsg("4:pigdog")
@ -48,7 +48,7 @@ class HotSwapSpec extends AkkaSpec with ImplicitSender {
"be able to become, with stacking, multiple times in its constructor" in {
val a = system.actorOf(Props(new Becomer {
for (i ← 1 to 4) context.become({ case always ⇒ sender() ! i + ":" + always; context.unbecome() }, false)
def receive = { case always ⇒ sender() ! "FAILURE" }
def receive = { case _ ⇒ sender() ! "FAILURE" }
}))
a ! "pigdog"
a ! "pigdog"

4
akka-actor-tests/src/test/scala/akka/util/ReflectSpec.scala
View file

@ -12,8 +12,8 @@ object ReflectSpec {
final class A
final class B
class One(a: A)
class Two(a: A, b: B)
class One(@unused a: A)
class Two(@unused a: A, @unused b: B)
class MultipleOne(a: A, b: B) {
def this(a: A) { this(a, null) }

4
akka-actor-typed/src/main/scala/akka/actor/typed/Props.scala
View file

@ -99,7 +99,7 @@ abstract class Props private[akka] () extends Product with Serializable {
@tailrec def select(d: Props, acc: List[Props]): List[Props] =
d match {
case EmptyProps ⇒ acc.reverse
case t: T ⇒ select(d.next, (d withNext EmptyProps) :: acc)
case _: T ⇒ select(d.next, (d withNext EmptyProps) :: acc)
case _ ⇒ select(d.next, acc)
}
select(this, Nil)
@ -114,7 +114,7 @@ abstract class Props private[akka] () extends Product with Serializable {
@tailrec def select(d: Props, acc: List[Props]): List[Props] =
d match {
case EmptyProps ⇒ acc
case t: T ⇒ select(d.next, acc)
case _: T ⇒ select(d.next, acc)
case _ ⇒ select(d.next, d :: acc)
}
@tailrec def link(l: List[Props], acc: Props): Props =

14
akka-actor-typed/src/main/scala/akka/actor/typed/internal/Supervision.scala
View file

@ -12,7 +12,7 @@ import akka.actor.typed.BehaviorInterceptor.{ PreStartTarget, ReceiveTarget, Sig
import akka.actor.typed.SupervisorStrategy._
import akka.actor.typed.scaladsl.Behaviors
import akka.annotation.InternalApi
import akka.util.OptionVal
import akka.util.{ OptionVal, unused }
import scala.concurrent.duration.{ Deadline, FiniteDuration }
import scala.reflect.ClassTag
@ -30,7 +30,7 @@ import scala.util.control.NonFatal
case r: Restart ⇒
Behaviors.intercept[T, T](new RestartSupervisor(initialBehavior, r))(initialBehavior)
case r: Stop ⇒
Behaviors.intercept[T, T](new StopSupervisor(initialBehavior, r))(initialBehavior)
Behaviors.intercept[T, T](new StopSupervisor(r))(initialBehavior)
case r: Backoff ⇒
Behaviors.intercept[T, T](new BackoffSupervisor(initialBehavior, r))(initialBehavior)
}
@ -86,7 +86,7 @@ private abstract class SimpleSupervisor[T, Thr <: Throwable: ClassTag](ss: Super
} catch handleReceiveException(ctx, target)
}
protected def handleException(ctx: ActorContext[T]): Catcher[Behavior[T]] = {
protected def handleException(@unused ctx: ActorContext[T]): Catcher[Behavior[T]] = {
case NonFatal(t: Thr) ⇒
Behavior.failed(t)
}
@ -100,7 +100,7 @@ private abstract class SimpleSupervisor[T, Thr <: Throwable: ClassTag](ss: Super
handleException(ctx)
}
private class StopSupervisor[T, Thr <: Throwable: ClassTag](initial: Behavior[T], strategy: Stop) extends SimpleSupervisor[T, Thr](strategy) {
private class StopSupervisor[T, Thr <: Throwable: ClassTag](strategy: Stop) extends SimpleSupervisor[T, Thr](strategy) {
override def handleException(ctx: ActorContext[T]): Catcher[Behavior[T]] = {
case NonFatal(t: Thr) ⇒
log(ctx, t)
@ -137,13 +137,13 @@ private class RestartSupervisor[T, Thr <: Throwable](initial: Behavior[T], strat
throw t
} else {
log(ctx, t)
restart(ctx, t)
restart()
aroundStart(ctx, target)
}
}
}
private def restart(ctx: ActorContext[_], t: Throwable) = {
private def restart() = {
val timeLeft = deadlineHasTimeLeft
val newDeadline = if (deadline.isDefined && timeLeft) deadline else OptionVal.Some(Deadline.now + strategy.withinTimeRange)
restarts = if (timeLeft) restarts + 1 else 1
@ -161,7 +161,7 @@ private class RestartSupervisor[T, Thr <: Throwable](initial: Behavior[T], strat
case NonFatal(ex) ⇒ ctx.asScala.log.error(ex, "failure during PreRestart")
}
log(ctx, t)
restart(ctx, t)
restart()
Behavior.validateAsInitial(Behavior.start(initial, ctx))
}
}

6
akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/ActorAdapter.scala
View file

@ -135,9 +135,9 @@ import akka.util.OptionVal
}
override def unhandled(msg: Any): Unit = msg match {
case t @ Terminated(ref) ⇒ throw DeathPactException(ref)
case msg: Signal ⇒ // that's ok
case other ⇒ super.unhandled(other)
case Terminated(ref) ⇒ throw DeathPactException(ref)
case _: Signal ⇒ // that's ok
case other ⇒ super.unhandled(other)
}
override val supervisorStrategy = untyped.OneForOneStrategy(loggingEnabled = false) {

6
akka-actor-typed/src/main/scala/akka/actor/typed/internal/adapter/ActorRefAdapter.scala
View file

@ -60,8 +60,8 @@ private[akka] object ActorRefAdapter {
sysmsg.Watch(
toUntyped(watchee),
toUntyped(watcher)))
case internal.Unwatch(watchee, watcher) ⇒ untyped.sendSystemMessage(sysmsg.Unwatch(toUntyped(watchee), toUntyped(watcher)))
case internal.DeathWatchNotification(ref, cause) ⇒ untyped.sendSystemMessage(sysmsg.DeathWatchNotification(toUntyped(ref), true, false))
case internal.NoMessage ⇒ // just to suppress the warning
case internal.Unwatch(watchee, watcher) ⇒ untyped.sendSystemMessage(sysmsg.Unwatch(toUntyped(watchee), toUntyped(watcher)))
case internal.DeathWatchNotification(ref, _) ⇒ untyped.sendSystemMessage(sysmsg.DeathWatchNotification(toUntyped(ref), true, false))
case internal.NoMessage ⇒ // just to suppress the warning
}
}

3
akka-actor-typed/src/main/scala/akka/actor/typed/javadsl/Behaviors.scala
View file

@ -13,6 +13,7 @@ import akka.actor.typed.scaladsl
import akka.annotation.ApiMayChange
import akka.japi.function.{ Function2 ⇒ JapiFunction2 }
import akka.japi.pf.PFBuilder
import akka.util.unused
import scala.collection.JavaConverters._
import scala.reflect.ClassTag
@ -156,7 +157,7 @@ object Behaviors {
* @param type the supertype of all messages accepted by this behavior
* @return the behavior builder
*/
def receive[T](`type`: Class[T]): BehaviorBuilder[T] = BehaviorBuilder.create[T]
def receive[T](@unused `type`: Class[T]): BehaviorBuilder[T] = BehaviorBuilder.create[T]
/**
* Construct an actor behavior that can react to lifecycle signals only.

8
akka-actor-typed/src/main/scala/akka/actor/typed/scaladsl/AskPattern.scala
View file

@ -6,14 +6,14 @@ package akka.actor.typed.scaladsl
import java.util.concurrent.TimeoutException
import scala.concurrent.Future
import akka.actor.{ Address, RootActorPath, Scheduler }
import akka.actor.typed.ActorRef
import akka.actor.typed.internal.{ adapter ⇒ adapt }
import akka.annotation.InternalApi
import akka.pattern.PromiseActorRef
import akka.util.Timeout
import scala.concurrent.Future
import akka.util.{ Timeout, unused }
import akka.actor.typed.RecipientRef
import akka.actor.typed.internal.InternalRecipientRef
@ -54,7 +54,7 @@ object AskPattern {
* val f: Future[Reply] = target ? replyTo => (Request("hello", replyTo))
* }}}
*/
def ?[U](replyTo: ActorRef[U] ⇒ T)(implicit timeout: Timeout, scheduler: Scheduler): Future[U] = {
def ?[U](replyTo: ActorRef[U] ⇒ T)(implicit timeout: Timeout, @unused scheduler: Scheduler): Future[U] = {
// We do not currently use the implicit scheduler, but want to require it
// because it might be needed when we move to a 'native' typed runtime, see #24219
ref match {

2
akka-actor/src/main/scala/akka/actor/dsl/Inbox.scala
View file

@ -97,7 +97,7 @@ trait Inbox { this: ActorDSL.type ⇒
case g: Get ⇒
if (messages.isEmpty) enqueueQuery(g)
else sender() ! messages.dequeue()
case s @ Select(_, predicate, _) ⇒
case s: Select ⇒
if (messages.isEmpty) enqueueQuery(s)
else {
currentSelect = s

10
akka-remote/src/main/scala/akka/remote/Endpoint.scala
View file

@ -220,7 +220,7 @@ private[remote] class ReliableDeliverySupervisor(
settings.SysResendTimeout, settings.SysResendTimeout, self, AttemptSysMsgRedelivery)
override val supervisorStrategy = OneForOneStrategy(loggingEnabled = false) {
case e @ (_: AssociationProblem) ⇒ Escalate
case _: AssociationProblem ⇒ Escalate
case NonFatal(e) ⇒
val causedBy = if (e.getCause == null) "" else s"Caused by: [${e.getCause.getMessage}]"
log.warning(
@ -359,10 +359,10 @@ private[remote] class ReliableDeliverySupervisor(
// Resending will be triggered by the incoming GotUid message after the connection finished
goToActive()
} else goToIdle()
case AttemptSysMsgRedelivery ⇒ // Ignore
case s @ Send(msg: SystemMessage, _, _, _) ⇒ tryBuffer(s.copy(seqOpt = Some(nextSeq())))
case s: Send ⇒ context.system.deadLetters ! s
case EndpointWriter.FlushAndStop ⇒ context.stop(self)
case AttemptSysMsgRedelivery ⇒ // Ignore
case s @ Send(_: SystemMessage, _, _, _) ⇒ tryBuffer(s.copy(seqOpt = Some(nextSeq())))
case s: Send ⇒ context.system.deadLetters ! s
case EndpointWriter.FlushAndStop ⇒ context.stop(self)
case EndpointWriter.StopReading(w, replyTo) ⇒
replyTo ! EndpointWriter.StoppedReading(w)
sender() ! EndpointWriter.StoppedReading(w)

10
akka-remote/src/main/scala/akka/remote/MessageSerializer.scala
View file

@ -10,6 +10,7 @@ import akka.actor.ExtendedActorSystem
import akka.annotation.InternalApi
import akka.remote.artery.{ EnvelopeBuffer, HeaderBuilder, OutboundEnvelope }
import akka.serialization._
import akka.util.unused
import scala.util.control.NonFatal
@ -83,8 +84,13 @@ private[akka] object MessageSerializer {
} finally Serialization.currentTransportInformation.value = oldInfo
}
def deserializeForArtery(system: ExtendedActorSystem, originUid: Long, serialization: Serialization,
serializer: Int, classManifest: String, envelope: EnvelopeBuffer): AnyRef = {
def deserializeForArtery(
@unused system: ExtendedActorSystem,
@unused originUid: Long,
serialization: Serialization,
serializer: Int,
classManifest: String,
envelope: EnvelopeBuffer): AnyRef = {
serialization.deserializeByteBuffer(envelope.byteBuffer, serializer, classManifest)
}
}

2
akka-remote/src/main/scala/akka/remote/RemoteDaemon.scala
View file

@ -212,7 +212,7 @@ private[akka] class RemoteSystemDaemon(
private def doCreateActor(message: DaemonMsg, props: Props, deploy: Deploy, path: String, supervisor: ActorRef) = {
path match {
case ActorPathExtractor(address, elems) if elems.nonEmpty && elems.head == "remote" ⇒
case ActorPathExtractor(_, elems) if elems.nonEmpty && elems.head == "remote" ⇒
// TODO RK currently the extracted “address” is just ignored, is that okay?
// TODO RK canonicalize path so as not to duplicate it always #1446
val subpath = elems.drop(1)

5
akka-remote/src/main/scala/akka/remote/RemoteTransport.scala
View file

@ -8,10 +8,11 @@ import akka.AkkaException
import akka.Done
import akka.actor._
import akka.event.LoggingAdapter
import scala.collection.immutable
import scala.concurrent.Future
import scala.util.control.NoStackTrace
import akka.util.OptionVal
import akka.util.{ OptionVal, unused }
/**
* RemoteTransportException represents a general failure within a RemoteTransport,
@ -79,7 +80,7 @@ private[akka] abstract class RemoteTransport(val system: ExtendedActorSystem, va
* @param cmd Command message to send to the transports.
* @return A Future that indicates when the message was successfully handled or dropped.
*/
def managementCommand(cmd: Any): Future[Boolean] = { Future.successful(false) }
def managementCommand(@unused cmd: Any): Future[Boolean] = { Future.successful(false) }
/**
* A Logger that can be used to log issues that may occur

21
akka-remote/src/main/scala/akka/remote/Remoting.scala
View file

@ -106,7 +106,7 @@ private[remote] object Remoting {
private[Remoting] class TransportSupervisor extends Actor with RequiresMessageQueue[UnboundedMessageQueueSemantics] {
override def supervisorStrategy = OneForOneStrategy() {
case NonFatal(e) ⇒ Restart
case NonFatal(_) ⇒ Restart
}
def receive = {
@ -327,7 +327,7 @@ private[remote] object EndpointManager {
def registerWritableEndpointUid(remoteAddress: Address, uid: Int): Unit = {
addressToWritable.get(remoteAddress) match {
case Some(Pass(ep, _)) ⇒ addressToWritable += remoteAddress → Pass(ep, Some(uid))
case other ⇒
case _ ⇒
}
}
@ -482,7 +482,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
override val supervisorStrategy = {
def hopeless(e: HopelessAssociation): SupervisorStrategy.Directive = e match {
case HopelessAssociation(localAddress, remoteAddress, Some(uid), reason) ⇒
case HopelessAssociation(_, remoteAddress, Some(uid), reason) ⇒
log.error(reason, "Association to [{}] with UID [{}] irrecoverably failed. Quarantining address.",
remoteAddress, uid)
settings.QuarantineDuration match {
@ -493,7 +493,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
}
Stop
case HopelessAssociation(localAddress, remoteAddress, None, _) ⇒
case HopelessAssociation(_, remoteAddress, None, _) ⇒
keepQuarantinedOr(remoteAddress) {
log.warning(
"Association to [{}] with unknown UID is irrecoverably failed. " +
@ -505,7 +505,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
}
OneForOneStrategy(loggingEnabled = false) {
case e @ InvalidAssociation(localAddress, remoteAddress, reason, disassiciationInfo) ⇒
case InvalidAssociation(localAddress, remoteAddress, reason, disassiciationInfo) ⇒
keepQuarantinedOr(remoteAddress) {
val causedBy = if (reason.getCause == null) "" else s"Caused by: [${reason.getCause.getMessage}]"
log.warning(
@ -522,7 +522,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
}
Stop
case ShutDownAssociation(localAddress, remoteAddress, _) ⇒
case ShutDownAssociation(_, remoteAddress, _) ⇒
keepQuarantinedOr(remoteAddress) {
log.debug(
"Remote system with address [{}] has shut down. " +
@ -655,7 +655,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
}
}
case s @ Send(message, senderOption, recipientRef, _) ⇒
case s @ Send(_, _, recipientRef, _) ⇒
val recipientAddress = recipientRef.path.address
def createAndRegisterWritingEndpoint(): ActorRef = {
@ -677,7 +677,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
case Some(Gated(timeOfRelease)) ⇒
if (timeOfRelease.isOverdue()) createAndRegisterWritingEndpoint() ! s
else extendedSystem.deadLetters ! s
case Some(Quarantined(uid, _)) ⇒
case Some(Quarantined(_, _)) ⇒
// timeOfRelease is only used for garbage collection reasons, therefore it is ignored here. We still have
// the Quarantined tombstone and we know what UID we don't want to accept, so use it.
createAndRegisterWritingEndpoint() ! s
@ -686,7 +686,7 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
}
case ia @ InboundAssociation(handle: AkkaProtocolHandle) ⇒
case ia @ InboundAssociation(_: AkkaProtocolHandle) ⇒
handleInboundAssociation(ia, writerIsIdle = false)
case EndpointWriter.StoppedReading(endpoint) ⇒
acceptPendingReader(takingOverFrom = endpoint)
@ -776,10 +776,11 @@ private[remote] class EndpointManager(conf: Config, log: LoggingAdapter) extends
endpoints.markAsQuarantined(handle.remoteAddress, uid, Deadline.now + settings.QuarantineDuration)
createAndRegisterEndpoint(handle)
}
case state ⇒
case _ ⇒
createAndRegisterEndpoint(handle)
}
}
case _ ⇒ // ignore
}
private def createAndRegisterEndpoint(handle: AkkaProtocolHandle): Unit = {

11
akka-remote/src/main/scala/akka/remote/artery/ArteryTransport.scala
View file

@ -24,7 +24,6 @@ import scala.util.Success
import scala.util.Try
import scala.util.control.NoStackTrace
import scala.util.control.NonFatal
import akka.Done
import akka.NotUsed
import akka.actor.Actor
@ -56,8 +55,7 @@ import akka.stream.SharedKillSwitch
import akka.stream.scaladsl.Flow
import akka.stream.scaladsl.Keep
import akka.stream.scaladsl.Sink
import akka.util.OptionVal
import akka.util.WildcardIndex
import akka.util.{ OptionVal, WildcardIndex, unused }
/**
* INTERNAL API
@ -243,8 +241,11 @@ private[remote] object FlushOnShutdown {
/**
* INTERNAL API
*/
private[remote] class FlushOnShutdown(done: Promise[Done], timeout: FiniteDuration,
inboundContext: InboundContext, associations: Set[Association]) extends Actor {
private[remote] class FlushOnShutdown(
done: Promise[Done],
timeout: FiniteDuration,
@unused inboundContext: InboundContext,
associations: Set[Association]) extends Actor {
var remaining = Map.empty[UniqueAddress, Int]

7
akka-remote/src/main/scala/akka/remote/artery/Association.scala
View file

@ -172,13 +172,10 @@ private[remote] class Association(
@volatile private[this] var queuesVisibility = false
private def controlQueue: SendQueue.ProducerApi[OutboundEnvelope] = queues(ControlQueueIndex)
private def largeQueue: SendQueue.ProducerApi[OutboundEnvelope] = queues(LargeQueueIndex)
@volatile private[this] var _outboundControlIngress: OptionVal[OutboundControlIngress] = OptionVal.None
@volatile private[this] var materializing = new CountDownLatch(1)
@volatile private[this] var outboundCompressionAccess: Vector[OutboundCompressionAccess] = Vector.empty
// in case there is a restart at the same time as a compression table update
private val changeCompressionTimeout = 5.seconds
// keyed by stream queue index
private[this] val streamMatValues = new AtomicReference(Map.empty[Int, OutboundStreamMatValues])
@ -335,7 +332,7 @@ private[remote] class Association(
if (log.isDebugEnabled) {
val reason =
if (removed) "removed unused quarantined association"
else s"overflow of send queue, size [$queueSize]"
else s"overflow of send queue, size [$qSize]"
log.debug(
"Dropping message [{}] from [{}] to [{}] due to {}",
Logging.messageClassName(message), sender.getOrElse(deadletters), recipient.getOrElse(recipient), reason)
@ -718,7 +715,7 @@ private[remote] class Association(
queues(queueIndex) = wrapper // use new underlying queue immediately for restarts
queuesVisibility = true // volatile write for visibility of the queues array
val (queueValue, testMgmt, changeCompression, completed) =
val (queueValue, _, changeCompression, completed) =
Source.fromGraph(new SendQueue[OutboundEnvelope](sendToDeadLetters))
.via(streamKillSwitch.flow)
.viaMat(transport.outboundTestFlow(this))(Keep.both)

20
akka-remote/src/main/scala/akka/remote/artery/Codecs.scala
View file

@ -6,6 +6,10 @@ package akka.remote.artery
import java.util.concurrent.TimeUnit
import scala.concurrent.duration._
import scala.concurrent.{ Future, Promise }
import scala.util.control.NonFatal
import akka.Done
import akka.actor.{ EmptyLocalActorRef, _ }
import akka.event.Logging
@ -17,11 +21,7 @@ import akka.remote.{ MessageSerializer, OversizedPayloadException, RemoteActorRe
import akka.serialization.{ Serialization, SerializationExtension, Serializers }
import akka.stream._
import akka.stream.stage._
import akka.util.{ OptionVal, Unsafe }
import scala.concurrent.duration._
import scala.concurrent.{ Future, Promise }
import scala.util.control.NonFatal
import akka.util.{ OptionVal, Unsafe, unused }
import akka.remote.artery.OutboundHandshake.HandshakeReq
/**
@ -43,7 +43,7 @@ private[remote] class Encoder(
system: ExtendedActorSystem,
outboundEnvelopePool: ObjectPool[ReusableOutboundEnvelope],
bufferPool: EnvelopeBufferPool,
streamId: Int,
@unused streamId: Int,
debugLogSend: Boolean,
version: Byte)
extends GraphStageWithMaterializedValue[FlowShape[OutboundEnvelope, EnvelopeBuffer], Encoder.OutboundCompressionAccess] {
@ -59,7 +59,6 @@ private[remote] class Encoder(
private val headerBuilder = HeaderBuilder.out()
headerBuilder.setVersion(version)
headerBuilder.setUid(uniqueLocalAddress.uid)
private val localAddress = uniqueLocalAddress.address
// lazy init of SerializationExtension to avoid loading serializers before ActorRefProvider has been initialized
private var _serialization: OptionVal[Serialization] = OptionVal.None
@ -331,7 +330,6 @@ private[remote] class Decoder(
override val compressions = inboundCompressions
private val localAddress = inboundContext.localAddress.address
private val headerBuilder = HeaderBuilder.in(compressions)
private val actorRefResolver: ActorRefResolveCacheWithAddress =
new ActorRefResolveCacheWithAddress(system.provider.asInstanceOf[RemoteActorRefProvider], uniqueLocalAddress)
@ -575,9 +573,9 @@ private[remote] class Decoder(
* INTERNAL API
*/
private[remote] class Deserializer(
inboundContext: InboundContext,
system: ExtendedActorSystem,
bufferPool: EnvelopeBufferPool) extends GraphStage[FlowShape[InboundEnvelope, InboundEnvelope]] {
@unused inboundContext: InboundContext,
system: ExtendedActorSystem,
bufferPool: EnvelopeBufferPool) extends GraphStage[FlowShape[InboundEnvelope, InboundEnvelope]] {
val in: Inlet[InboundEnvelope] = Inlet("Artery.Deserializer.in")
val out: Outlet[InboundEnvelope] = Outlet("Artery.Deserializer.out")

13
akka-remote/src/main/scala/akka/remote/artery/Handshake.scala
View file

@ -4,20 +4,19 @@
package akka.remote.artery
import akka.actor.ActorSystem
import scala.concurrent.duration._
import scala.concurrent.Future
import scala.util.control.NoStackTrace
import akka.actor.ActorSystem
import akka.remote.UniqueAddress
import akka.stream.Attributes
import akka.stream.FlowShape
import akka.stream.Inlet
import akka.stream.Outlet
import akka.stream.stage._
import akka.util.OptionVal
import akka.util.{ OptionVal, unused }
import akka.Done
import scala.concurrent.Future
import akka.actor.Address
/**
@ -50,7 +49,7 @@ private[remote] object OutboundHandshake {
* INTERNAL API
*/
private[remote] class OutboundHandshake(
system: ActorSystem,
@unused system: ActorSystem,
outboundContext: OutboundContext,
outboundEnvelopePool: ObjectPool[ReusableOutboundEnvelope],
timeout: FiniteDuration,
@ -129,7 +128,7 @@ private[remote] class OutboundHandshake(
// when it receives the HandshakeRsp reply
implicit val ec = materializer.executionContext
uniqueRemoteAddress.foreach {
getAsyncCallback[UniqueAddress] { a ⇒
getAsyncCallback[UniqueAddress] { _ ⇒
if (handshakeState != Completed) {
handshakeCompleted()
if (isAvailable(out))

1
akka-remote/src/main/scala/akka/remote/artery/MessageDispatcher.scala
View file

@ -41,7 +41,6 @@ private[remote] class MessageDispatcher(
}
val sender: ActorRef = senderOption.getOrElse(system.deadLetters)
val originalReceiver = recipient.path
recipient match {

9
akka-remote/src/main/scala/akka/remote/artery/RemoteInstrument.scala
View file

@ -4,12 +4,13 @@
package akka.remote.artery
import akka.actor.{ ActorRef, ExtendedActorSystem }
import akka.event.{ Logging, LoggingAdapter }
import akka.util.OptionVal
import java.nio.ByteBuffer
import scala.annotation.tailrec
import scala.util.control.NonFatal
import akka.actor.{ ActorRef, ExtendedActorSystem }
import akka.event.{ Logging, LoggingAdapter }
import akka.util.{ OptionVal, unused }
/**
* INTERNAL API
@ -287,7 +288,7 @@ private[remote] object RemoteInstruments {
def getKey(kl: Int): Byte = (kl >>> 26).toByte
def getLength(kl: Int): Int = kl & lengthMask
def create(system: ExtendedActorSystem, log: LoggingAdapter): Vector[RemoteInstrument] = {
def create(system: ExtendedActorSystem, @unused log: LoggingAdapter): Vector[RemoteInstrument] = {
val c = system.settings.config
val path = "akka.remote.artery.advanced.instruments"
import scala.collection.JavaConverters._

2
akka-remote/src/main/scala/akka/remote/artery/SystemMessageDelivery.scala
View file

@ -91,7 +91,6 @@ import akka.util.OptionVal
private var incarnation = outboundContext.associationState.incarnation
private val unacknowledged = new ArrayDeque[OutboundEnvelope]
private var resending = new ArrayDeque[OutboundEnvelope]
private var resendingFromSeqNo = -1L
private var stopping = false
private val giveUpAfterNanos = outboundContext.settings.Advanced.GiveUpSystemMessageAfter.toNanos
@ -288,7 +287,6 @@ import akka.util.OptionVal
incarnation = outboundContext.associationState.incarnation
unacknowledged.clear()
resending.clear()
resendingFromSeqNo = -1L
cancelTimer(resendInterval)
}

2
akka-remote/src/main/scala/akka/remote/artery/aeron/AeronSource.scala
View file

@ -34,7 +34,7 @@ private[remote] object AeronSource {
() ⇒
{
handler.reset
val fragmentsRead = sub.poll(handler.fragmentsHandler, 1)
sub.poll(handler.fragmentsHandler, 1)
val msg = handler.messageReceived
handler.reset() // for GC
if (msg ne null) {

4
akka-remote/src/main/scala/akka/remote/artery/compress/InboundCompressions.scala
View file

@ -14,7 +14,7 @@ import akka.actor.Address
import akka.event.Logging
import akka.event.LoggingAdapter
import akka.remote.artery._
import akka.util.OptionVal
import akka.util.{ OptionVal, unused }
import org.agrona.collections.Long2ObjectHashMap
/**
@ -376,7 +376,7 @@ private[remote] abstract class InboundCompression[T >: Null](
* Add `n` occurrence for the given key and call `heavyHittedDetected` if element has become a heavy hitter.
* Empty keys are omitted.
*/
def increment(remoteAddress: Address, value: T, n: Long): Unit = {
def increment(@unused remoteAddress: Address, value: T, n: Long): Unit = {
val count = cms.addObjectAndEstimateCount(value, n)
addAndCheckIfheavyHitterDetected(value, count)
alive = true

4
akka-remote/src/main/scala/akka/remote/artery/compress/TopHeavyHitters.scala
View file

@ -169,7 +169,7 @@ private[remote] final class TopHeavyHitters[T >: Null](val max: Int)(implicit cl
true
} else {
// The entry exists, let's update it.
updateExistingHeavyHitter(actualIdx, hashCode, item, count)
updateExistingHeavyHitter(actualIdx, count)
// not a "new" heavy hitter, since we only replaced it (so it was signaled as new once before)
false
}
@ -220,7 +220,7 @@ private[remote] final class TopHeavyHitters[T >: Null](val max: Int)(implicit cl
* Replace existing heavy hitter – give it a new `count` value. This will also restore the heap property, so this
* might make a previously lowest hitter no longer be one.
*/
private def updateExistingHeavyHitter(foundHashIndex: Int, hashCode: HashCodeVal, item: T, count: Long): Unit = {
private def updateExistingHeavyHitter(foundHashIndex: Int, count: Long): Unit = {
if (weights(foundHashIndex) > count)
throw new IllegalArgumentException(s"Weights can be only incremented or kept the same, not decremented. " +
s"Previous weight was [${weights(foundHashIndex)}], attempted to modify it to [$count].")

17
akka-remote/src/main/scala/akka/remote/artery/tcp/ArteryTcpTransport.scala
View file

@ -41,10 +41,8 @@ import akka.stream.scaladsl.RestartFlow
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source
import akka.stream.scaladsl.Tcp
import akka.stream.scaladsl.Tcp.IncomingConnection
import akka.stream.scaladsl.Tcp.ServerBinding
import akka.util.ByteString
import akka.util.OptionVal
import akka.util.{ ByteString, OptionVal }
/**
* INTERNAL API
@ -242,7 +240,7 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
// If something in the inboundConnectionFlow fails, e.g. framing, the connection will be teared down,
// but other parts of the inbound streams don't have to restarted.
def inboundConnectionFlow(inboundConnection: IncomingConnection): Flow[ByteString, ByteString, NotUsed] = {
def inboundConnectionFlow: Flow[ByteString, ByteString, NotUsed] = {
// must create new Flow for each connection because of the FlightRecorder that can't be shared
val afr = createFlightRecorderEventSink()
Flow[ByteString]
@ -279,7 +277,7 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
afr.loFreq(
TcpInbound_Connected,
s"${connection.remoteAddress.getHostString}:${connection.remoteAddress.getPort}")
connection.handleWith(inboundConnectionFlow(connection))
connection.handleWith(inboundConnectionFlow)
})
.run()
.recoverWith {
@ -329,8 +327,7 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
.toMat(inboundControlSink)({ case (a, (c, d)) ⇒ (a, c, d) })
.run()(controlMaterializer)
attachControlMessageObserver(ctrl)
implicit val ec: ExecutionContext = materializer.executionContext
updateStreamMatValues(ControlStreamId, completed)
updateStreamMatValues(completed)
(hub, completed)
}
@ -380,7 +377,7 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
setInboundCompressionAccess(inboundCompressionAccess)
updateStreamMatValues(OrdinaryStreamId, completed)
updateStreamMatValues(completed)
(inboundHub, completed)
}
@ -395,12 +392,12 @@ private[remote] class ArteryTcpTransport(_system: ExtendedActorSystem, _provider
.toMat(inboundSink(largeEnvelopeBufferPool))(Keep.both)
.run()(materializer)
updateStreamMatValues(LargeStreamId, completed)
updateStreamMatValues(completed)
(hub, completed)
}
private def updateStreamMatValues(streamId: Int, completed: Future[Done]): Unit = {
private def updateStreamMatValues(completed: Future[Done]): Unit = {
implicit val ec: ExecutionContext = materializer.executionContext
updateStreamMatValues(ControlStreamId, InboundStreamMatValues[NotUsed](
NotUsed,

5
akka-remote/src/main/scala/akka/remote/artery/tcp/SSLEngineProvider.scala
View file

@ -24,8 +24,6 @@ import akka.event.LogMarker
import akka.event.Logging
import akka.event.MarkerLoggingAdapter
import akka.japi.Util.immutableSeq
import akka.stream.IgnoreComplete
import akka.stream.TLSClosing
import akka.stream.TLSRole
import com.typesafe.config.Config
import javax.net.ssl.KeyManager
@ -158,8 +156,7 @@ class SslTransportException(message: String, cause: Throwable) extends RuntimeEx
sslContext: SSLContext,
role: TLSRole,
hostname: String,
port: Int,
closing: TLSClosing = IgnoreComplete): SSLEngine = {
port: Int): SSLEngine = {
val engine = sslContext.createSSLEngine(hostname, port)

4
akka-remote/src/main/scala/akka/remote/routing/RemoteRouterConfig.scala
View file

@ -69,9 +69,9 @@ final case class RemoteRouterConfig(local: Pool, nodes: Iterable[Address]) exten
override def resizer: Option[Resizer] = local.resizer
override def withFallback(other: RouterConfig): RouterConfig = other match {
case RemoteRouterConfig(local: RemoteRouterConfig, nodes) ⇒ throw new IllegalStateException(
case RemoteRouterConfig(_: RemoteRouterConfig, _) ⇒ throw new IllegalStateException(
"RemoteRouterConfig is not allowed to wrap a RemoteRouterConfig")
case RemoteRouterConfig(local: Pool, nodes) ⇒
case RemoteRouterConfig(local: Pool, _) ⇒
copy(local = this.local.withFallback(local).asInstanceOf[Pool])
case _ ⇒ copy(local = this.local.withFallback(other).asInstanceOf[Pool])
}

4
akka-remote/src/main/scala/akka/remote/serialization/ActorRefResolveCache.scala
View file

@ -13,7 +13,7 @@ import akka.actor.ExtensionId
import akka.actor.ExtensionIdProvider
import akka.remote.RemoteActorRefProvider
import akka.remote.artery.LruBoundedCache
import akka.util.Unsafe
import akka.util.{ Unsafe, unused }
/**
* INTERNAL API: Thread local cache per actor system
@ -45,7 +45,7 @@ private[akka] class ActorRefResolveThreadLocalCache(val system: ExtendedActorSys
override def initialValue: ActorRefResolveCache = new ActorRefResolveCache(provider)
}
def threadLocalCache(provider: RemoteActorRefProvider): ActorRefResolveCache =
def threadLocalCache(@unused provider: RemoteActorRefProvider): ActorRefResolveCache =
current.get
}

4
akka-remote/src/main/scala/akka/remote/serialization/DaemonMsgCreateSerializer.scala
View file

@ -6,11 +6,11 @@ package akka.remote.serialization
import akka.serialization.{ BaseSerializer, SerializationExtension, SerializerWithStringManifest }
import akka.protobuf.ByteString
import com.typesafe.config.{ Config, ConfigFactory }
import akka.actor.{ Deploy, ExtendedActorSystem, NoScopeGiven, Props, Scope }
import akka.remote.DaemonMsgCreate
import akka.remote.WireFormats.{ DaemonMsgCreateData, DeployData, PropsData }
import akka.routing.{ NoRouter, RouterConfig }
import com.typesafe.config.{ Config, ConfigFactory }
import scala.reflect.ClassTag
import util.{ Failure, Success }
@ -177,8 +177,6 @@ private[akka] final class DaemonMsgCreateSerializer(val system: ExtendedActorSys
supervisor = deserializeActorRef(system, proto.getSupervisor))
}
private def oldSerialize(any: Any): ByteString = ByteString.copyFrom(serialization.serialize(any.asInstanceOf[AnyRef]).get)
private def serialize(any: Any): (Int, Boolean, String, Array[Byte]) = {
val m = any.asInstanceOf[AnyRef]
val serializer = serialization.findSerializerFor(m)

4
akka-remote/src/main/scala/akka/remote/transport/AkkaProtocolTransport.scala
View file

@ -554,7 +554,7 @@ private[transport] class ProtocolStateActor(
}
onTermination {
case StopEvent(reason, _, OutboundUnassociated(remoteAddress, statusPromise, transport)) ⇒
case StopEvent(reason, _, OutboundUnassociated(_, statusPromise, _)) ⇒
statusPromise.tryFailure(reason match {
case FSM.Failure(info: DisassociateInfo) ⇒ disassociateException(info)
case _ ⇒ new AkkaProtocolException("Transport disassociated before handshake finished")
@ -611,7 +611,7 @@ private[transport] class ProtocolStateActor(
case FSM.Failure(ForbiddenUidReason) ⇒ // no logging
case FSM.Failure(TimeoutReason(errorMessage)) ⇒
log.info(errorMessage)
case other ⇒ super.logTermination(reason)
case _ ⇒ super.logTermination(reason)
}
private def listenForListenerRegistration(readHandlerPromise: Promise[HandleEventListener]): Unit =

15
akka-remote/src/main/scala/akka/remote/transport/TestTransport.scala
View file

@ -4,16 +4,17 @@
package akka.remote.transport
import TestTransport._
import java.util.concurrent.{ CopyOnWriteArrayList, ConcurrentHashMap }
import akka.actor._
import akka.remote.transport.AssociationHandle._
import akka.remote.transport.Transport._
import akka.util.ByteString
import com.typesafe.config.Config
import java.util.concurrent.{ CopyOnWriteArrayList, ConcurrentHashMap }
import TestTransport._
import scala.concurrent.duration._
import scala.concurrent.{ Await, Future, Promise }
import scala.concurrent.ExecutionContext.Implicits.global
/**
@ -38,8 +39,6 @@ class TestTransport(
conf.getString("scheme-identifier"))
}
import akka.remote.transport.TestTransport._
override def isResponsibleFor(address: Address): Boolean = true
private val associationListenerPromise = Promise[AssociationEventListener]()
@ -218,7 +217,7 @@ object TestTransport {
* The constant the future will be completed with.
*/
def pushConstant(c: B): Unit = push {
(x) ⇒ Future.successful(c)
_ ⇒ Future.successful(c)
}
/**
@ -228,7 +227,7 @@ object TestTransport {
* The throwable the failed future will contain.
*/
def pushError(e: Throwable): Unit = push {
(x) ⇒ Future.failed(e)
_ ⇒ Future.failed(e)
}
/**
@ -243,7 +242,7 @@ object TestTransport {
val originalBehavior = currentBehavior
push(
(params: A) ⇒ for (delayed ← controlPromise.future; original ← originalBehavior(params)) yield original)
(params: A) ⇒ controlPromise.future.flatMap(_ ⇒ originalBehavior(params)))
controlPromise
}

4
akka-remote/src/main/scala/akka/remote/transport/ThrottlerTransportAdapter.scala
View file

@ -496,7 +496,7 @@ private[transport] class ThrottledAssociation(
inboundThrottleMode = mode
sender() ! SetThrottleAck
stay()
case Event(Disassociated(info), _) ⇒
case Event(Disassociated(_), _) ⇒
stop() // not notifying the upstream handler is intentional: we are relying on heartbeating
case Event(FailWith(reason), _) ⇒
if (upstreamListener ne null) upstreamListener notify Disassociated(reason)
@ -513,7 +513,7 @@ private[transport] class ThrottledAssociation(
} catch {
// This layer should not care about malformed packets. Also, this also useful for testing, because
// arbitrary payload could be passed in
case NonFatal(e) ⇒ None
case NonFatal(_) ⇒ None
}
}

8
akka-remote/src/main/scala/akka/remote/transport/Transport.scala
View file

@ -5,12 +5,12 @@
package akka.remote.transport
import scala.concurrent.{ Future, Promise }
import scala.util.control.NoStackTrace
import akka.actor.{ ActorRef, Address, NoSerializationVerificationNeeded }
import akka.util.ByteString
import akka.util.{ ByteString, unused }
import akka.remote.transport.AssociationHandle.HandleEventListener
import akka.AkkaException
import scala.util.control.NoStackTrace
import akka.actor.DeadLetterSuppression
import akka.event.LoggingAdapter
@ -143,7 +143,7 @@ trait Transport {
* @param cmd Command message to the transport
* @return Future that succeeds when the command was handled or dropped
*/
def managementCommand(cmd: Any): Future[Boolean] = { Future.successful(false) }
def managementCommand(@unused cmd: Any): Future[Boolean] = { Future.successful(false) }
}

15
akka-remote/src/main/scala/akka/remote/transport/netty/NettyHelpers.scala
View file

@ -6,7 +6,10 @@ package akka.remote.transport.netty
import akka.AkkaException
import java.nio.channels.ClosedChannelException
import akka.util.unused
import org.jboss.netty.channel._
import scala.util.control.NonFatal
/**
@ -14,17 +17,17 @@ import scala.util.control.NonFatal
*/
private[netty] trait NettyHelpers {
protected def onConnect(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = ()
protected def onConnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
protected def onDisconnect(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = ()
protected def onDisconnect(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
protected def onOpen(ctx: ChannelHandlerContext, e: ChannelStateEvent): Unit = ()
protected def onOpen(@unused ctx: ChannelHandlerContext, @unused e: ChannelStateEvent): Unit = ()
protected def onMessage(ctx: ChannelHandlerContext, e: MessageEvent): Unit = ()
protected def onMessage(@unused ctx: ChannelHandlerContext, @unused e: MessageEvent): Unit = ()
protected def onException(ctx: ChannelHandlerContext, e: ExceptionEvent): Unit = ()
protected def onException(@unused ctx: ChannelHandlerContext, @unused e: ExceptionEvent): Unit = ()
final protected def transformException(ctx: ChannelHandlerContext, ev: ExceptionEvent): Unit = {
final protected def transformException(@unused ctx: ChannelHandlerContext, ev: ExceptionEvent): Unit = {
val cause = if (ev.getCause ne null) ev.getCause else new AkkaException("Unknown cause")
cause match {
case _: ClosedChannelException ⇒ // Ignore

2
akka-remote/src/main/scala/akka/remote/transport/netty/NettyTransport.scala
View file

@ -505,7 +505,7 @@ class NettyTransport(val settings: NettyTransportSettings, val system: ExtendedA
} catch {
case NonFatal(e) ⇒ {
log.error("failed to bind to {}, shutting down Netty transport", address)
try { shutdown() } catch { case NonFatal(e) ⇒ } // ignore possible exception during shutdown
try { shutdown() } catch { case NonFatal(_) ⇒ } // ignore possible exception during shutdown
throw e
}
}

7
akka-remote/src/main/scala/akka/remote/transport/netty/SSLEngineProvider.scala
View file

@ -20,8 +20,6 @@ import akka.event.Logging
import akka.event.MarkerLoggingAdapter
import akka.remote.RemoteTransportException
import akka.remote.artery.tcp.SecureRandomFactory
import akka.stream.IgnoreComplete
import akka.stream.TLSClosing
import akka.stream.TLSRole
import javax.net.ssl.KeyManager
import javax.net.ssl.KeyManagerFactory
@ -108,10 +106,7 @@ import javax.net.ssl.TrustManagerFactory
createSSLEngine(sslContext, role)
}
private def createSSLEngine(
sslContext: SSLContext,
role: TLSRole,
closing: TLSClosing = IgnoreComplete): SSLEngine = {
private def createSSLEngine(sslContext: SSLContext, role: TLSRole): SSLEngine = {
val engine = sslContext.createSSLEngine()

4
akka-remote/src/test/scala/akka/remote/transport/SwitchableLoggedBehaviorSpec.scala
View file

@ -85,7 +85,7 @@ class SwitchableLoggedBehaviorSpec extends AkkaSpec with DefaultTimeout {
Await.result(behavior(()), timeout.duration) should ===(3)
}
"enable delayed completition" in {
"enable delayed completion" in {
val behavior = defaultBehavior
val controlPromise = behavior.pushDelayed
val f = behavior(())
@ -96,7 +96,7 @@ class SwitchableLoggedBehaviorSpec extends AkkaSpec with DefaultTimeout {
awaitCond(f.isCompleted)
}
"log calls and parametrers" in {
"log calls and parameters" in {
val logPromise = Promise[Int]()
val behavior = new SwitchableLoggedBehavior[Int, Int]((i) ⇒ Future.successful(3), (i) ⇒ logPromise.success(i))

6
akka-testkit/src/main/scala/akka/testkit/ExplicitlyTriggeredScheduler.scala
View file

@ -8,8 +8,6 @@ import java.util.concurrent.ThreadFactory
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong
import com.typesafe.config.Config
import scala.annotation.tailrec
import scala.collection.JavaConverters._
import scala.concurrent.ExecutionContext
@ -19,6 +17,8 @@ import scala.util.Try
import akka.actor.Cancellable
import akka.actor.Scheduler
import akka.event.LoggingAdapter
import akka.util.unused
import com.typesafe.config.Config
/**
* For testing: scheduler that does not look at the clock, but must be
@ -30,7 +30,7 @@ import akka.event.LoggingAdapter
* easier, but these tests might fail to catch race conditions that only
* happen when tasks are scheduled in parallel in 'real time'.
*/
class ExplicitlyTriggeredScheduler(config: Config, log: LoggingAdapter, tf: ThreadFactory) extends Scheduler {
class ExplicitlyTriggeredScheduler(@unused config: Config, log: LoggingAdapter, @unused tf: ThreadFactory) extends Scheduler {
private case class Item(time: Long, interval: Option[FiniteDuration], runnable: Runnable)