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

Make use of SAM instead.

Browse source
This commit is contained in:
He-Pin 2023-09-23 22:32:54 +08:00 • committed by kerr
parent 07c5935ddb
commit 3ae618450f
3 changed files with 10 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

9
actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala
View file

@ -15,7 +15,6 @@ package org.apache.pekko.pattern
import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import java.util.function.BiConsumer
import scala.concurrent.{ ExecutionContext, Future, Promise }
import scala.concurrent.duration.FiniteDuration
@ -78,11 +77,9 @@ trait FutureTimeoutSupport {
using.scheduleOnce(duration) {
try {
val future = value
future.whenComplete(new BiConsumer[T, Throwable] {
override def accept(t: T, ex: Throwable): Unit = {
if (t != null) p.complete(t)
if (ex != null) p.completeExceptionally(ex)
}
future.handle[Unit]((t: T, ex: Throwable) => {
if (t != null) p.complete(t)
if (ex != null) p.completeExceptionally(ex)
})
} catch {
case NonFatal(ex) => p.completeExceptionally(ex)

17
actor/src/main/scala/org/apache/pekko/pattern/PipeToSupport.scala
View file

@ -14,7 +14,6 @@
package org.apache.pekko.pattern
import java.util.concurrent.CompletionStage
import java.util.function.BiConsumer
import scala.concurrent.{ ExecutionContext, Future }
import scala.util.{ Failure, Success }
@ -56,19 +55,15 @@ trait PipeToSupport {
final class PipeableCompletionStage[T](val future: CompletionStage[T])(
implicit @unused executionContext: ExecutionContext) {
def pipeTo(recipient: ActorRef)(implicit sender: ActorRef = Actor.noSender): CompletionStage[T] = {
future.whenComplete(new BiConsumer[T, Throwable] {
override def accept(t: T, ex: Throwable): Unit = {
if (t != null) recipient ! t
if (ex != null) recipient ! Status.Failure(ex)
}
future.whenComplete((t: T, ex: Throwable) => {
if (t != null) recipient ! t
if (ex != null) recipient ! Status.Failure(ex)
})
}
def pipeToSelection(recipient: ActorSelection)(implicit sender: ActorRef = Actor.noSender): CompletionStage[T] = {
future.whenComplete(new BiConsumer[T, Throwable] {
override def accept(t: T, ex: Throwable): Unit = {
if (t != null) recipient ! t
if (ex != null) recipient ! Status.Failure(ex)
}
future.whenComplete((t: T, ex: Throwable) => {
if (t != null) recipient ! t
if (ex != null) recipient ! Status.Failure(ex)
})
}
def to(recipient: ActorRef): PipeableCompletionStage[T] = to(recipient, Actor.noSender)