Expose nettyRemotingServer (#153)

* add constructor, expose nettyRemotingServer

* code review

Author: hzh0425

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

@@ -43,6 +43,7 @@
 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;
@@ -96,6 +97,14 @@ public Thread newThread(Runnable r) {
     });
 
     public DLedgerRpcNettyService(DLedgerServer dLedgerServer) {
+        this(dLedgerServer, null, null, null);
+    }
+
+    public DLedgerRpcNettyService(DLedgerServer dLedgerServer, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig) {
+        this(dLedgerServer, nettyServerConfig, nettyClientConfig, null);
+    }
+
+    public DLedgerRpcNettyService(DLedgerServer dLedgerServer, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
         this.dLedgerServer = dLedgerServer;
         this.memberState = dLedgerServer.getMemberState();
         NettyRequestProcessor protocolProcessor = new NettyRequestProcessor() {
@@ -110,9 +119,11 @@ public boolean rejectRequest() {
             }
         };
         //start the remoting server
-        NettyServerConfig nettyServerConfig = new NettyServerConfig();
-        nettyServerConfig.setListenPort(Integer.valueOf(memberState.getSelfAddr().split(":")[1]));
-        this.remotingServer = new NettyRemotingServer(nettyServerConfig, null);
+        if (nettyServerConfig == null) {
+            nettyServerConfig = new NettyServerConfig();
+        }
+        nettyServerConfig.setListenPort(Integer.parseInt(memberState.getSelfAddr().split(":")[1]));
+        this.remotingServer = new NettyRemotingServer(nettyServerConfig, channelEventListener);
         this.remotingServer.registerProcessor(DLedgerRequestCode.METADATA.getCode(), protocolProcessor, null);
         this.remotingServer.registerProcessor(DLedgerRequestCode.APPEND.getCode(), protocolProcessor, null);
         this.remotingServer.registerProcessor(DLedgerRequestCode.GET.getCode(), protocolProcessor, null);
@@ -123,8 +134,10 @@ public boolean rejectRequest() {
         this.remotingServer.registerProcessor(DLedgerRequestCode.LEADERSHIP_TRANSFER.getCode(), protocolProcessor, null);
 
         //start the remoting client
-        this.remotingClient = new NettyRemotingClient(new NettyClientConfig(), null);
-
+        if (nettyClientConfig == null) {
+            nettyClientConfig = new NettyClientConfig();
+        }
+        this.remotingClient = new NettyRemotingClient(nettyClientConfig, null);
     }
 
     private String getPeerAddr(RequestOrResponse request) {
@@ -476,4 +489,8 @@ public DLedgerServer getdLedgerServer() {
     public void setdLedgerServer(DLedgerServer dLedgerServer) {
         this.dLedgerServer = dLedgerServer;
     }
+
+    public NettyRemotingServer getRemotingServer() {
+        return remotingServer;
+    }
 }

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

@@ -57,6 +57,10 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.CompletableFuture;
 
+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;
 
@@ -76,10 +80,18 @@ public class DLedgerServer implements DLedgerProtocolHandler {
     private Optional<StateMachineCaller> fsmCaller;
 
     public DLedgerServer(DLedgerConfig dLedgerConfig) {
+        this(dLedgerConfig, null, null, null);
+    }
+
+    public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig) {
+        this(dLedgerConfig, nettyServerConfig, nettyClientConfig, null);
+    }
+
+    public DLedgerServer(DLedgerConfig dLedgerConfig, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig, ChannelEventListener channelEventListener) {
         this.dLedgerConfig = dLedgerConfig;
         this.memberState = new MemberState(dLedgerConfig);
         this.dLedgerStore = createDLedgerStore(dLedgerConfig.getStoreType(), this.dLedgerConfig, this.memberState);
-        dLedgerRpcService = new DLedgerRpcNettyService(this);
+        dLedgerRpcService = new DLedgerRpcNettyService(this, nettyServerConfig, nettyClientConfig, channelEventListener);
         dLedgerEntryPusher = new DLedgerEntryPusher(dLedgerConfig, memberState, dLedgerStore, dLedgerRpcService);
         dLedgerLeaderElector = new DLedgerLeaderElector(dLedgerConfig, memberState, dLedgerRpcService);
         executorService = Executors.newSingleThreadScheduledExecutor(r -> {
@@ -433,4 +445,11 @@ public DLedgerLeaderElector getdLedgerLeaderElector() {
     public DLedgerConfig getdLedgerConfig() {
         return dLedgerConfig;
     }
+
+    public NettyRemotingServer getRemotingServer() {
+        if (this.dLedgerRpcService instanceof DLedgerRpcNettyService) {
+            return ((DLedgerRpcNettyService)this.dLedgerRpcService).getRemotingServer();
+        }
+        return null;
+    }
 }