[ISSUE #3949] Refector SimpleChannel

2026-09-24 16:04:00 +08:00 · 2022-07-13 11:29:09 +08:00
parent 8d924e9e69
commit 115030a80f
7 changed files with 144 additions and 88 deletions
@@ -15,7 +15,7 @@
 * limitations under the License.
 */

-package org.apache.rocketmq.proxy.grpc.adapter.channel;
+package org.apache.rocketmq.proxy.channel;

 import com.google.common.base.Strings;
 import io.grpc.Context;
@@ -25,16 +25,16 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.proxy.configuration.ConfigurationManager;
+import org.apache.rocketmq.proxy.grpc.adapter.channel.SendMessageChannel;
 import org.apache.rocketmq.proxy.grpc.common.InterceptorConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-public class ChannelManager<R, W> {
+public class ChannelManager {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME);
+    private final ConcurrentMap<String, SimpleChannel> clientIdChannelMap = new ConcurrentHashMap<>();

-    private final ConcurrentMap<String, SimpleChannel<R, W>> clientIdChannelMap = new ConcurrentHashMap<>();
-
-    public SimpleChannel<R, W> createChannel() {
+    public SimpleChannel createChannel() {
        final String clientId = anonymousChannelId();
        if (Strings.isNullOrEmpty(clientId)) {
            LOGGER.warn("ClientId is unexpected null or empty");
@@ -45,8 +45,9 @@ public class ChannelManager<R, W> {
            clientIdChannelMap.putIfAbsent(clientId, createChannelInner());
        }

-        return clientIdChannelMap.get(clientId)
-            .updateLastAccessTime();
+        SimpleChannel channel = clientIdChannelMap.get(clientId);
+        channel.updateLastAccessTime();
+        return channel;
    }

    private String anonymousChannelId() {
@@ -57,12 +58,12 @@ public class ChannelManager<R, W> {
        return clientHost + "@" + localAddress;
    }

-    private SimpleChannel<R, W> createChannelInner() {
+    private SimpleChannel createChannelInner() {
        final String clientHost = InterceptorConstants.METADATA.get(Context.current())
            .get(InterceptorConstants.REMOTE_ADDRESS);
        final String localAddress = InterceptorConstants.METADATA.get(Context.current())
            .get(InterceptorConstants.LOCAL_ADDRESS);
-        return new SimpleChannel<>(null, clientHost, localAddress, ConfigurationManager.getProxyConfig().getExpiredChannelTimeSec());
+        return new SimpleChannel(null, clientHost, localAddress, ConfigurationManager.getProxyConfig().getExpiredChannelTimeSec());
    }

    /**
@@ -70,16 +71,18 @@ public class ChannelManager<R, W> {
     */
    public void scanAndCleanChannels() {
        try {
-            Iterator<Map.Entry<String, SimpleChannel<R, W>>> iterator = clientIdChannelMap.entrySet()
+            Iterator<Map.Entry<String, SimpleChannel>> iterator = clientIdChannelMap.entrySet()
                .iterator();
            while (iterator.hasNext()) {
-                Map.Entry<String, SimpleChannel<R, W>> entry = iterator.next();
+                Map.Entry<String, SimpleChannel> entry = iterator.next();
                if (!entry.getValue()
                    .isActive()) {
                    iterator.remove();
                } else {
-                    entry.getValue()
-                        .cleanExpiredRequests();
+                    if (entry.getValue() instanceof SendMessageChannel) {
+                        SendMessageChannel channel = (SendMessageChannel) entry.getValue();
+                        channel.cleanExpiredRequests();
+                    }
                }
            }
        } catch (Throwable e) {
@@ -0,0 +1,83 @@
+/*
+ * 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.channel;
+
+import io.netty.channel.ChannelFuture;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public class InvocationChannel<R, W> extends SimpleChannel {
+    protected final ConcurrentMap<Integer, InvocationContext<R, W>> inFlightRequestMap;
+
+    public InvocationChannel(SimpleChannel simpleChannel) {
+        super(simpleChannel);
+        this.inFlightRequestMap = new ConcurrentHashMap<>();
+    }
+
+    @Override
+    public ChannelFuture writeAndFlush(Object msg) {
+        if (msg instanceof RemotingCommand) {
+            RemotingCommand responseCommand = (RemotingCommand) msg;
+            inFlightRequestMap.remove(responseCommand.getOpaque());
+        }
+        return super.writeAndFlush(msg);
+    }
+
+    public boolean isWritable(int opaque) {
+        if (!inFlightRequestMap.containsKey(opaque)) {
+            return false;
+        }
+
+        InvocationContext<R, W> invocationContext = inFlightRequestMap.get(opaque);
+        if (null != invocationContext) {
+            CompletableFuture<?> future = invocationContext.getResponse();
+            return null != future && !future.isCancelled() && !future.isCompletedExceptionally() && !future.isDone();
+        }
+        return false;
+    }
+
+    public void registerInvocationContext(int opaque, InvocationContext<R, W> context) {
+        inFlightRequestMap.put(opaque, context);
+    }
+
+    public void eraseInvocationContext(int opaque) {
+        inFlightRequestMap.remove(opaque);
+    }
+
+    public void cleanExpiredRequests() {
+        Iterator<Map.Entry<Integer, InvocationContext<R, W>>> iterator = inFlightRequestMap.entrySet().iterator();
+        int count = 0;
+        while (iterator.hasNext()) {
+            Map.Entry<Integer, InvocationContext<R, W>> entry = iterator.next();
+            if (entry.getValue().expired(expiredTimeSec)) {
+                iterator.remove();
+                count++;
+                LOGGER.debug("An expired request is found, created time-point: {}, Request: {}",
+                    entry.getValue().getTimestamp(), entry.getValue().getRequest());
+            }
+        }
+        if (count > 0) {
+            LOGGER.warn("[BUG] {} expired in-flight requests is cleaned.", count);
+        }
+    }
+}
@@ -15,7 +15,7 @@
 * limitations under the License.
 */

-package org.apache.rocketmq.proxy.grpc.adapter.channel;
+package org.apache.rocketmq.proxy.channel;

 import com.google.common.base.Strings;
 import io.netty.channel.AbstractChannel;
@@ -29,14 +29,7 @@ import io.netty.channel.EventLoop;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import org.apache.rocketmq.common.constant.LoggerName;
-import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

@@ -45,17 +38,15 @@ import org.slf4j.LoggerFactory;
 * @see io.netty.channel.ChannelHandlerContext#writeAndFlush
 * @see io.netty.channel.Channel#writeAndFlush
 */
-public class SimpleChannel<R, W> extends AbstractChannel {
+public class SimpleChannel extends AbstractChannel {

-    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME);
+    protected static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME);

-    private final String remoteAddress;
-    private final String localAddress;
-    private final long expiredTimeSec;
+    protected final String remoteAddress;
+    protected final String localAddress;
+    protected final long expiredTimeSec;

-    private long lastAccessTime;
-
-    protected final ConcurrentMap<Integer, InvocationContext<R, W>> inFlightRequestMap;
+    protected long lastAccessTime;

    /**
     * Creates a new instance.
@@ -70,16 +61,14 @@ public class SimpleChannel<R, W> extends AbstractChannel {
        lastAccessTime = System.currentTimeMillis();
        this.remoteAddress = remoteAddress;
        this.localAddress = localAddress;
-        this.inFlightRequestMap = new ConcurrentHashMap<>();
        this.expiredTimeSec = expiredTimeSec;
    }

-    public SimpleChannel(SimpleChannel<R, W> other) {
+    public SimpleChannel(SimpleChannel other) {
        super(other);
-        lastAccessTime = System.currentTimeMillis();
+        lastAccessTime = other.lastAccessTime;
        this.remoteAddress = other.remoteAddress;
        this.localAddress = other.localAddress;
-        this.inFlightRequestMap = other.inFlightRequestMap;
        this.expiredTimeSec = other.expiredTimeSec;
    }

@@ -158,19 +147,6 @@ public class SimpleChannel<R, W> extends AbstractChannel {

    }

-    public boolean isWritable(int opaque) {
-        if (!inFlightRequestMap.containsKey(opaque)) {
-            return false;
-        }
-
-        InvocationContext<R, W> invocationContext = inFlightRequestMap.get(opaque);
-        if (null != invocationContext) {
-            CompletableFuture<?> future = invocationContext.getResponse();
-            return null != future && !future.isCancelled() && !future.isCompletedExceptionally() && !future.isDone();
-        }
-        return false;
-    }
-
    @Override
    public ChannelConfig config() {
        return null;
@@ -198,44 +174,13 @@ public class SimpleChannel<R, W> extends AbstractChannel {

    @Override
    public ChannelFuture writeAndFlush(Object msg) {
-        if (msg instanceof RemotingCommand) {
-            RemotingCommand responseCommand = (RemotingCommand) msg;
-            inFlightRequestMap.remove(responseCommand.getOpaque());
-        }
-
        DefaultChannelPromise promise = new DefaultChannelPromise(this, GlobalEventExecutor.INSTANCE);
        promise.setSuccess();
        return promise;
    }

-    public void registerInvocationContext(int opaque, InvocationContext<R, W> context) {
-        inFlightRequestMap.put(opaque, context);
-    }
-
-    public void eraseInvocationContext(int opaque) {
-        inFlightRequestMap.remove(opaque);
-    }
-
-    public void cleanExpiredRequests() {
-        Iterator<Map.Entry<Integer, InvocationContext<R, W>>> iterator = inFlightRequestMap.entrySet().iterator();
-        int count = 0;
-        while (iterator.hasNext()) {
-            Map.Entry<Integer, InvocationContext<R, W>> entry = iterator.next();
-            if (entry.getValue().expired(expiredTimeSec)) {
-                iterator.remove();
-                count++;
-                LOGGER.debug("An expired request is found, created time-point: {}, Request: {}",
-                    entry.getValue().getTimestamp(), entry.getValue().getRequest());
-            }
-        }
-        if (count > 0) {
-            LOGGER.warn("[BUG] {} expired in-flight requests is cleaned.", count);
-        }
-    }
-
-    public SimpleChannel<R, W> updateLastAccessTime() {
+    public void updateLastAccessTime() {
        lastAccessTime = System.currentTimeMillis();
-        return this;
    }
 }

@@ -15,7 +15,7 @@
 * limitations under the License.
 */

-package org.apache.rocketmq.proxy.grpc.adapter.channel;
+package org.apache.rocketmq.proxy.channel;

 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
@@ -17,22 +17,38 @@

 package org.apache.rocketmq.proxy.grpc;

+import apache.rocketmq.v1.HealthCheckRequest;
+import apache.rocketmq.v1.HealthCheckResponse;
+import apache.rocketmq.v1.HeartbeatRequest;
+import apache.rocketmq.v1.HeartbeatResponse;
 import apache.rocketmq.v1.MessagingServiceGrpc;
 import apache.rocketmq.v1.SendMessageRequest;
 import apache.rocketmq.v1.SendMessageResponse;
 import io.grpc.Context;
 import io.grpc.stub.StreamObserver;
 import java.util.concurrent.CompletableFuture;
+import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.proxy.grpc.common.ResponseWriter;
 import org.apache.rocketmq.proxy.grpc.service.GrpcForwardService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;

 public class GrpcMessagingProcessor extends MessagingServiceGrpc.MessagingServiceImplBase {
+    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME);
    private final GrpcForwardService grpcForwardService;

    public GrpcMessagingProcessor(GrpcForwardService grpcForwardService) {
        this.grpcForwardService = grpcForwardService;
    }

+    public void heartbeat(HeartbeatRequest request, StreamObserver<HeartbeatResponse> responseObserver) {
+
+    }
+
+    @Override
+    public void healthCheck(HealthCheckRequest request, StreamObserver<HealthCheckResponse> responseObserver) {
+    }
+
    @Override
    public void sendMessage(SendMessageRequest request, StreamObserver<SendMessageResponse> responseObserver) {
        CompletableFuture<SendMessageResponse> future = grpcForwardService.sendMessage(Context.current(), request);
@@ -20,18 +20,20 @@ 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.InvocationChannel;
+import org.apache.rocketmq.proxy.channel.SimpleChannel;
 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 SimpleChannel<SendMessageRequest, SendMessageResponse> {
+public class SendMessageChannel extends InvocationChannel<SendMessageRequest, SendMessageResponse> {
    private final SendMessageResponseHandler handler;

-    public static SendMessageChannel create(SimpleChannel<SendMessageRequest, SendMessageResponse> other, SendMessageResponseHandler handler) {
-        return new SendMessageChannel(other, handler);
+    public static SendMessageChannel create(SimpleChannel simpleChannel, SendMessageResponseHandler handler) {
+        return new SendMessageChannel(simpleChannel, handler);
    }

-    private SendMessageChannel(SimpleChannel<SendMessageRequest, SendMessageResponse> other, SendMessageResponseHandler handler) {
+    private SendMessageChannel(SimpleChannel other, SendMessageResponseHandler handler) {
        super(other);
        this.handler = handler;
    }
@@ -52,6 +52,7 @@ import apache.rocketmq.v1.ReportThreadStackTraceRequest;
 import apache.rocketmq.v1.ReportThreadStackTraceResponse;
 import apache.rocketmq.v1.SendMessageRequest;
 import apache.rocketmq.v1.SendMessageResponse;
+import com.google.rpc.Code;
 import io.grpc.Context;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
@@ -63,11 +64,12 @@ import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.RequestCode;
 import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
 import org.apache.rocketmq.proxy.grpc.adapter.InvocationContext;
-import org.apache.rocketmq.proxy.grpc.adapter.channel.ChannelManager;
+import org.apache.rocketmq.proxy.channel.ChannelManager;
 import org.apache.rocketmq.proxy.grpc.adapter.channel.SendMessageChannel;
-import org.apache.rocketmq.proxy.grpc.adapter.channel.SimpleChannelHandlerContext;
+import org.apache.rocketmq.proxy.channel.SimpleChannelHandlerContext;
 import org.apache.rocketmq.proxy.grpc.adapter.handler.SendMessageResponseHandler;
 import org.apache.rocketmq.proxy.grpc.common.Converter;
+import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -78,11 +80,11 @@ public class LocalGrpcService implements GrpcForwardService {
    private final BrokerController brokerController;
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
        new ThreadFactoryImpl("LocalGrpcServiceScheduledThread"));
-    private final ChannelManager<SendMessageRequest, SendMessageResponse> sendChannelManager;
+    private final ChannelManager sendChannelManager;

    public LocalGrpcService(BrokerController brokerController) {
        this.brokerController = brokerController;
-        this.sendChannelManager = new ChannelManager<>();
+        this.sendChannelManager = new ChannelManager();
    }

    @Override public CompletableFuture<QueryRouteResponse> queryRoute(Context ctx, QueryRouteRequest request) {
@@ -90,11 +92,16 @@ public class LocalGrpcService implements GrpcForwardService {
    }

    @Override public CompletableFuture<HeartbeatResponse> heartbeat(Context ctx, HeartbeatRequest request) {
+
        return null;
    }

    @Override public CompletableFuture<HealthCheckResponse> healthCheck(Context ctx, HealthCheckRequest request) {
-        return null;
+        LOGGER.trace("Received health check request from client: {}", request.getClientHost());
+        final HealthCheckResponse response = HealthCheckResponse.newBuilder()
+            .setCommon(ResponseBuilder.buildCommon(Code.OK, "ok"))
+            .build();
+        return CompletableFuture.completedFuture(response);
    }

    @Override