diff --git a/broker/src/main/java/io/moquette/imhandler/IMHandler.java b/broker/src/main/java/io/moquette/imhandler/IMHandler.java index db6334dc..786e9643 100755 --- a/broker/src/main/java/io/moquette/imhandler/IMHandler.java +++ b/broker/src/main/java/io/moquette/imhandler/IMHandler.java @@ -205,9 +205,7 @@ abstract public class IMHandler { } } - if(errorCode != ErrorCode.ERROR_CODE_ASYNC_HANDLER) { - response(ackPayload, errorCode, callback); - } + response(ackPayload, errorCode, callback); } else { LOG.error("Handler {} preAction failure", this.getClass().getName()); ByteBuf ackPayload = Unpooled.buffer(1); diff --git a/broker/src/main/java/io/moquette/persistence/RPCCenter.java b/broker/src/main/java/io/moquette/persistence/RPCCenter.java index 2796b3a7..3d112767 100644 --- a/broker/src/main/java/io/moquette/persistence/RPCCenter.java +++ b/broker/src/main/java/io/moquette/persistence/RPCCenter.java @@ -73,7 +73,7 @@ public class RPCCenter { info.future.cancel(true); if (info.callback != null) { info.callback.getResponseExecutor().execute(() -> { - if (errorCode == 0) { + if (errorCode == 0 || errorCode == ErrorCode.ERROR_CODE_ASYNC_HANDLER.getCode()) { info.callback.onSuccess(message); } else { info.callback.onError(ErrorCode.fromCode(errorCode)); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java index 99289a84..2665d1b9 100755 --- a/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java +++ b/broker/src/main/java/io/moquette/spi/impl/Qos1PublishHandler.java @@ -23,8 +23,10 @@ import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_NOT_IMPLEMENT; import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_OVER_FREQUENCY; import static cn.wildfirechat.common.ErrorCode.ERROR_CODE_SUCCESS; +import java.io.ByteArrayOutputStream; import java.io.IOException; import java.util.*; +import java.util.zip.GZIPOutputStream; import cn.wildfirechat.server.ThreadPoolExecutorWrapper; import com.google.gson.Gson; @@ -174,8 +176,23 @@ public class Qos1PublishHandler extends QosPublishHandler { data = AES.AESEncrypt(data, ""); } else { MemorySessionStore.Session session = m_sessionStore.getSession(clientID); - if (session != null && session.getUsername().equals(fromUser)) - data = AES.AESEncrypt(data, session.getSecret()); + if (session != null && session.getUsername().equals(fromUser)) { + if (data.length > 10*1024) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + GZIPOutputStream gzip; + try { + gzip = new GZIPOutputStream(out); + gzip.write(data); + gzip.close(); + } catch (Exception e) { + e.printStackTrace(); + } + data = out.toByteArray(); + code = (byte)ErrorCode.ERROR_CODE_ASYNC_HANDLER.code; + } + + data = AES.AESEncrypt(data, session.getSecret()); + } } } ackPayload.clear(); diff --git a/common/src/main/java/cn/wildfirechat/common/ErrorCode.java b/common/src/main/java/cn/wildfirechat/common/ErrorCode.java index 381bbd80..a5f78484 100755 --- a/common/src/main/java/cn/wildfirechat/common/ErrorCode.java +++ b/common/src/main/java/cn/wildfirechat/common/ErrorCode.java @@ -56,7 +56,7 @@ public enum ErrorCode { ERROR_CODE_NOT_IMPLEMENT(254, "not implement"), - ERROR_CODE_ASYNC_HANDLER(255, "异步执行,服务器内部逻辑需要此代码,为正常情况,不能返回客户端"),; + ERROR_CODE_SUCCESS_GZIPED(255, "success withe gzip response"),; public int code; public String msg;