From cc47ff65681f86b4100809f9585625ce4de2e234 Mon Sep 17 00:00:00 2001
From: jaysunxiao <jaysunxiao@gmail.com>
Date: Mon, 23 Aug 2021 19:46:27 +0800
Subject: [PATCH] =?UTF-8?q?feat[net]:=20=E6=96=B0=E5=A2=9E=E5=BC=82?=
 =?UTF-8?q?=E6=AD=A5=E5=BC=82=E5=B8=B8=E5=9B=9E=E8=B0=83=E6=96=B9=E6=B3=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../dispatcher/manager/PacketDispatcher.java   | 18 +++++++++++-------
 .../dispatcher/model/answer/AsyncAnswer.java   | 12 ++++++++++++
 .../dispatcher/model/answer/IAsyncAnswer.java  |  5 +++++
 3 files changed, 28 insertions(+), 7 deletions(-)

diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java
index 0a842108..9390711f 100644
--- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java
+++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java
@@ -236,21 +236,25 @@ public class PacketDispatcher implements IPacketDispatcher {
                         }
                         return response;
                     })
-                    .whenCompleteAsync((responsePacket, e) -> {
+                    .whenCompleteAsync((responsePacket, throwable) -> {
                         try {
                             PacketSignal.removeSignalAttachment(clientAttachment);
 
-                            // 如果有异常的话，whenCompleteAsync的下一个thenAccept不会执行
-                            if (e != null) {
-                                logger.error(ExceptionUtils.getMessage(e));
-                                return;
-                            }
-
                             // 接收者在同步或异步的消息处理中，又调用了异步的方法，这时候threadServerAttachment不为空
                             if (serverSignalPacketAttachment != null) {
                                 serverReceiveSignalPacketAttachment.set(serverSignalPacketAttachment);
                             }
 
+                            // 如果有异常的话，whenCompleteAsync的下一个thenAccept不会执行
+                            if (throwable != null) {
+                                var exceptionCallback = asyncAnswer.getExceptionCallback();
+                                if (exceptionCallback != null) {
+                                    exceptionCallback.accept(throwable);
+                                }
+                                logger.error(ExceptionUtils.getMessage(throwable));
+                                return;
+                            }
+
                             // 异步返回，回调业务逻辑
                             asyncAnswer.setFuturePacket((T) responsePacket);
                             asyncAnswer.consume();
diff --git a/net/src/main/java/com/zfoo/net/dispatcher/model/answer/AsyncAnswer.java b/net/src/main/java/com/zfoo/net/dispatcher/model/answer/AsyncAnswer.java
index 7a40dc08..2b2d96ae 100644
--- a/net/src/main/java/com/zfoo/net/dispatcher/model/answer/AsyncAnswer.java
+++ b/net/src/main/java/com/zfoo/net/dispatcher/model/answer/AsyncAnswer.java
@@ -33,6 +33,9 @@ public class AsyncAnswer<T extends IPacket> implements IAsyncAnswer<T> {
 
     private Runnable askCallback;
 
+    private Consumer<Throwable> exceptionCallback;
+
+
     @Override
     public IAsyncAnswer<T> thenAccept(Consumer<T> consumer) {
         consumerList.add(consumer);
@@ -45,6 +48,11 @@ public class AsyncAnswer<T extends IPacket> implements IAsyncAnswer<T> {
         askCallback.run();
     }
 
+    @Override
+    public void exceptionally(Consumer<Throwable> exceptionCallback) {
+        this.exceptionCallback = exceptionCallback;
+    }
+
     public void consume() {
         consumerList.forEach(it -> it.accept(futurePacket));
     }
@@ -72,4 +80,8 @@ public class AsyncAnswer<T extends IPacket> implements IAsyncAnswer<T> {
     public void setAskCallback(Runnable askCallback) {
         this.askCallback = askCallback;
     }
+
+    public Consumer<Throwable> getExceptionCallback() {
+        return exceptionCallback;
+    }
 }
diff --git a/net/src/main/java/com/zfoo/net/dispatcher/model/answer/IAsyncAnswer.java b/net/src/main/java/com/zfoo/net/dispatcher/model/answer/IAsyncAnswer.java
index 0266987d..49456fe3 100644
--- a/net/src/main/java/com/zfoo/net/dispatcher/model/answer/IAsyncAnswer.java
+++ b/net/src/main/java/com/zfoo/net/dispatcher/model/answer/IAsyncAnswer.java
@@ -30,4 +30,9 @@ public interface IAsyncAnswer<T extends IPacket> {
      */
     void whenComplete(Consumer<T> consumer);
 
+    /**
+     * 当异步种有异常会回调的方法
+     */
+    void exceptionally(Consumer<Throwable> exceptionCallback);
+
 }