support option to enable compression and hide details of compression from users about pegasus-java-client HOT 1 OPEN

class TableOptions {
  // Compression is one of the options.
  bool enableCompression() {}
  bool enableAutoRetry() {}
  bool enableBackupRequest() {}

  TableOptions withZstdCompression() {}
  TableOptions withAutoRetry(int maxRetries, BackoffPolicy backoff) {}
  TableOptions withBackupRequest(int backupRequestDelayMs) {}
}

class PegasusClientInterface {
  PegasusTableInterface openTable(String tableName, TableOptions opts);
}

// TableInterceptor is a plugin to control the behavior of an RPC to a Pegasus table.
// 
//        request
//  client ----> Pegasus Table
//           |
//           | ---- TableInterceptor is a middle layer to control the RPC process.
//           |
//         <---- 
//        response 
//
interface TableInterceptor {
  // The behavior before the ReplicaSession sends the RPC.
  void interceptBefore(ClientRequestRound rpc, TableHandler table, TableOptions opts) throws Exception;
  // The behavior after the ReplicaSession get reply or failure of the RPC.
  void interceptAfter(ClientRequestRound rpc, error_types errno, TableHandler table, TableOptions opts) Exception;
}

class BackupRequestInterceptor implements TableInterceptor {
  void interceptBefore(ClientRequestRound rpc, TableHandler table, TableOptions opts) throws Exception {
    if (opts.enableBackupRequest()) {
      rpc.backupRequestTask = executor.schedule();
    }
  }

  void interceptAfter(ClientRequestRound rpc, TableHandler table, TableOptions opts) throws Exception {}

  private int backupRequestDelayMs;
}

class CompressionInterceptor implements TableInterceptor {
  void interceptBefore(ClientRequestRound rpc, TableHandler table, TableOptions opts) throws Exception {
    if(opts.enableCompression()) {
      rpc.operator.compress();
    }
  }

  void interceptAfter(ClientRequestRound rpc, TableHandler table, TableOptions opts) throws Exception {
    if(opts.enableCompression()) {
      rpc.operator.decompress();
    }
  }
}

class TableHandler {
  public TableHandler(ClusterManager mgr, String name, TableOptions options) {
    ...
    interceptors.add(new CompressionInterceptor()).add(new BackupRequestInterceptor());    
  }

  void call() {
    for(TableInterceptor interceptor : interceptors) {
      interceptor.interceptBefore();
    }
    primarySession.send( () -> {
      for(TableInterceptor interceptor : interceptors) {
        interceptor.interceptAfter();
      }
    }
  }

  private List<TableInterceptor> interceptors;
}