This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch avoid-anonymous-inner-classes-use-lambda in repository https://gitbox.apache.org/repos/asf/pekko.git
commit d288c9c6bebec017fa5ebfc693dc0b40a23712b7 Author: 虎鸣 <[email protected]> AuthorDate: Sat Jul 25 21:35:31 2026 +0800 refactor: replace anonymous inner classes with lambdas Motivation: Now that Scala 2.12 support has been dropped, we can use SAM (Single Abstract Method) conversion to replace verbose anonymous inner classes with concise lambda expressions across the codebase. Modification: Replace anonymous inner classes with lambda expressions for SAM types including Runnable, Callable, Consumer, Comparator, Ordering, Predicate, BiFunction, JFunction, JPredicate, and LongFunction across actor, actor-typed, persistence, persistence-typed, remote, cluster, cluster-sharding, cluster-typed, coordination, and discovery modules. Remove unused imports that are no longer needed after the conversion. Result: Cleaner, more idiomatic Scala 2.13+ code with ~115 fewer lines while maintaining identical runtime behavior. Tests: - sbt actor/compile actor-typed/compile persistence/compile persistence-typed/compile remote/compile cluster/compile cluster-sharding/compile cluster-typed/compile cluster-sharding-typed/compile coordination/compile discovery/compile - sbt "actor-tests/testOnly org.apache.pekko.pattern.CircuitBreakerSpec" (63 passed) - sbt "actor-tests/testOnly org.apache.pekko.event.*" (88 passed) - sbt "persistence-typed-tests/testOnly org.apache.pekko.persistence.typed.javadsl.EventSourcedBehaviorJavaDslTest" (15 passed) References: Fixes #2201 --- .../pekko/actor/typed/javadsl/ReceiveBuilder.scala | 17 ++----- .../pekko/dispatch/BalancingDispatcher.scala | 7 ++- .../scala/org/apache/pekko/event/EventBus.scala | 17 +++---- actor/src/main/scala/org/apache/pekko/io/Dns.scala | 37 +++++++------- .../scala/org/apache/pekko/io/SimpleDnsCache.scala | 7 +-- .../org/apache/pekko/pattern/CircuitBreaker.scala | 20 ++------ .../pekko/pattern/FutureTimeoutSupport.scala | 22 ++++----- .../main/scala/org/apache/pekko/util/Helpers.scala | 12 ++--- .../typed/internal/ClusterShardingImpl.scala | 11 ++--- .../external/ExternalShardAllocation.scala | 6 +-- .../internal/AdaptedClusterSingletonImpl.scala | 5 +- .../scala/org/apache/pekko/cluster/Member.scala | 7 +-- .../lease/scaladsl/LeaseProvider.scala | 57 ++++++++++------------ .../org/apache/pekko/discovery/Discovery.scala | 5 +- .../persistence/typed/javadsl/CommandHandler.scala | 24 +++------ .../typed/javadsl/CommandHandlerWithReply.scala | 24 +++------ .../persistence/typed/javadsl/EventHandler.scala | 16 ++---- .../typed/state/javadsl/CommandHandler.scala | 24 +++------ .../state/javadsl/CommandHandlerWithReply.scala | 24 +++------ .../org/apache/pekko/persistence/Persistence.scala | 17 +++---- .../persistence/journal/japi/AsyncRecovery.scala | 6 +-- .../org/apache/pekko/remote/AckedDelivery.scala | 10 ++-- .../remote/artery/compress/CompressionTable.scala | 6 +-- .../artery/compress/InboundCompressions.scala | 10 ++-- 24 files changed, 138 insertions(+), 253 deletions(-) diff --git a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/ReceiveBuilder.scala b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/ReceiveBuilder.scala index fa706b65fb..2e594d566c 100644 --- a/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/ReceiveBuilder.scala +++ b/actor-typed/src/main/scala/org/apache/pekko/actor/typed/javadsl/ReceiveBuilder.scala @@ -88,13 +88,8 @@ final class ReceiveBuilder[T] private ( */ def onMessageEquals(msg: T, handler: Creator[Behavior[T]]): ReceiveBuilder[T] = withMessage(OptionVal.Some(msg.getClass), - OptionVal.Some(new JPredicate[T] { - override def test(param: T): Boolean = param == msg - }), - new JFunction[T, Behavior[T]] { - // invoke creator without the message - override def apply(param: T): Behavior[T] = handler.create() - }) + OptionVal.Some[JPredicate[T]]((param: T) => param == msg), + (_: T) => handler.create()) /** * Add a new case to the message handling matching any message. Subsequent `onMessage` clauses will @@ -141,12 +136,8 @@ final class ReceiveBuilder[T] private ( */ def onSignalEquals(signal: Signal, handler: Creator[Behavior[T]]): ReceiveBuilder[T] = withSignal(signal.getClass, - OptionVal.Some(new JPredicate[Signal] { - override def test(param: Signal): Boolean = param == signal - }), - new JFunction[Signal, Behavior[T]] { - override def apply(param: Signal): Behavior[T] = handler.create() - }) + OptionVal.Some[JPredicate[Signal]]((param: Signal) => param == signal), + (_: Signal) => handler.create()) private def withMessage[M <: T]( `type`: OptionVal[Class[M]], diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/BalancingDispatcher.scala b/actor/src/main/scala/org/apache/pekko/dispatch/BalancingDispatcher.scala index 79afca38ac..9be780ea68 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/BalancingDispatcher.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/BalancingDispatcher.scala @@ -13,7 +13,7 @@ package org.apache.pekko.dispatch -import java.util.{ Comparator, Iterator } +import java.util.Iterator import java.util.concurrent.ConcurrentSkipListSet import scala.annotation.tailrec @@ -64,9 +64,8 @@ private[pekko] class BalancingDispatcher( * INTERNAL API */ private[pekko] val team = - new ConcurrentSkipListSet[ActorCell](Helpers.identityHashComparator(new Comparator[ActorCell] { - def compare(l: ActorCell, r: ActorCell) = l.self.path.compareTo(r.self.path) - })) + new ConcurrentSkipListSet[ActorCell](Helpers.identityHashComparator((l: ActorCell, r: ActorCell) => + l.self.path.compareTo(r.self.path))) /** * INTERNAL API diff --git a/actor/src/main/scala/org/apache/pekko/event/EventBus.scala b/actor/src/main/scala/org/apache/pekko/event/EventBus.scala index 4e68fe1b81..d143366209 100644 --- a/actor/src/main/scala/org/apache/pekko/event/EventBus.scala +++ b/actor/src/main/scala/org/apache/pekko/event/EventBus.scala @@ -13,7 +13,6 @@ package org.apache.pekko.event -import java.util.Comparator import java.util.concurrent.ConcurrentSkipListSet import java.util.concurrent.atomic.AtomicReference @@ -93,9 +92,7 @@ trait PredicateClassifier { this: EventBus => trait LookupClassification { this: EventBus => protected final val subscribers = new Index[Classifier, Subscriber](mapSize(), - new Comparator[Subscriber] { - def compare(a: Subscriber, b: Subscriber): Int = compareSubscribers(a, b) - }) + (a: Subscriber, b: Subscriber) => compareSubscribers(a, b)) /** * This is a size hint for the number of Classifiers you expect to have (use powers of 2) @@ -218,13 +215,11 @@ trait SubchannelClassification { this: EventBus => */ trait ScanningClassification { self: EventBus => protected final val subscribers = - new ConcurrentSkipListSet[(Classifier, Subscriber)](new Comparator[(Classifier, Subscriber)] { - def compare(a: (Classifier, Subscriber), b: (Classifier, Subscriber)): Int = - compareClassifiers(a._1, b._1) match { - case 0 => compareSubscribers(a._2, b._2) - case other => other - } - }) + new ConcurrentSkipListSet[(Classifier, Subscriber)]((a: (Classifier, Subscriber), b: (Classifier, Subscriber)) => + compareClassifiers(a._1, b._1) match { + case 0 => compareSubscribers(a._2, b._2) + case other => other + }) /** * Provides a total ordering of Classifiers (think java.util.Comparator.compare) diff --git a/actor/src/main/scala/org/apache/pekko/io/Dns.scala b/actor/src/main/scala/org/apache/pekko/io/Dns.scala index 9421705247..b0a9a14a36 100644 --- a/actor/src/main/scala/org/apache/pekko/io/Dns.scala +++ b/actor/src/main/scala/org/apache/pekko/io/Dns.scala @@ -15,7 +15,6 @@ package org.apache.pekko.io import java.net.{ Inet4Address, Inet6Address, InetAddress } import java.util.concurrent.ConcurrentHashMap -import java.util.function.{ Function => JFunction } import scala.annotation.nowarn @@ -96,25 +95,23 @@ class DnsExt private[pekko] (val system: ExtendedActorSystem, resolverName: Stri // This can't pass in `this` as then AsyncDns would pick up the system settings asyncDns.computeIfAbsent( managerName, - new JFunction[String, ActorRef] { - override def apply(r: String): ActorRef = { - val settings = - new Settings(system.settings.config.getConfig("pekko.io.dns"), "async-dns") - val provider = system.dynamicAccess.createInstanceFor[DnsProvider](settings.ProviderObjectName, Nil).get - Logging(system, classOf[DnsExt]) - .info("Creating async dns resolver {} with manager name {}", settings.Resolver, managerName) - - system.systemActorOf( - props = Props( - provider.managerClass, - settings.Resolver, - system, - settings.ResolverConfig, - provider.cache, - settings.Dispatcher, - provider).withDeploy(Deploy.local).withDispatcher(settings.Dispatcher), - name = managerName) - } + (_: String) => { + val settings = + new Settings(system.settings.config.getConfig("pekko.io.dns"), "async-dns") + val provider = system.dynamicAccess.createInstanceFor[DnsProvider](settings.ProviderObjectName, Nil).get + Logging(system, classOf[DnsExt]) + .info("Creating async dns resolver {} with manager name {}", settings.Resolver, managerName) + + system.systemActorOf( + props = Props( + provider.managerClass, + settings.Resolver, + system, + settings.ResolverConfig, + provider.cache, + settings.Dispatcher, + provider).withDeploy(Deploy.local).withDispatcher(settings.Dispatcher), + name = managerName) }) } diff --git a/actor/src/main/scala/org/apache/pekko/io/SimpleDnsCache.scala b/actor/src/main/scala/org/apache/pekko/io/SimpleDnsCache.scala index 27e735ff2f..e635021504 100644 --- a/actor/src/main/scala/org/apache/pekko/io/SimpleDnsCache.scala +++ b/actor/src/main/scala/org/apache/pekko/io/SimpleDnsCache.scala @@ -136,9 +136,6 @@ object SimpleDnsCache { * INTERNAL API */ @InternalApi - private[io] def expiryEntryOrdering[K]() = new Ordering[ExpiryEntry[K]] { - override def compare(x: ExpiryEntry[K], y: ExpiryEntry[K]): Int = { - x.until.compareTo(y.until) - } - } + private[io] def expiryEntryOrdering[K](): Ordering[ExpiryEntry[K]] = + (x: ExpiryEntry[K], y: ExpiryEntry[K]) => x.until.compareTo(y.until) } diff --git a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala index 0e4cba4bcf..80fd434d95 100644 --- a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala +++ b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala @@ -363,9 +363,7 @@ class CircuitBreaker( * `scala.concurrent.TimeoutException` if the call timed out */ def callWithCircuitBreakerCS[T](body: Callable[CompletionStage[T]]): CompletionStage[T] = - callWithCircuitBreaker(new Callable[Future[T]] { - override def call(): Future[T] = body.call().asScala - }).asJava + callWithCircuitBreaker(() => body.call().asScala).asJava /** * Java API (8) for `withCircuitBreaker`. @@ -378,9 +376,7 @@ class CircuitBreaker( def callWithCircuitBreakerCS[T]( body: Callable[CompletionStage[T]], defineFailureFn: BiFunction[Optional[T], Optional[Throwable], java.lang.Boolean]): CompletionStage[T] = - callWithCircuitBreaker(new Callable[Future[T]] { - override def call(): Future[T] = body.call().asScala - }, defineFailureFn).asJava + callWithCircuitBreaker(() => body.call().asScala, defineFailureFn).asJava /** * Wraps invocations of synchronous calls that need to be protected. @@ -560,9 +556,7 @@ class CircuitBreaker( * @return CircuitBreaker for fluent usage */ def onCallSuccess(callback: Long => Unit): CircuitBreaker = - addOnCallSuccessListener(new Consumer[Long] { - def accept(result: Long): Unit = callback(result) - }) + addOnCallSuccessListener(result => callback(result)) /** * JavaAPI for onCallSuccess @@ -584,9 +578,7 @@ class CircuitBreaker( * @return CircuitBreaker for fluent usage */ def onCallFailure(callback: Long => Unit): CircuitBreaker = - addOnCallFailureListener(new Consumer[Long] { - def accept(result: Long): Unit = callback(result) - }) + addOnCallFailureListener(result => callback(result)) /** * JavaAPI for onCallFailure. @@ -608,9 +600,7 @@ class CircuitBreaker( * @return CircuitBreaker for fluent usage */ def onCallTimeout(callback: Long => Unit): CircuitBreaker = - addOnCallTimeoutListener(new Consumer[Long] { - def accept(result: Long): Unit = callback(result) - }) + addOnCallTimeoutListener(result => callback(result)) /** * JavaAPI for onCallTimeout. diff --git a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala index 3e696758f9..7106067f45 100644 --- a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala +++ b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala @@ -74,17 +74,17 @@ trait FutureTimeoutSupport { val p = new CompletableFuture[T] using.scheduleOnce( duration, - new Runnable { - override def run(): Unit = - try { - val future = value - future.handle[Unit]((t: T, ex: Throwable) => { - if (t != null) p.complete(t) - if (ex ne null) p.completeExceptionally(ex) - }) - } catch { - case NonFatal(ex) => p.completeExceptionally(ex) - } + () => { + try { + val future = value + future.handle[Unit]((t: T, ex: Throwable) => { + if (t != null) p.complete(t) + if (ex ne null) p.completeExceptionally(ex) + }) + } catch { + case NonFatal(ex) => p.completeExceptionally(ex) + } + () }) p } diff --git a/actor/src/main/scala/org/apache/pekko/util/Helpers.scala b/actor/src/main/scala/org/apache/pekko/util/Helpers.scala index 0f2e82b681..1661dbaadd 100644 --- a/actor/src/main/scala/org/apache/pekko/util/Helpers.scala +++ b/actor/src/main/scala/org/apache/pekko/util/Helpers.scala @@ -63,12 +63,12 @@ object Helpers { * consistent with equals, otherwise it would not be an enhancement over * the identityHashCode. */ - def identityHashComparator[T <: AnyRef](comp: Comparator[T]): Comparator[T] = new Comparator[T] { - def compare(a: T, b: T): Int = compareIdentityHash(a, b) match { - case 0 if a != b => comp.compare(a, b) - case x => x - } - } + def identityHashComparator[T <: AnyRef](comp: Comparator[T]): Comparator[T] = + (a: T, b: T) => + compareIdentityHash(a, b) match { + case 0 if a != b => comp.compare(a, b) + case x => x + } /** * Converts a "currentTimeMillis"-obtained timestamp accordingly: diff --git a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala index 1794da8b86..9031d7aaf3 100644 --- a/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala +++ b/cluster-sharding-typed/src/main/scala/org/apache/pekko/cluster/sharding/typed/internal/ClusterShardingImpl.scala @@ -193,13 +193,10 @@ import pekko.util.Timeout val shardCommandDelegator: ActorRef[scaladsl.ClusterSharding.ShardCommand] = shardCommandActors.computeIfAbsent( typeKey.name, - new java.util.function.Function[String, ActorRef[scaladsl.ClusterSharding.ShardCommand]] { - override def apply(t: String): ActorRef[scaladsl.ClusterSharding.ShardCommand] = { - system.systemActorOf( - ShardCommandActor.behavior(stopMessage.getOrElse(PoisonPill)), - URLEncoder.encode(typeKey.name, StandardCharsets.UTF_8) + "ShardCommandDelegator") - } - }) + (_: String) => + system.systemActorOf( + ShardCommandActor.behavior(stopMessage.getOrElse(PoisonPill)), + URLEncoder.encode(typeKey.name, StandardCharsets.UTF_8) + "ShardCommandDelegator")) def poisonPillInterceptor(behv: Behavior[M]): Behavior[M] = { stopMessage match { diff --git a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/external/ExternalShardAllocation.scala b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/external/ExternalShardAllocation.scala index 68c6ec8634..117d480265 100644 --- a/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/external/ExternalShardAllocation.scala +++ b/cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/external/ExternalShardAllocation.scala @@ -33,10 +33,8 @@ final class ExternalShardAllocation(system: ExtendedActorSystem) extends Extensi private val clients = new ConcurrentHashMap[String, ExternalShardAllocationClientImpl] - private val factory = new JFunction[String, ExternalShardAllocationClientImpl] { - override def apply(typeName: String): ExternalShardAllocationClientImpl = - new ExternalShardAllocationClientImpl(system, typeName) - } + private val factory: JFunction[String, ExternalShardAllocationClientImpl] = + (typeName: String) => new ExternalShardAllocationClientImpl(system, typeName) /** * Scala API diff --git a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/AdaptedClusterSingletonImpl.scala b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/AdaptedClusterSingletonImpl.scala index 032400e67d..464b3965c0 100644 --- a/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/AdaptedClusterSingletonImpl.scala +++ b/cluster-typed/src/main/scala/org/apache/pekko/cluster/typed/internal/AdaptedClusterSingletonImpl.scala @@ -79,8 +79,8 @@ private[pekko] final class AdaptedClusterSingletonImpl(system: ActorSystem[?]) e } private def getProxy[T](name: String, settings: ClusterSingletonSettings): ActorRef[T] = { - val proxyCreator = new JFunction[(String, Option[DataCenter]), ActorRef[?]] { - def apply(singletonNameAndDc: (String, Option[DataCenter])): ActorRef[?] = { + val proxyCreator: JFunction[(String, Option[DataCenter]), ActorRef[?]] = + (singletonNameAndDc: (String, Option[DataCenter])) => { val (singletonName, _) = singletonNameAndDc val proxyName = s"singletonProxy$singletonName-${settings.dataCenter.getOrElse("no-dc")}" classicSystem.systemActorOf( @@ -88,7 +88,6 @@ private[pekko] final class AdaptedClusterSingletonImpl(system: ActorSystem[?]) e .props(s"/system/${managerNameFor(singletonName)}", settings.toProxySettings(singletonName)), proxyName) } - } proxies.computeIfAbsent((name, settings.dataCenter), proxyCreator).asInstanceOf[ActorRef[T]] } } diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/Member.scala b/cluster/src/main/scala/org/apache/pekko/cluster/Member.scala index a741f89fbe..b4a1649dd8 100644 --- a/cluster/src/main/scala/org/apache/pekko/cluster/Member.scala +++ b/cluster/src/main/scala/org/apache/pekko/cluster/Member.scala @@ -160,11 +160,8 @@ object Member { /** * `Member` ordering type class, sorts members by host and port. */ - implicit val ordering: Ordering[Member] = new Ordering[Member] { - def compare(a: Member, b: Member): Int = { - a.uniqueAddress.compare(b.uniqueAddress) - } - } + implicit val ordering: Ordering[Member] = + (a: Member, b: Member) => a.uniqueAddress.compare(b.uniqueAddress) /** * Sort members by age, i.e. using [[Member#isOlderThan]]. diff --git a/coordination/src/main/scala/org/apache/pekko/coordination/lease/scaladsl/LeaseProvider.scala b/coordination/src/main/scala/org/apache/pekko/coordination/lease/scaladsl/LeaseProvider.scala index ddf770a7fb..278191029d 100644 --- a/coordination/src/main/scala/org/apache/pekko/coordination/lease/scaladsl/LeaseProvider.scala +++ b/coordination/src/main/scala/org/apache/pekko/coordination/lease/scaladsl/LeaseProvider.scala @@ -14,7 +14,6 @@ package org.apache.pekko.coordination.lease.scaladsl import java.util.concurrent.ConcurrentHashMap -import java.util.function.{ Function => JFunction } import scala.collection.immutable import scala.reflect.ClassTag @@ -66,36 +65,34 @@ final class LeaseProvider(system: ExtendedActorSystem) extends Extension { val leaseKey = LeaseKey(leaseName, configPath, ownerName) leases.computeIfAbsent( leaseKey, - new JFunction[LeaseKey, Lease] { - override def apply(t: LeaseKey): Lease = { - val leaseConfig = system.settings.config - .getConfig(configPath) - .withFallback(system.settings.config.getConfig("pekko.coordination.lease")) - - val settings = LeaseSettings(leaseConfig, leaseName, ownerName) - - // Try and load a scala implementation - val lease: Try[Lease] = - loadLease[Lease](settings).recoverWith { - case _: ClassCastException => - // Try and load a java implementation - loadLease[pekko.coordination.lease.javadsl.Lease](settings).map(javaLease => - new LeaseAdapterToScala(javaLease)(system.dispatchers.internalDispatcher)) - } - - lease match { - case Success(value) => value - case Failure(e) => - log.error( - e, - "Invalid lease configuration for leaseName [{}], configPath [{}] lease-class [{}]. " + - "The class must implement scaladsl.Lease or javadsl.Lease and have constructor with LeaseSettings parameter and " + - "optionally ActorSystem parameter.", - settings.leaseName, - configPath, - settings.leaseConfig.getString("lease-class")) - throw e + (_: LeaseKey) => { + val leaseConfig = system.settings.config + .getConfig(configPath) + .withFallback(system.settings.config.getConfig("pekko.coordination.lease")) + + val settings = LeaseSettings(leaseConfig, leaseName, ownerName) + + // Try and load a scala implementation + val lease: Try[Lease] = + loadLease[Lease](settings).recoverWith { + case _: ClassCastException => + // Try and load a java implementation + loadLease[pekko.coordination.lease.javadsl.Lease](settings).map(javaLease => + new LeaseAdapterToScala(javaLease)(system.dispatchers.internalDispatcher)) } + + lease match { + case Success(value) => value + case Failure(e) => + log.error( + e, + "Invalid lease configuration for leaseName [{}], configPath [{}] lease-class [{}]. " + + "The class must implement scaladsl.Lease or javadsl.Lease and have constructor with LeaseSettings parameter and " + + "optionally ActorSystem parameter.", + settings.leaseName, + configPath, + settings.leaseConfig.getString("lease-class")) + throw e } }) } diff --git a/discovery/src/main/scala/org/apache/pekko/discovery/Discovery.scala b/discovery/src/main/scala/org/apache/pekko/discovery/Discovery.scala index bbe48f59b3..c0d1b7935d 100644 --- a/discovery/src/main/scala/org/apache/pekko/discovery/Discovery.scala +++ b/discovery/src/main/scala/org/apache/pekko/discovery/Discovery.scala @@ -25,9 +25,8 @@ import pekko.annotation.InternalApi final class Discovery(implicit system: ExtendedActorSystem) extends Extension { private val implementations = new ConcurrentHashMap[String, ServiceDiscovery] - private val factory = new JFunction[String, ServiceDiscovery] { - override def apply(method: String): ServiceDiscovery = createServiceDiscovery(method) - } + private val factory: JFunction[String, ServiceDiscovery] = + (method: String) => createServiceDiscovery(method) private lazy val _defaultImplMethod = system.settings.config.getString("pekko.discovery.method") match { diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandler.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandler.scala index 457cc76efd..8725e8280a 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandler.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandler.scala @@ -166,9 +166,7 @@ final class CommandHandlerBuilder[Command, Event, State]() { object CommandHandlerBuilderByState { - private val _trueStatePredicate: Predicate[Any] = new Predicate[Any] { - override def test(t: Any): Boolean = true - } + private val _trueStatePredicate: Predicate[Any] = (_: Any) => true private def trueStatePredicate[S]: Predicate[S] = _trueStatePredicate.asInstanceOf[Predicate[S]] @@ -244,9 +242,7 @@ final class CommandHandlerBuilderByState[Command, Event, S <: State, State] @Int predicate: Predicate[Command], handler: JFunction[Command, Effect[Event, State]]): CommandHandlerBuilderByState[Command, Event, S, State] = { addCase(cmd => predicate.test(cmd), - new BiFunction[S, Command, Effect[Event, State]] { - override def apply(state: S, cmd: Command): Effect[Event, State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) this } @@ -280,9 +276,7 @@ final class CommandHandlerBuilderByState[Command, Event, S <: State, State] @Int commandClass: Class[C], handler: JFunction[C, Effect[Event, State]]): CommandHandlerBuilderByState[Command, Event, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, Effect[Event, State]] { - override def apply(state: S, cmd: C): Effect[Event, State] = handler(cmd) - }) + (_: S, cmd: C) => handler(cmd)) } /** @@ -298,9 +292,7 @@ final class CommandHandlerBuilderByState[Command, Event, S <: State, State] @Int commandClass: Class[C], handler: Supplier[Effect[Event, State]]): CommandHandlerBuilderByState[Command, Event, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, Effect[Event, State]] { - override def apply(state: S, cmd: C): Effect[Event, State] = handler.get() - }) + (_: S, _: C) => handler.get()) } /** @@ -342,9 +334,7 @@ final class CommandHandlerBuilderByState[Command, Event, S <: State, State] @Int */ def onAnyCommand(handler: JFunction[Command, Effect[Event, State]]): CommandHandler[Command, Event, State] = { addCase(_ => true, - new BiFunction[S, Command, Effect[Event, State]] { - override def apply(state: S, cmd: Command): Effect[Event, State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) build() } @@ -367,9 +357,7 @@ final class CommandHandlerBuilderByState[Command, Event, S <: State, State] @Int */ def onAnyCommand(handler: Supplier[Effect[Event, State]]): CommandHandler[Command, Event, State] = { addCase(_ => true, - new BiFunction[S, Command, Effect[Event, State]] { - override def apply(state: S, cmd: Command): Effect[Event, State] = handler.get() - }) + (_: S, _: Command) => handler.get()) build() } diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandlerWithReply.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandlerWithReply.scala index 7d3274ea92..1a44d417e0 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandlerWithReply.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/CommandHandlerWithReply.scala @@ -176,9 +176,7 @@ final class CommandHandlerWithReplyBuilder[Command, Event, State]() { object CommandHandlerWithReplyBuilderByState { - private val _trueStatePredicate: Predicate[Any] = new Predicate[Any] { - override def test(t: Any): Boolean = true - } + private val _trueStatePredicate: Predicate[Any] = (_: Any) => true private def trueStatePredicate[S]: Predicate[S] = _trueStatePredicate.asInstanceOf[Predicate[S]] @@ -255,9 +253,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, Event, S <: State, St : CommandHandlerWithReplyBuilderByState[Command, Event, S, State] = { addCase( cmd => predicate.test(cmd), - new BiFunction[S, Command, ReplyEffect[Event, State]] { - override def apply(state: S, cmd: Command): ReplyEffect[Event, State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) this } @@ -289,9 +285,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, Event, S <: State, St def onCommand[C <: Command](commandClass: Class[C], handler: JFunction[C, ReplyEffect[Event, State]]) : CommandHandlerWithReplyBuilderByState[Command, Event, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, ReplyEffect[Event, State]] { - override def apply(state: S, cmd: C): ReplyEffect[Event, State] = handler(cmd) - }) + (_: S, cmd: C) => handler(cmd)) } /** @@ -307,9 +301,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, Event, S <: State, St commandClass: Class[C], handler: Supplier[ReplyEffect[Event, State]]): CommandHandlerWithReplyBuilderByState[Command, Event, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, ReplyEffect[Event, State]] { - override def apply(state: S, cmd: C): ReplyEffect[Event, State] = handler.get() - }) + (_: S, _: C) => handler.get()) } /** @@ -353,9 +345,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, Event, S <: State, St def onAnyCommand( handler: JFunction[Command, ReplyEffect[Event, State]]): CommandHandlerWithReply[Command, Event, State] = { addCase(_ => true, - new BiFunction[S, Command, ReplyEffect[Event, State]] { - override def apply(state: S, cmd: Command): ReplyEffect[Event, State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) build() } @@ -378,9 +368,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, Event, S <: State, St */ def onAnyCommand(handler: Supplier[ReplyEffect[Event, State]]): CommandHandlerWithReply[Command, Event, State] = { addCase(_ => true, - new BiFunction[S, Command, ReplyEffect[Event, State]] { - override def apply(state: S, cmd: Command): ReplyEffect[Event, State] = handler.get() - }) + (_: S, _: Command) => handler.get()) build() } diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/EventHandler.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/EventHandler.scala index 271fae6468..d1993616ec 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/EventHandler.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/javadsl/EventHandler.scala @@ -167,9 +167,7 @@ final class EventHandlerBuilder[State, Event]() { object EventHandlerBuilderByState { - private val _trueStatePredicate: Predicate[Any] = new Predicate[Any] { - override def test(t: Any): Boolean = true - } + private val _trueStatePredicate: Predicate[Any] = (_: Any) => true private def trueStatePredicate[S]: Predicate[S] = _trueStatePredicate.asInstanceOf[Predicate[S]] @@ -242,9 +240,7 @@ final class EventHandlerBuilderByState[S <: State, State, Event]( eventClass: Class[E], handler: JFunction[E, State]): EventHandlerBuilderByState[S, State, Event] = { onEvent[E](eventClass, - new BiFunction[S, E, State] { - override def apply(state: S, event: E): State = handler(event) - }) + (_: S, event: E) => handler(event)) } /** @@ -260,9 +256,7 @@ final class EventHandlerBuilderByState[S <: State, State, Event]( eventClass: Class[E], handler: Supplier[State]): EventHandlerBuilderByState[S, State, Event] = { - val supplierBiFunction = new BiFunction[S, E, State] { - def apply(t: S, u: E): State = handler.get() - } + val supplierBiFunction: BiFunction[S, E, State] = (_: S, _: E) => handler.get() onEvent(eventClass, supplierBiFunction) } @@ -300,9 +294,7 @@ final class EventHandlerBuilderByState[S <: State, State, Event]( * @return An EventHandler from the appended states. */ def onAnyEvent(handler: JFunction[Event, State]): EventHandler[State, Event] = { - onAnyEvent(new BiFunction[State, Event, State] { - override def apply(state: State, event: Event): State = handler(event) - }) + onAnyEvent((_: State, event: Event) => handler(event)) build() } diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandler.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandler.scala index 571330bfb6..49361a6d29 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandler.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandler.scala @@ -169,9 +169,7 @@ final class CommandHandlerBuilder[Command, State]() { object CommandHandlerBuilderByState { - private val _trueStatePredicate: Predicate[Any] = new Predicate[Any] { - override def test(t: Any): Boolean = true - } + private val _trueStatePredicate: Predicate[Any] = (_: Any) => true private def trueStatePredicate[S]: Predicate[S] = _trueStatePredicate.asInstanceOf[Predicate[S]] @@ -245,9 +243,7 @@ final class CommandHandlerBuilderByState[Command, S <: State, State] @InternalAp predicate: Predicate[Command], handler: JFunction[Command, Effect[State]]): CommandHandlerBuilderByState[Command, S, State] = { addCase(cmd => predicate.test(cmd), - new BiFunction[S, Command, Effect[State]] { - override def apply(state: S, cmd: Command): Effect[State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) this } @@ -281,9 +277,7 @@ final class CommandHandlerBuilderByState[Command, S <: State, State] @InternalAp commandClass: Class[C], handler: JFunction[C, Effect[State]]): CommandHandlerBuilderByState[Command, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, Effect[State]] { - override def apply(state: S, cmd: C): Effect[State] = handler(cmd) - }) + (_: S, cmd: C) => handler(cmd)) } /** @@ -299,9 +293,7 @@ final class CommandHandlerBuilderByState[Command, S <: State, State] @InternalAp commandClass: Class[C], handler: Supplier[Effect[State]]): CommandHandlerBuilderByState[Command, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, Effect[State]] { - override def apply(state: S, cmd: C): Effect[State] = handler.get() - }) + (_: S, _: C) => handler.get()) } /** @@ -343,9 +335,7 @@ final class CommandHandlerBuilderByState[Command, S <: State, State] @InternalAp */ def onAnyCommand(handler: JFunction[Command, Effect[State]]): CommandHandler[Command, State] = { addCase(_ => true, - new BiFunction[S, Command, Effect[State]] { - override def apply(state: S, cmd: Command): Effect[State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) build() } @@ -368,9 +358,7 @@ final class CommandHandlerBuilderByState[Command, S <: State, State] @InternalAp */ def onAnyCommand(handler: Supplier[Effect[State]]): CommandHandler[Command, State] = { addCase(_ => true, - new BiFunction[S, Command, Effect[State]] { - override def apply(state: S, cmd: Command): Effect[State] = handler.get() - }) + (_: S, _: Command) => handler.get()) build() } diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandlerWithReply.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandlerWithReply.scala index 5907be2b4c..3bb0519d91 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandlerWithReply.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/state/javadsl/CommandHandlerWithReply.scala @@ -177,9 +177,7 @@ final class CommandHandlerWithReplyBuilder[Command, State]() { object CommandHandlerWithReplyBuilderByState { - private val _trueStatePredicate: Predicate[Any] = new Predicate[Any] { - override def test(t: Any): Boolean = true - } + private val _trueStatePredicate: Predicate[Any] = (_: Any) => true private def trueStatePredicate[S]: Predicate[S] = _trueStatePredicate.asInstanceOf[Predicate[S]] @@ -255,9 +253,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, S <: State, State] @I predicate: Predicate[Command], handler: JFunction[Command, ReplyEffect[State]]): CommandHandlerWithReplyBuilderByState[Command, S, State] = { addCase(cmd => predicate.test(cmd), - new BiFunction[S, Command, ReplyEffect[State]] { - override def apply(state: S, cmd: Command): ReplyEffect[State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) this } @@ -291,9 +287,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, S <: State, State] @I commandClass: Class[C], handler: JFunction[C, ReplyEffect[State]]): CommandHandlerWithReplyBuilderByState[Command, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, ReplyEffect[State]] { - override def apply(state: S, cmd: C): ReplyEffect[State] = handler(cmd) - }) + (_: S, cmd: C) => handler(cmd)) } /** @@ -309,9 +303,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, S <: State, State] @I commandClass: Class[C], handler: Supplier[ReplyEffect[State]]): CommandHandlerWithReplyBuilderByState[Command, S, State] = { onCommand[C](commandClass, - new BiFunction[S, C, ReplyEffect[State]] { - override def apply(state: S, cmd: C): ReplyEffect[State] = handler.get() - }) + (_: S, _: C) => handler.get()) } /** @@ -353,9 +345,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, S <: State, State] @I */ def onAnyCommand(handler: JFunction[Command, ReplyEffect[State]]): CommandHandlerWithReply[Command, State] = { addCase(_ => true, - new BiFunction[S, Command, ReplyEffect[State]] { - override def apply(state: S, cmd: Command): ReplyEffect[State] = handler(cmd) - }) + (_: S, cmd: Command) => handler(cmd)) build() } @@ -378,9 +368,7 @@ final class CommandHandlerWithReplyBuilderByState[Command, S <: State, State] @I */ def onAnyCommand(handler: Supplier[ReplyEffect[State]]): CommandHandlerWithReply[Command, State] = { addCase(_ => true, - new BiFunction[S, Command, ReplyEffect[State]] { - override def apply(state: S, cmd: Command): ReplyEffect[State] = handler.get() - }) + (_: S, _: Command) => handler.get()) build() } diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/Persistence.scala b/persistence/src/main/scala/org/apache/pekko/persistence/Persistence.scala index 66dacddf0e..c63a4caec3 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/Persistence.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/Persistence.scala @@ -14,7 +14,6 @@ package org.apache.pekko.persistence import java.util.concurrent.atomic.AtomicReference -import java.util.function.Consumer import scala.annotation.tailrec import scala.collection.immutable @@ -268,19 +267,15 @@ class Persistence(val system: ExtendedActorSystem) extends Extension { config .getStringList("journal.auto-start-journals") - .forEach(new Consumer[String] { - override def accept(id: String): Unit = { - log.info("Auto-starting journal plugin `{}`", id) - journalFor(id) - } + .forEach((id: String) => { + log.info("Auto-starting journal plugin `{}`", id) + journalFor(id) }) config .getStringList("snapshot-store.auto-start-snapshot-stores") - .forEach(new Consumer[String] { - override def accept(id: String): Unit = { - log.info("Auto-starting snapshot store `{}`", id) - snapshotStoreFor(id) - } + .forEach((id: String) => { + log.info("Auto-starting snapshot store `{}`", id) + snapshotStoreFor(id) }) /** diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/journal/japi/AsyncRecovery.scala b/persistence/src/main/scala/org/apache/pekko/persistence/journal/japi/AsyncRecovery.scala index 38375b399d..9eb79fbe48 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/journal/japi/AsyncRecovery.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/journal/japi/AsyncRecovery.scala @@ -13,8 +13,6 @@ package org.apache.pekko.persistence.journal.japi -import java.util.function.Consumer - import scala.concurrent.ExecutionContext import scala.concurrent.Future import scala.jdk.FutureConverters._ @@ -33,9 +31,7 @@ abstract class AsyncRecovery extends SAsyncReplay with AsyncRecoveryPlugin { thi final def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)( replayCallback: (PersistentRepr) => Unit) = doAsyncReplayMessages(persistenceId, fromSequenceNr, toSequenceNr, max, - new Consumer[PersistentRepr] { - def accept(p: PersistentRepr) = replayCallback(p) - }).asScala.map(scalaAnyToUnit)(ExecutionContext.parasitic) + (p: PersistentRepr) => replayCallback(p)).asScala.map(scalaAnyToUnit)(ExecutionContext.parasitic) final def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = doAsyncReadHighestSequenceNr(persistenceId, fromSequenceNr: Long) diff --git a/remote/src/main/scala/org/apache/pekko/remote/AckedDelivery.scala b/remote/src/main/scala/org/apache/pekko/remote/AckedDelivery.scala index 16fc083272..6c5f18877a 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/AckedDelivery.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/AckedDelivery.scala @@ -21,12 +21,11 @@ import pekko.PekkoException @deprecated("Classic remoting is deprecated, use Artery", "Akka 2.6.0") object SeqNo { - implicit val ord: Ordering[SeqNo] = new Ordering[SeqNo] { - override def compare(x: SeqNo, y: SeqNo): Int = { + implicit val ord: Ordering[SeqNo] = + (x: SeqNo, y: SeqNo) => { val sgn = if (x.rawValue < y.rawValue) -1 else if (x.rawValue > y.rawValue) 1 else 0 if (((x.rawValue - y.rawValue) * sgn) < 0L) -sgn else sgn } - } } @@ -57,9 +56,8 @@ final case class SeqNo(rawValue: Long) extends Ordered[SeqNo] { @deprecated("Classic remoting is deprecated, use Artery", "Akka 2.6.0") object HasSequenceNumber { - implicit def seqOrdering[T <: HasSequenceNumber]: Ordering[T] = new Ordering[T] { - def compare(x: T, y: T) = x.seq.compare(y.seq) - } + implicit def seqOrdering[T <: HasSequenceNumber]: Ordering[T] = + (x: T, y: T) => x.seq.compare(y.seq) } /** diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala index 56a1aec8aa..5ecdc2c346 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/CompressionTable.scala @@ -101,10 +101,8 @@ private[remote] final class CompressionTable[T]( private[remote] object CompressionTable { final val NotCompressedId = -1 - final val CompareBy2ndValue: Comparator[(Object, Int)] = new Comparator[(Object, Int)] { - override def compare(o1: (Object, Int), o2: (Object, Int)): Int = - o1._2.compare(o2._2) - } + final val CompareBy2ndValue: Comparator[(Object, Int)] = + (o1: (Object, Int), o2: (Object, Int)) => o1._2.compare(o2._2) def compareBy2ndValue[T]: Comparator[Tuple2[T, Int]] = CompareBy2ndValue.asInstanceOf[Comparator[(T, Int)]] private def newObject2IntHashMap[T](initialCapacity: Int): Object2IntHashMap[T] = { diff --git a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala index a97c28c08e..b05511509a 100644 --- a/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala +++ b/remote/src/main/scala/org/apache/pekko/remote/artery/compress/InboundCompressions.scala @@ -121,24 +121,22 @@ private[remote] final class InboundCompressionsImpl( private val _actorRefsIns = new Long2ObjectHashMap[InboundActorRefCompression]() private val _inboundActorRefsLog = Logging(system, classOf[InboundActorRefCompression]) - private val createInboundActorRefsForOrigin = new LongFunction[InboundActorRefCompression] { - override def apply(originUid: Long): InboundActorRefCompression = { + private val createInboundActorRefsForOrigin: LongFunction[InboundActorRefCompression] = + (originUid: Long) => { val actorRefHitters = new TopHeavyHitters[ActorRef](settings.ActorRefs.Max) new InboundActorRefCompression(_inboundActorRefsLog, settings, originUid, inboundContext, actorRefHitters) } - } private def actorRefsIn(originUid: Long): InboundActorRefCompression = _actorRefsIns.computeIfAbsent(originUid, createInboundActorRefsForOrigin) private val _classManifestsIns = new Long2ObjectHashMap[InboundManifestCompression]() private val _inboundManifestLog = Logging(system, classOf[InboundManifestCompression]) - private val createInboundManifestsForOrigin = new LongFunction[InboundManifestCompression] { - override def apply(originUid: Long): InboundManifestCompression = { + private val createInboundManifestsForOrigin: LongFunction[InboundManifestCompression] = + (originUid: Long) => { val manifestHitters = new TopHeavyHitters[String](settings.Manifests.Max) new InboundManifestCompression(_inboundManifestLog, settings, originUid, inboundContext, manifestHitters) } - } private def classManifestsIn(originUid: Long): InboundManifestCompression = _classManifestsIns.computeIfAbsent(originUid, createInboundManifestsForOrigin) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
