feat: using "interceptor" to enhance the api(compress)

src/main/java/com/xiaomi/infra/pegasus/operator/rrdb_check_and_mutate_operator.java

@@ -47,6 +47,10 @@ public check_and_mutate_response get_response() {
     return resp;
   }
 
+  public check_and_mutate_request get_request() {
+    return request;
+  }
+
   private check_and_mutate_request request;
   private check_and_mutate_response resp;
 }

src/main/java/com/xiaomi/infra/pegasus/operator/rrdb_check_and_set_operator.java

@@ -47,6 +47,10 @@ public check_and_set_response get_response() {
     return resp;
   }
 
+  public check_and_set_request get_request() {
+    return request;
+  }
+
   private check_and_set_request request;
   private check_and_set_response resp;
 }

src/main/java/com/xiaomi/infra/pegasus/operator/rrdb_multi_put_operator.java

@@ -47,6 +47,10 @@ public update_response get_response() {
     return resp;
   }
 
+  public multi_put_request get_request() {
+    return request;
+  }
+
   private multi_put_request request;
   private update_response resp;
 }

src/main/java/com/xiaomi/infra/pegasus/operator/rrdb_put_operator.java

@@ -46,6 +46,10 @@ public update_response get_response() {
     return resp;
   }
 
+  public update_request get_request() {
+    return request;
+  }
+
   private update_request request;
   private update_response resp;
 }

src/main/java/com/xiaomi/infra/pegasus/rpc/TableOptions.java

@@ -7,6 +7,7 @@
 public class TableOptions {
   private final KeyHasher keyHasher;
   private final int backupRequestDelayMs;
+  private final boolean enableCompress;
 
   public KeyHasher keyHasher() {
     return this.keyHasher;
@@ -17,15 +18,20 @@ public int backupRequestDelayMs() {
   }
 
   public static TableOptions forTest() {
-    return new TableOptions(KeyHasher.DEFAULT, 0);
+    return new TableOptions(KeyHasher.DEFAULT, 0, false);
   }
 
-  public TableOptions(KeyHasher h, int backupRequestDelay) {
+  public TableOptions(KeyHasher h, int backupRequestDelay, boolean enableCompress) {
     this.keyHasher = h;
     this.backupRequestDelayMs = backupRequestDelay;
+    this.enableCompress = enableCompress;
   }
 
   public boolean enableBackupRequest() {
     return backupRequestDelayMs > 0;
   }
+
+  public boolean enableCompress() {
+    return enableCompress;
+  }
 }

src/main/java/com/xiaomi/infra/pegasus/rpc/interceptor/CompressInterceptor.java

@@ -0,0 +1,118 @@
+package com.xiaomi.infra.pegasus.rpc.interceptor;
+
+import com.xiaomi.infra.pegasus.apps.key_value;
+import com.xiaomi.infra.pegasus.apps.mutate;
+import com.xiaomi.infra.pegasus.base.error_code.error_types;
+import com.xiaomi.infra.pegasus.client.PException;
+import com.xiaomi.infra.pegasus.operator.rrdb_check_and_mutate_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_check_and_set_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_get_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_multi_get_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_multi_put_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_put_operator;
+import com.xiaomi.infra.pegasus.operator.rrdb_scan_operator;
+import com.xiaomi.infra.pegasus.rpc.async.ClientRequestRound;
+import com.xiaomi.infra.pegasus.rpc.async.TableHandler;
+import com.xiaomi.infra.pegasus.tools.ZstdWrapper;
+import java.util.List;
+
+public class CompressInterceptor implements TableInterceptor {
+  private boolean isOpen;
+
+  public CompressInterceptor(boolean isOpen) {
+    this.isOpen = isOpen;
+  }
+
+  @Override
+  public void before(ClientRequestRound clientRequestRound, TableHandler tableHandler)
+      throws PException {
+    if (!isOpen) {
+      return;
+    }
+    tryCompress(clientRequestRound);
+  }
+
+  @Override
+  public void after(
+      ClientRequestRound clientRequestRound, error_types errno, TableHandler tableHandler)
+      throws PException {
+    if (errno != error_types.ERR_OK || !isOpen) {
+      return;
+    }
+    tryDecompress(clientRequestRound);
+  }
+
+  private void tryCompress(ClientRequestRound clientRequestRound) throws PException {
+    String operatorName = clientRequestRound.getOperator().name();
+    switch (operatorName) {
+      case "put":
+        {
+          rrdb_put_operator operator = (rrdb_put_operator) clientRequestRound.getOperator();
+          operator.get_request().value.data =
+              ZstdWrapper.compress(operator.get_request().value.data);
+        }
+      case "multi_put":
+        {
+          rrdb_multi_put_operator operator =
+              (rrdb_multi_put_operator) clientRequestRound.getOperator();
+          List<key_value> kvs = operator.get_request().kvs;
+          for (key_value kv : kvs) {
+            kv.value.data = ZstdWrapper.compress(kv.value.data);
+          }
+          break;
+        }
+      case "check_and_set":
+        {
+          rrdb_check_and_set_operator operator =
+              (rrdb_check_and_set_operator) clientRequestRound.getOperator();
+          operator.get_request().set_value.data =
+              ZstdWrapper.compress(operator.get_request().set_value.data);
+          break;
+        }
+      case "check_and_mutate":
+        {
+          rrdb_check_and_mutate_operator operator =
+              (rrdb_check_and_mutate_operator) clientRequestRound.getOperator();
+          List<mutate> mutates = operator.get_request().mutate_list;
+          for (mutate mu : mutates) {
+            mu.value.data = ZstdWrapper.compress(mu.value.data);
+          }
+        }
+      default:
+        throw new PException("unsupported operator = " + operatorName);
+    }
+  }
+
+  private void tryDecompress(ClientRequestRound clientRequestRound) throws PException {
+    String operatorName = clientRequestRound.getOperator().name();
+    switch (operatorName) {
+      case "get":
+        {
+          ZstdWrapper.tryDecompress(
+              ((rrdb_get_operator) clientRequestRound.getOperator()).get_response().value.data);
+          break;
+        }
+      case "multi_get":
+        {
+          rrdb_multi_get_operator operator =
+              (rrdb_multi_get_operator) clientRequestRound.getOperator();
+          List<key_value> kvs = operator.get_response().kvs;
+          for (key_value kv : kvs) {
+            kv.value.data = ZstdWrapper.tryDecompress(kv.value.data);
+          }
+          break;
+        }
+      case "scan":
+        {
+          rrdb_scan_operator operator = (rrdb_scan_operator) clientRequestRound.getOperator();
+          List<key_value> kvs = operator.get_response().kvs;
+          for (key_value kv : kvs) {
+            kv.value.data = ZstdWrapper.tryDecompress(kv.value.data);
+          }
+          break;
+        }
+      default:
+        throw new PException("unsupported operator = " + operatorName);
+    }
+  }
+}

src/main/java/com/xiaomi/infra/pegasus/rpc/interceptor/InterceptorManger.java

@@ -1,18 +1,22 @@
 package com.xiaomi.infra.pegasus.rpc.interceptor;
 
 import com.xiaomi.infra.pegasus.base.error_code.error_types;
+import com.xiaomi.infra.pegasus.client.PException;
 import com.xiaomi.infra.pegasus.rpc.TableOptions;
 import com.xiaomi.infra.pegasus.rpc.async.ClientRequestRound;
 import com.xiaomi.infra.pegasus.rpc.async.TableHandler;
 import java.util.ArrayList;
 import java.util.List;
+import org.slf4j.Logger;
 
 public class InterceptorManger {
+  private static final Logger logger = org.slf4j.LoggerFactory.getLogger(InterceptorManger.class);
 
   private List<TableInterceptor> interceptors = new ArrayList<>();
 
   public InterceptorManger(TableOptions options) {
-    register(new BackupRequestInterceptor(options.enableBackupRequest()));
+    this.register(new BackupRequestInterceptor(options.enableBackupRequest()))
+        .register(new CompressInterceptor(options.enableCompress()));
   }
 
   private InterceptorManger register(TableInterceptor interceptor) {
@@ -22,14 +26,22 @@ private InterceptorManger register(TableInterceptor interceptor) {
 
   public void before(ClientRequestRound clientRequestRound, TableHandler tableHandler) {
     for (TableInterceptor interceptor : interceptors) {
-      interceptor.before(clientRequestRound, tableHandler);
+      try {
+        interceptor.before(clientRequestRound, tableHandler);
+      } catch (PException e) {
+        logger.warn("interceptor-before execute failed!", e);
+      }
     }
   }
 
   public void after(
       ClientRequestRound clientRequestRound, error_types errno, TableHandler tableHandler) {
     for (TableInterceptor interceptor : interceptors) {
-      interceptor.after(clientRequestRound, errno, tableHandler);
+      try {
+        interceptor.after(clientRequestRound, errno, tableHandler);
+      } catch (PException e) {
+        logger.warn("interceptor-after execute failed!", e);
+      }
     }
   }
 }

src/main/java/com/xiaomi/infra/pegasus/rpc/interceptor/TableInterceptor.java

@@ -1,12 +1,14 @@
 package com.xiaomi.infra.pegasus.rpc.interceptor;
 
 import com.xiaomi.infra.pegasus.base.error_code.error_types;
+import com.xiaomi.infra.pegasus.client.PException;
 import com.xiaomi.infra.pegasus.rpc.async.ClientRequestRound;
 import com.xiaomi.infra.pegasus.rpc.async.TableHandler;
 
 public interface TableInterceptor {
   // The behavior before sending the RPC to a table.
-  void before(ClientRequestRound clientRequestRound, TableHandler tableHandler);
+  void before(ClientRequestRound clientRequestRound, TableHandler tableHandler) throws PException;
   // The behavior after getting reply or failure of the RPC.
-  void after(ClientRequestRound clientRequestRound, error_types errno, TableHandler tableHandler);
+  void after(ClientRequestRound clientRequestRound, error_types errno, TableHandler tableHandler)
+      throws PException;
 }

src/main/java/com/xiaomi/infra/pegasus/tools/ZstdWrapper.java

@@ -12,6 +12,23 @@ public class ZstdWrapper {
 
   private ZstdWrapper() {}
 
+  /**
+   * try decompress the `src`, return `src` directly if failed
+   *
+   * @param src
+   * @return
+   */
+  public static byte[] tryDecompress(byte[] src) {
+    byte[] decompressedValue;
+    try {
+      decompressedValue = decompress(src);
+    } catch (PException e) {
+      // decompress fail
+      decompressedValue = src;
+    }
+    return decompressedValue;
+  }
+
   /**
    * Compresses the `src` and returns the compressed.
    *