[ISSUE #3949] Simplify Channels

proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java

@@ -26,7 +26,7 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.function.Supplier;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.proxy.config.ConfigurationManager;
-import org.apache.rocketmq.proxy.grpc.adapter.channel.SendMessageChannel;
+import org.apache.rocketmq.proxy.common.Cleaner;
 import org.apache.rocketmq.proxy.grpc.common.InterceptorConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -102,13 +102,12 @@ public class ChannelManager {
             Iterator<Map.Entry<String, SimpleChannel>> iterator = clientIdChannelMap.entrySet().iterator();
             while (iterator.hasNext()) {
                 Map.Entry<String, SimpleChannel> entry = iterator.next();
-                if (!entry.getValue()
-                    .isActive()) {
+                if (!entry.getValue().isActive()) {
                     iterator.remove();
                 } else {
-                    if (entry.getValue() instanceof SendMessageChannel) {
-                        SendMessageChannel channel = (SendMessageChannel) entry.getValue();
-                        channel.cleanExpiredRequests();
+                    if (entry.getValue() instanceof Cleaner) {
+                        Cleaner cleaner = (Cleaner) entry.getValue();
+                        cleaner.clean();
                     }
                 }
             }

proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java

@@ -23,21 +23,29 @@ import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.proxy.common.Cleaner;
 import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
+import org.apache.rocketmq.proxy.grpc.adapter.handler.ResponseHandler;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
-public class InvocationChannel<R, W> extends SimpleChannel {
+public abstract class InvocationChannel<R, W> extends SimpleChannel implements Cleaner {
     protected final ConcurrentMap<Integer, InvocationContext<R, W>> inFlightRequestMap;
+    protected final ResponseHandler<R, W> handler;
 
-    public InvocationChannel(SimpleChannel simpleChannel) {
-        super(simpleChannel);
+    public InvocationChannel(ResponseHandler<R, W> handler) {
+        super(ChannelManager.createSimpleChannelDirectly());
         this.inFlightRequestMap = new ConcurrentHashMap<>();
+        this.handler = handler;
     }
 
     @Override
     public ChannelFuture writeAndFlush(Object msg) {
         if (msg instanceof RemotingCommand) {
             RemotingCommand responseCommand = (RemotingCommand) msg;
+            InvocationContext<R, W> context = inFlightRequestMap.remove(responseCommand.getOpaque());
+            if (null != context) {
+                handler.handle(responseCommand, context);
+            }
             inFlightRequestMap.remove(responseCommand.getOpaque());
         }
         return super.writeAndFlush(msg);
@@ -64,7 +72,8 @@ public class InvocationChannel<R, W> extends SimpleChannel {
         inFlightRequestMap.remove(opaque);
     }
 
-    public void cleanExpiredRequests() {
+    @Override
+    public void clean() {
         Iterator<Map.Entry<Integer, InvocationContext<R, W>>> iterator = inFlightRequestMap.entrySet().iterator();
         int count = 0;
         while (iterator.hasNext()) {

proxy/src/main/java/org/apache/rocketmq/proxy/common/Cleaner.java

@@ -0,0 +1,22 @@
+/*
+ * 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 org.apache.rocketmq.proxy.common;
+
+public interface Cleaner {
+    void clean();
+}

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/ReceiveMessageChannel.java

@@ -19,30 +19,11 @@ package org.apache.rocketmq.proxy.grpc.adapter.channel;
 
 import apache.rocketmq.v1.ReceiveMessageRequest;
 import apache.rocketmq.v1.ReceiveMessageResponse;
-import io.netty.channel.ChannelFuture;
-import org.apache.rocketmq.proxy.channel.ChannelManager;
 import org.apache.rocketmq.proxy.channel.InvocationChannel;
-import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
 import org.apache.rocketmq.proxy.grpc.adapter.handler.ReceiveMessageResponseHandler;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 public class ReceiveMessageChannel extends InvocationChannel<ReceiveMessageRequest, ReceiveMessageResponse> {
-    private final ReceiveMessageResponseHandler handler;
-
     public ReceiveMessageChannel(ReceiveMessageResponseHandler handler) {
-        super(ChannelManager.createSimpleChannelDirectly());
-        this.handler = handler;
-    }
-
-    @Override
-    public ChannelFuture writeAndFlush(Object msg) {
-        if (msg instanceof RemotingCommand) {
-            RemotingCommand responseCommand = (RemotingCommand) msg;
-            InvocationContext<ReceiveMessageRequest, ReceiveMessageResponse> context = inFlightRequestMap.remove(responseCommand.getOpaque());
-            if (null != context) {
-                handler.handle(responseCommand, context);
-            }
-        }
-        return super.writeAndFlush(msg);
+        super(handler);
     }
 }

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SendMessageChannel.java

@@ -19,31 +19,11 @@ package org.apache.rocketmq.proxy.grpc.adapter.channel;
 
 import apache.rocketmq.v1.SendMessageRequest;
 import apache.rocketmq.v1.SendMessageResponse;
-import io.netty.channel.ChannelFuture;
-import org.apache.rocketmq.proxy.channel.ChannelManager;
 import org.apache.rocketmq.proxy.channel.InvocationChannel;
-import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
 import org.apache.rocketmq.proxy.grpc.adapter.handler.SendMessageResponseHandler;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 public class SendMessageChannel extends InvocationChannel<SendMessageRequest, SendMessageResponse> {
-    private final SendMessageResponseHandler handler;
-
     public SendMessageChannel(SendMessageResponseHandler handler) {
-        super(ChannelManager.createSimpleChannelDirectly());
-        this.handler = handler;
-    }
-
-    @Override
-    public ChannelFuture writeAndFlush(Object msg) {
-        if (msg instanceof RemotingCommand) {
-            RemotingCommand responseCommand = (RemotingCommand) msg;
-            InvocationContext<SendMessageRequest, SendMessageResponse> context = inFlightRequestMap.remove(responseCommand.getOpaque());
-            if (null != context) {
-                handler.handle(responseCommand, context);
-            }
-        }
-
-        return super.writeAndFlush(msg);
+        super(handler);
     }
 }