use lambdas instead of new Runnable

actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala

@@ -130,12 +130,10 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val finished = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            release.await()
-            finished.countDown()
-          }
+        executor.execute(() => {
+          started.countDown()
+          release.await()
+          finished.countDown()
         })
 
         started.await(5, TimeUnit.SECONDS) should ===(true)
@@ -168,15 +166,14 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val interrupted = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            try release.await()
-            catch {
-              case _: InterruptedException => interrupted.countDown()
-            }
+        executor.execute(() => {
+          started.countDown()
+          try release.await()
+          catch {
+            case _: InterruptedException => interrupted.countDown()
           }
-        })
+        }
+        )
 
         started.await(5, TimeUnit.SECONDS) should ===(true)
         executor.shutdownNow()
@@ -205,13 +202,11 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val interrupted = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            try release.await()
-            catch {
-              case _: InterruptedException => interrupted.countDown()
-            }
+        executor.execute(() => {
+          started.countDown()
+          try release.await()
+          catch {
+            case _: InterruptedException => interrupted.countDown()
           }
         })
 

actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala

@@ -107,20 +107,14 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
         val completed = new AtomicInteger(0)
         var innerBatch: Runnable = null
 
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            executor.execute(new Runnable {
-              override def run(): Unit = completed.incrementAndGet()
-            })
-            // Model a pool worker executing another submitted batch while helping a join.
-            innerBatch.run()
-            completed.incrementAndGet()
-          }
+        executor.execute(() => {
+          executor.execute(() => completed.incrementAndGet())
+          // Model a pool worker executing another submitted batch while helping a join.
+          innerBatch.run()
+          completed.incrementAndGet()
         })
         val outerBatch = submitted.remove()
-        executor.execute(new Runnable {
-          override def run(): Unit = completed.incrementAndGet()
-        })
+        executor.execute(() => completed.incrementAndGet())
         innerBatch = submitted.remove()
 
         outerBatch.run()
@@ -196,14 +190,11 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
     "work with same-thread executor plus blocking" in {
       val ec = scala.concurrent.ExecutionContext.parasitic
       var x = 0
-      ec.execute(new Runnable {
-        override def run = {
-          ec.execute(new Runnable {
-            override def run = blocking {
-              x = 1
-            }
+      ec.execute(() => {
+        ec.execute(() =>
+          blocking {
+            x = 1
           })
-        }
       })
       x should be(1)
     }
@@ -255,7 +246,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
     "be suspendable and resumable" in {
       val sec = SerializedSuspendableExecutionContext(1)(ExecutionContext.global)
       val counter = new AtomicInteger(0)
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
       perform(_ + 1)
       perform(x => { sec.suspend(); x * 2 })
       awaitCond(counter.get == 2)
@@ -282,7 +273,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val throughput = 25
       val sec = SerializedSuspendableExecutionContext(throughput)(underlying)
       sec.suspend()
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
 
       val total = 1000
       (1 to total).foreach { _ =>
@@ -299,7 +290,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val sec = SerializedSuspendableExecutionContext(1)(ExecutionContext.global)
       val total = 10000
       val counter = new AtomicInteger(0)
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
 
       (1 to total).foreach { i =>
         perform(c => if (c == (i - 1)) c + 1 else c)
@@ -318,7 +309,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val throughput = 25
       val sec = SerializedSuspendableExecutionContext(throughput)(underlying)
       sec.suspend()
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
       perform(_ + 1)
       (1 to 10).foreach { _ =>
         perform(identity)

actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala

@@ -156,9 +156,7 @@ class ActorSystemSpec
       withSystem("thread", Behaviors.empty[String]) { sys =>
         val p = Promise[Int]()
         sys.threadFactory
-          .newThread(new Runnable {
-            def run(): Unit = p.success(42)
-          })
+          .newThread(() => p.success(42))
           .start()
         p.future.futureValue should ===(42)
       }

actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala

@@ -1102,7 +1102,7 @@ private[pekko] class ActorSystemImpl(
     }
 
   def start(): this.type = _start
-  def registerOnTermination[T](code: => T): Unit = { registerOnTermination(new Runnable { def run() = code }) }
+  def registerOnTermination[T](code: => T): Unit = { registerOnTermination((() => code): Runnable) }
   def registerOnTermination(code: Runnable): Unit = { terminationCallbacks.add(code) }
 
   @volatile private var terminating = false

actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala

@@ -417,8 +417,8 @@ object LightArrayRevolverScheduler {
     override def isCancelled: Boolean = task eq CancelledTask
   }
 
-  private val CancelledTask = new Runnable { def run = () }
-  private val ExecutedTask = new Runnable { def run = () }
+  private val CancelledTask: Runnable = () => ()
+  private val ExecutedTask: Runnable = () => ()
 
   private val NotCancellable: TimerTask = new TimerTask {
     def cancel(): Boolean = false

actor/src/main/scala/org/apache/pekko/actor/Scheduler.scala

@@ -156,12 +156,10 @@ trait Scheduler {
       implicit
       executor: ExecutionContext,
       sender: ActorRef = Actor.noSender): Cancellable = {
-    scheduleWithFixedDelay(initialDelay, delay)(new Runnable {
-      def run(): Unit = {
-        receiver ! message
-        if (receiver.isTerminated)
-          throw SchedulerException("timer active for terminated actor")
-      }
+    scheduleWithFixedDelay(initialDelay, delay)(() => {
+      receiver ! message
+      if (receiver.isTerminated)
+        throw SchedulerException("timer active for terminated actor")
     })
   }
 
@@ -301,12 +299,10 @@ trait Scheduler {
     schedule(
       initialDelay,
       interval,
-      new Runnable {
-        def run(): Unit = {
-          receiver ! message
-          if (receiver.isTerminated)
-            throw SchedulerException("timer active for terminated actor")
-        }
+      () => {
+        receiver ! message
+        if (receiver.isTerminated)
+          throw SchedulerException("timer active for terminated actor")
       })
 
   /**
@@ -360,10 +356,7 @@ trait Scheduler {
       implicit
       executor: ExecutionContext,
       sender: ActorRef = Actor.noSender): Cancellable =
-    scheduleOnce(delay,
-      new Runnable {
-        override def run(): Unit = receiver ! message
-      })
+    scheduleOnce(delay, () => receiver ! message)
 
   /**
    * Java API: Schedules a message to be sent once with a delay, i.e. a time period that has
@@ -396,7 +389,7 @@ trait Scheduler {
   final def scheduleOnce(delay: FiniteDuration)(f: => Unit)(
       implicit
       executor: ExecutionContext): Cancellable =
-    scheduleOnce(delay, new Runnable { override def run(): Unit = f })
+    scheduleOnce(delay, () => f)
 
   /**
    * Scala API: Schedules a Runnable to be run once with a delay, i.e. a time period that

actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala

@@ -100,22 +100,20 @@ final class VirtualizedExecutorService(
       } else if (underlyingShutdownScheduled.compareAndSet(false, true)) {
         VirtualThreadSupport.startVirtualThread(
           "pekko-virtualized-executor-shutdown",
-          new Runnable {
-            override def run(): Unit = {
-              var interrupted = false
-              try {
-                while (!executor.isTerminated) {
-                  try executor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS)
-                  catch {
-                    case _: InterruptedException => interrupted = true
-                  }
-                }
-              } finally {
-                shutdownUnderlying()
-                if (interrupted) {
-                  Thread.currentThread().interrupt()
+          () => {
+            var interrupted = false
+            try {
+              while (!executor.isTerminated) {
+                try executor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS)
+                catch {
+                  case _: InterruptedException => interrupted = true
                 }
               }
+            } finally {
+              shutdownUnderlying()
+              if (interrupted) {
+                Thread.currentThread().interrupt()
+              }
             }
           })
       }

actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala

@@ -497,7 +497,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onOpen(callback: => Unit): CircuitBreaker = addOnOpenListener(new Runnable { def run = callback })
+  def onOpen(callback: => Unit): CircuitBreaker = addOnOpenListener(() => callback)
 
   /**
    * Java API for onOpen
@@ -517,7 +517,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onHalfOpen(callback: => Unit): CircuitBreaker = addOnHalfOpenListener(new Runnable { def run = callback })
+  def onHalfOpen(callback: => Unit): CircuitBreaker = addOnHalfOpenListener(() => callback)
 
   /**
    * JavaAPI for onHalfOpen
@@ -538,7 +538,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onClose(callback: => Unit): CircuitBreaker = addOnCloseListener(new Runnable { def run = callback })
+  def onClose(callback: => Unit): CircuitBreaker = addOnCloseListener(() => callback)
 
   /**
    * JavaAPI for onClose
@@ -632,7 +632,7 @@ class CircuitBreaker(
    * @return CircuitBreaker for fluent usage
    */
   def onCallBreakerOpen(callback: => Unit): CircuitBreaker =
-    addOnCallBreakerOpenListener(new Runnable { def run = callback })
+    addOnCallBreakerOpenListener(() => callback)
 
   /**
    * JavaAPI for onCallBreakerOpen.
@@ -686,9 +686,7 @@ class CircuitBreaker(
     val iterator = successListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 
@@ -702,9 +700,7 @@ class CircuitBreaker(
     val iterator = callFailureListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 
@@ -718,9 +714,7 @@ class CircuitBreaker(
     val iterator = callTimeoutListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 

bench-jmh/src/main/scala/org/apache/pekko/stream/impl/OutputStreamSourceStageBenchmark.scala

@@ -43,15 +43,13 @@ class OutputStreamSourceStageBenchmark {
   @OperationsPerInvocation(WritesPerBench)
   def consumeWrites(): Unit = {
     val (os, done) = StreamConverters.asOutputStream().toMat(Sink.ignore)(Keep.both).run()
-    new Thread(new Runnable {
-      def run(): Unit = {
-        var counter = 0
-        while (counter > WritesPerBench) {
-          os.write(bytes)
-          counter += 1
-        }
-        os.close()
+    new Thread(() => {
+      var counter = 0
+      while (counter > WritesPerBench) {
+        os.write(bytes)
+        counter += 1
       }
+      os.close()
     }).start()
     Await.result(done, 30.seconds)
   }

cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/ActorSystemSpec.scala

@@ -205,9 +205,7 @@ class ActorSystemSpec
       withSystem("thread", Behaviors.empty[String]) { sys =>
         val p = Promise[Int]()
         sys.threadFactory
-          .newThread(new Runnable {
-            def run(): Unit = p.success(42)
-          })
+          .newThread(() => p.success(42))
           .start()
         p.future.futureValue should ===(42)
       }

cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala

@@ -439,7 +439,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * a certain size.
    */
   def registerOnMemberUp[T](code: => T): Unit =
-    registerOnMemberUp(new Runnable { def run() = code })
+    registerOnMemberUp((() => code): Runnable)
 
   /**
    * Java API: The supplied callback will be run, once, when current cluster member is `Up`.
@@ -457,7 +457,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * is not invoked. It's often better to use [[pekko.actor.CoordinatedShutdown]] for this purpose.
    */
   def registerOnMemberRemoved[T](code: => T): Unit =
-    registerOnMemberRemoved(new Runnable { override def run(): Unit = code })
+    registerOnMemberRemoved((() => code): Runnable)
 
   /**
    * Java API: The supplied thunk will be run, once, when current cluster member is `Removed`.