slave sync is run on server

geekhuyang · Jan 22, 2019 · 2439f87 · 2439f87
1 parent f8aa621
commit 2439f87
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 2 deletions.
diff --git a/qmq-remoting/src/main/java/qunar/tc/qmq/netty/NettyClientConfig.java b/qmq-remoting/src/main/java/qunar/tc/qmq/netty/NettyClientConfig.java
@@ -26,6 +26,7 @@ public class NettyClientConfig {
     private int clientChannelMaxIdleTimeSeconds = 120;
     private int clientSocketSndBufSize = Integer.getInteger("qmq.remoting.socket.sndbuf.size", 65535);
     private int clientSocketRcvBufSize = Integer.getInteger("qmq.remoting.socket.rcvbuf.size", 65535);
+    private boolean isServer = false;
 
     public int getClientWorkerThreads() {
         return clientWorkerThreads;
@@ -74,4 +75,12 @@ public int getClientSocketRcvBufSize() {
     public void setClientSocketRcvBufSize(int clientSocketRcvBufSize) {
         this.clientSocketRcvBufSize = clientSocketRcvBufSize;
     }
+
+    public boolean isServer() {
+        return isServer;
+    }
+
+    public void setServer(boolean server) {
+        isServer = server;
+    }
 }
diff --git a/qmq-remoting/src/main/java/qunar/tc/qmq/netty/client/NettyClient.java b/qmq-remoting/src/main/java/qunar/tc/qmq/netty/client/NettyClient.java
@@ -72,7 +72,7 @@ protected ChannelInitializer<SocketChannel> newChannelInitializer(final NettyCli
             protected void initChannel(SocketChannel ch) throws Exception {
                 ch.pipeline().addLast(eventExecutors,
                         new EncodeHandler(),
-                        new DecodeHandler(false),
+                        new DecodeHandler(config.isServer()),
                         new IdleStateHandler(0, 0, config.getClientChannelMaxIdleTimeSeconds()),
                         connectManager,
                         clientHandler);

diff --git a/qmq-sync/src/main/java/qunar/tc/qmq/sync/SlaveSyncClient.java b/qmq-sync/src/main/java/qunar/tc/qmq/sync/SlaveSyncClient.java
@@ -38,7 +38,9 @@ public class SlaveSyncClient {
 
     public SlaveSyncClient(DynamicConfig config) {
         this.client = NettyClient.getClient();
-        this.client.start(new NettyClientConfig());
+        NettyClientConfig clientConfig = new NettyClientConfig();
+        clientConfig.setServer(true);
+        this.client.start(clientConfig);
         this.master = BrokerConfig.getMasterAddress();
 
         config.addListener(conf -> timeout = conf.getLong("slave.sync.timeout", 3000L));