diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/ChannelManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java similarity index 73% rename from proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/ChannelManager.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java index 66f8d86275..2c9ff07d9a 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/ChannelManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java @@ -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 { +public class ChannelManager { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME); + private final ConcurrentMap clientIdChannelMap = new ConcurrentHashMap<>(); - private final ConcurrentMap> clientIdChannelMap = new ConcurrentHashMap<>(); - - public SimpleChannel 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 { 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 { return clientHost + "@" + localAddress; } - private SimpleChannel 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 { */ public void scanAndCleanChannels() { try { - Iterator>> iterator = clientIdChannelMap.entrySet() + Iterator> iterator = clientIdChannelMap.entrySet() .iterator(); while (iterator.hasNext()) { - Map.Entry> entry = iterator.next(); + Map.Entry 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) { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java new file mode 100644 index 0000000000..4c70cc74c8 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/InvocationChannel.java @@ -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 extends SimpleChannel { + protected final ConcurrentMap> 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 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 context) { + inFlightRequestMap.put(opaque, context); + } + + public void eraseInvocationContext(int opaque) { + inFlightRequestMap.remove(opaque); + } + + public void cleanExpiredRequests() { + Iterator>> iterator = inFlightRequestMap.entrySet().iterator(); + int count = 0; + while (iterator.hasNext()) { + Map.Entry> 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); + } + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannel.java similarity index 63% rename from proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannel.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannel.java index eb170ffbc1..a7305e1c92 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannel.java @@ -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 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> inFlightRequestMap; + protected long lastAccessTime; /** * Creates a new instance. @@ -70,16 +61,14 @@ public class SimpleChannel extends AbstractChannel { lastAccessTime = System.currentTimeMillis(); this.remoteAddress = remoteAddress; this.localAddress = localAddress; - this.inFlightRequestMap = new ConcurrentHashMap<>(); this.expiredTimeSec = expiredTimeSec; } - public SimpleChannel(SimpleChannel 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 extends AbstractChannel { } - public boolean isWritable(int opaque) { - if (!inFlightRequestMap.containsKey(opaque)) { - return false; - } - - InvocationContext 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 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 context) { - inFlightRequestMap.put(opaque, context); - } - - public void eraseInvocationContext(int opaque) { - inFlightRequestMap.remove(opaque); - } - - public void cleanExpiredRequests() { - Iterator>> iterator = inFlightRequestMap.entrySet().iterator(); - int count = 0; - while (iterator.hasNext()) { - Map.Entry> 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 updateLastAccessTime() { + public void updateLastAccessTime() { lastAccessTime = System.currentTimeMillis(); - return this; } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannelHandlerContext.java b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannelHandlerContext.java similarity index 99% rename from proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannelHandlerContext.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannelHandlerContext.java index 62d0c2a543..a78c5727f7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SimpleChannelHandlerContext.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/channel/SimpleChannelHandlerContext.java @@ -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; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcMessagingProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcMessagingProcessor.java index dfa2006fe4..69edec9c96 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcMessagingProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcMessagingProcessor.java @@ -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 responseObserver) { + + } + + @Override + public void healthCheck(HealthCheckRequest request, StreamObserver responseObserver) { + } + @Override public void sendMessage(SendMessageRequest request, StreamObserver responseObserver) { CompletableFuture future = grpcForwardService.sendMessage(Context.current(), request); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SendMessageChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SendMessageChannel.java index 646df00fc7..08894d6fd3 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SendMessageChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/SendMessageChannel.java @@ -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 { +public class SendMessageChannel extends InvocationChannel { private final SendMessageResponseHandler handler; - public static SendMessageChannel create(SimpleChannel other, SendMessageResponseHandler handler) { - return new SendMessageChannel(other, handler); + public static SendMessageChannel create(SimpleChannel simpleChannel, SendMessageResponseHandler handler) { + return new SendMessageChannel(simpleChannel, handler); } - private SendMessageChannel(SimpleChannel other, SendMessageResponseHandler handler) { + private SendMessageChannel(SimpleChannel other, SendMessageResponseHandler handler) { super(other); this.handler = handler; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java index 329fe71c62..a2e140d91d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java @@ -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 sendChannelManager; + private final ChannelManager sendChannelManager; public LocalGrpcService(BrokerController brokerController) { this.brokerController = brokerController; - this.sendChannelManager = new ChannelManager<>(); + this.sendChannelManager = new ChannelManager(); } @Override public CompletableFuture queryRoute(Context ctx, QueryRouteRequest request) { @@ -90,11 +92,16 @@ public class LocalGrpcService implements GrpcForwardService { } @Override public CompletableFuture heartbeat(Context ctx, HeartbeatRequest request) { + return null; } @Override public CompletableFuture 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