feat(multi-raft): update conflict code

1.update conflict code

Author: TheR1sing3un

src/main/java/io/openmessaging/storage/dledger/DLedgerRpcNettyService.java

@@ -41,9 +41,13 @@
 import io.openmessaging.storage.dledger.protocol.VoteResponse;
 import io.openmessaging.storage.dledger.utils.DLedgerUtils;
 
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.rocketmq.remoting.ChannelEventListener;
 import org.apache.rocketmq.remoting.netty.NettyClientConfig;
 import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
 import org.apache.rocketmq.remoting.netty.NettyRemotingServer;
@@ -65,13 +69,9 @@ public class DLedgerRpcNettyService extends DLedgerRpcService {
     private NettyRemotingServer remotingServer;
     private NettyRemotingClient remotingClient;
 
-    // selfId -> remotingClient
-    private ConcurrentHashMap<String, NettyRemotingClient> clientMap;
-
     private DLedgerProxy dLedgerProxy;
     private MemberState memberState;
 
-
     private ExecutorService futureExecutor = Executors.newFixedThreadPool(4, new ThreadFactory() {
         private AtomicInteger threadIndex = new AtomicInteger(0);
 
@@ -100,8 +100,15 @@ public Thread newThread(Runnable r) {
     });
 
     public DLedgerRpcNettyService(DLedgerProxy dLedgerProxy) {
+        this(dLedgerProxy, null, null, null);
+    }
+
+    public DLedgerRpcNettyService(DLedgerProxy dLedgerProxy, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig) {
+        this(dLedgerProxy, nettyServerConfig, nettyClientConfig, null);
+    }
+
+    public DLedgerRpcNettyService(DLedgerProxy dLedgerProxy, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
         this.dLedgerProxy = dLedgerProxy;
-        this.clientMap = new ConcurrentHashMap<>();
         DLedgerProxyConfig dLedgerProxyConfig = this.dLedgerProxy.getConfigManager().getdLedgerProxyConfig();
         NettyRequestProcessor protocolProcessor = new NettyRequestProcessor() {
             @Override
@@ -114,20 +121,25 @@ public boolean rejectRequest() {
                 return false;
             }
         };
+
         //register remoting server(We will only listen to one port. Limit in the configuration file)
         //check if the config has more than one port to bind
         if (!checkOnePort(dLedgerProxyConfig)) {
             logger.error("Bind the port error, because of more than one port in config");
             throw new RuntimeException("Bind the port error, because of more than one port in config");
         }
         DLedgerConfig dLedgerConfig = dLedgerProxyConfig.getConfigs().get(0);
-        NettyRemotingServer nettyRemotingServer = registerRemotingServer(dLedgerConfig.getSelfAddress(), protocolProcessor);
-        this.remotingServer = nettyRemotingServer;
+        String address = dLedgerConfig.getSelfAddress();
+        if (nettyServerConfig == null) {
+            nettyServerConfig = new NettyServerConfig();
+        }
+        nettyServerConfig.setListenPort(Integer.valueOf(address.split(":")[1]));
+        this.remotingServer = registerRemotingServer(nettyServerConfig, channelEventListener, protocolProcessor);
         //start the remoting client
-        for (DLedgerConfig config : dLedgerProxyConfig.getConfigs()) {
-            this.clientMap.put(config.getSelfId(), new NettyRemotingClient(new NettyClientConfig(), null));
+        if (nettyClientConfig == null) {
+            nettyClientConfig = new NettyClientConfig();
         }
-        this.remotingClient = new NettyRemotingClient(new NettyClientConfig(), null);
+        this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
     }
 
     private boolean checkOnePort(DLedgerProxyConfig dLedgerProxyConfig) {
@@ -156,10 +168,8 @@ private void registerProcessor(NettyRemotingServer remotingServer, NettyRequestP
         remotingServer.registerProcessor(DLedgerRequestCode.LEADERSHIP_TRANSFER.getCode(), protocolProcessor, null);
     }
 
-    private NettyRemotingServer registerRemotingServer(String address, NettyRequestProcessor protocolProcessor) {
-        NettyServerConfig nettyServerConfig = new NettyServerConfig();
-        nettyServerConfig.setListenPort(Integer.valueOf(address.split(":")[1]));
-        NettyRemotingServer remotingServer = new NettyRemotingServer(nettyServerConfig, null);
+    private NettyRemotingServer registerRemotingServer(NettyServerConfig nettyServerConfig, ChannelEventListener channelEventListener, NettyRequestProcessor protocolProcessor) {
+        NettyRemotingServer remotingServer = new NettyRemotingServer(nettyServerConfig, channelEventListener);
         registerProcessor(remotingServer, protocolProcessor);
         return remotingServer;
     }
@@ -514,4 +524,8 @@ public DLedgerProxy getdLedgerProxy() {
     public void setdLedgerProxy(DLedgerProxy dLedgerProxy) {
         this.dLedgerProxy = dLedgerProxy;
     }
+
+    public NettyRemotingServer getRemotingServer() {
+        return remotingServer;
+    }
 }

src/main/java/io/openmessaging/storage/dledger/dledger/DLedgerManager.java

@@ -26,7 +26,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.HashMap;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class DLedgerManager {
@@ -53,7 +52,7 @@ public DLedgerServer getDLedgerServer(final String selfId) {
         return this.servers.get(selfId);
     }
 
-    public DLedgerServer getDLedgerServer(final  String groupId, final String selfId){
+    public DLedgerServer getDLedgerServer(final String groupId, final String selfId) {
         DLedgerServer dLedgerServer = this.servers.get(selfId);
         return dLedgerServer.getdLedgerConfig().getGroup().equals(groupId) ? dLedgerServer : null;
     }

src/main/java/io/openmessaging/storage/dledger/dledger/DLedgerProxy.java

@@ -16,7 +16,11 @@
 
 package io.openmessaging.storage.dledger.dledger;
 
-import io.openmessaging.storage.dledger.*;
+import io.openmessaging.storage.dledger.AppendFuture;
+import io.openmessaging.storage.dledger.DLedgerConfig;
+import io.openmessaging.storage.dledger.DLedgerRpcNettyService;
+import io.openmessaging.storage.dledger.DLedgerRpcService;
+import io.openmessaging.storage.dledger.DLedgerServer;
 import io.openmessaging.storage.dledger.exception.DLedgerException;
 import io.openmessaging.storage.dledger.protocol.AppendEntryRequest;
 import io.openmessaging.storage.dledger.protocol.AppendEntryResponse;
@@ -37,6 +41,10 @@
 import io.openmessaging.storage.dledger.protocol.VoteRequest;
 import io.openmessaging.storage.dledger.protocol.VoteResponse;
 import io.openmessaging.storage.dledger.utils.PreConditions;
+import org.apache.rocketmq.remoting.ChannelEventListener;
+import org.apache.rocketmq.remoting.netty.NettyClientConfig;
+import org.apache.rocketmq.remoting.netty.NettyRemotingServer;
+import org.apache.rocketmq.remoting.netty.NettyServerConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,10 +63,19 @@ public class DLedgerProxy implements DLedgerProtocolHandler {
 
     private DLedgerRpcService dLedgerRpcService;
 
-    public DLedgerProxy(final DLedgerProxyConfig dLedgerProxyConfig) {
+    public DLedgerProxy(DLedgerProxyConfig dLedgerProxyConfig) {
+        this(dLedgerProxyConfig, null, null, null);
+    }
+
+    public DLedgerProxy(DLedgerProxyConfig dLedgerProxyConfig, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig) {
+        this(dLedgerProxyConfig, nettyServerConfig, nettyClientConfig, null);
+    }
+
+
+    public DLedgerProxy(DLedgerProxyConfig dLedgerProxyConfig, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
         try {
             this.configManager = new ConfigManager(dLedgerProxyConfig);
-            this.dLedgerRpcService = new DLedgerRpcNettyService(this);
+            this.dLedgerRpcService = new DLedgerRpcNettyService(this, nettyServerConfig, nettyClientConfig, channelEventListener);
             this.dLedgerManager = new DLedgerManager(dLedgerProxyConfig, this.dLedgerRpcService);
         } catch (Exception e) {
             logger.error("[Proxy][DLedgerProxy] fail to construct", e);
@@ -253,4 +270,11 @@ public void shutdown() {
         this.dLedgerRpcService.shutdown();
     }
 
+    public NettyRemotingServer getRemotingServer() {
+        if (this.dLedgerRpcService instanceof DLedgerRpcNettyService) {
+            return ((DLedgerRpcNettyService) this.dLedgerRpcService).getRemotingServer();
+        }
+        return null;
+    }
+
 }