Prerelease

klein-common/pom.xml

@@ -20,10 +20,6 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
     <dependencies>
-        <dependency>
-            <groupId>com.caucho</groupId>
-            <artifactId>hessian</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.javassist</groupId>
             <artifactId>javassist</artifactId>

klein-consensus/klein-consensus-facade/pom.xml

@@ -28,6 +28,10 @@
             <groupId>com.ofcoder.klein.storage.api</groupId>
             <artifactId>klein-storage-facade</artifactId>
         </dependency>
+        <dependency>
+            <artifactId>klein-serializer-facade</artifactId>
+            <groupId>com.ofcoder.klein.serializer.facade</groupId>
+        </dependency>
     </dependencies>
 
 </project>

klein-consensus/klein-consensus-facade/src/main/java/com/ofcoder/klein/consensus/facade/AbstractInvokeCallback.java

@@ -16,17 +16,22 @@
  */
 package com.ofcoder.klein.consensus.facade;
 
-import java.nio.ByteBuffer;
-
-import com.ofcoder.klein.common.serialization.Hessian2Util;
 import com.ofcoder.klein.rpc.facade.InvokeCallback;
+import com.ofcoder.klein.serializer.Serializer;
+import com.ofcoder.klein.spi.ExtensionLoader;
 
 /**
  * Invoke callback.
  *
  * @author far.liu
  */
 public abstract class AbstractInvokeCallback<RES> implements InvokeCallback {
+    private final Serializer serializer;
+
+    public AbstractInvokeCallback() {
+        serializer = ExtensionLoader.getExtensionLoader(Serializer.class).register("hessian2");
+    }
+
     /**
      * invoke completed.
      *
@@ -35,8 +40,8 @@ public abstract class AbstractInvokeCallback<RES> implements InvokeCallback {
     public abstract void complete(RES result);
 
     @Override
-    public void complete(final ByteBuffer result) {
-        RES deserialize = Hessian2Util.deserialize(result.array());
+    public void complete(final byte[] result) {
+        RES deserialize = serializer.deserialize(result);
         complete(deserialize);
     }
 }

klein-consensus/klein-consensus-facade/src/main/java/com/ofcoder/klein/consensus/facade/AbstractRpcProcessor.java

@@ -16,18 +16,22 @@
  */
 package com.ofcoder.klein.consensus.facade;
 
-import java.nio.ByteBuffer;
-
 import com.ofcoder.klein.rpc.facade.RpcContext;
 import com.ofcoder.klein.rpc.facade.RpcProcessor;
-import com.ofcoder.klein.common.serialization.Hessian2Util;
+import com.ofcoder.klein.serializer.Serializer;
+import com.ofcoder.klein.spi.ExtensionLoader;
 
 /**
  * rpc processor.
  *
  * @author 释慧利
  */
 public abstract class AbstractRpcProcessor<R> implements RpcProcessor {
+    private final Serializer serializer;
+
+    public AbstractRpcProcessor() {
+        serializer = ExtensionLoader.getExtensionLoader(Serializer.class).register("hessian2");
+    }
 
     /**
      * handle request.
@@ -38,8 +42,13 @@ public abstract class AbstractRpcProcessor<R> implements RpcProcessor {
     public abstract void handleRequest(R request, RpcContext context);
 
     @Override
-    public void handleRequest(final ByteBuffer request, final RpcContext context) {
-        R deserialize = Hessian2Util.deserialize(request.array());
+    public void handleRequest(final byte[] request, final RpcContext context) {
+        R deserialize = serializer.deserialize(request);
         handleRequest(deserialize, context);
     }
+
+    protected void response(final Object msg, final RpcContext context) {
+        byte[] serialize = serializer.serialize(msg);
+        context.response(serialize);
+    }
 }

klein-consensus/klein-consensus-facade/src/main/java/com/ofcoder/klein/consensus/facade/Command.java

@@ -22,34 +22,14 @@
  * consensus content.
  */
 public interface Command extends Serializable {
-
-    Command NOOP = new Command() {
-
-        @Override
-        public String getGroup() {
-            return Noop.GROUP;
-        }
-
-        @Override
-        public Object getData() {
-            return Noop.DEFAULT;
-        }
-    };
-
     String getGroup();
 
     /**
      * client's input data.
      *
      * @return data
      */
-    Object getData();
+    byte[] getData();
 
-    /**
-     * No operation proposal.
-     */
-    class Noop implements Serializable {
-        public static final Noop DEFAULT = new Noop();
-        public static final String GROUP = "NOOP";
-    }
+    boolean getIfSystemOp();
 }

klein-consensus/klein-consensus-facade/src/main/java/com/ofcoder/klein/consensus/facade/Consensus.java

@@ -16,8 +16,6 @@
  */
 package com.ofcoder.klein.consensus.facade;
 
-import java.io.Serializable;
-
 import com.ofcoder.klein.consensus.facade.sm.SM;
 import com.ofcoder.klein.spi.SPI;
 
@@ -45,11 +43,9 @@ public interface Consensus extends Cluster {
      *              e.g. The input value of the state machine
      * @param apply Whether you need to wait until the state machine is applied
      *              If true, wait until the state machine is applied before returning
-     * @param <D>   result type
-     * @param <E>   request type
      * @return whether success
      */
-    <E extends Serializable, D extends Serializable> Result<D> propose(String group, E data, boolean apply);
+    Result propose(String group, byte[] data, boolean apply);
 
     /**
      * Obtain the consensus reached instanceId.
@@ -58,7 +54,7 @@ public interface Consensus extends Cluster {
      * @param group group name
      * @return instance id
      */
-    Result<Long> readIndex(String group);
+    Long readIndex(String group);
 
     /**
      * preheating.

klein-consensus/klein-consensus-facade/src/main/java/com/ofcoder/klein/consensus/facade/NoopCommand.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.ofcoder.klein.consensus.facade;
+
+/**
+ * noop command.
+ * A no-operation command implementation that represents a null operation in the consensus system.
+ * This command is typically used as a placeholder in the consensus protocol when there are no actual
+ * commands to process, helping maintain the consensus mechanism's continuity.
+ */
+public enum NoopCommand implements Command {
+    NOOP;
+
+    @Override
+    public String getGroup() {
+        return "NOOP";
+    }
+
+    @Override
+    public byte[] getData() {
+        return new byte[0];
+    }
+
+    @Override
+    public boolean getIfSystemOp() {
+        return true;
+    }
+}