diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java new file mode 100644 index 0000000000..fcc6bb02ff --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java @@ -0,0 +1,23 @@ +/* + * 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 class ContextVariable { + public final static String REMOTE_ADDRESS = "remote-address"; + public final static String LOCAL_ADDRESS = "local-address"; +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java index eb3e8c9b47..2f629db4cd 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java @@ -51,14 +51,12 @@ public class GrpcClientChannel extends ProxyChannel { private final AtomicReference> telemetryCommandRef = new AtomicReference<>(); private final String group; private final String clientId; - private final String remoteAddress; - private final String localAddress; public GrpcClientChannel(ProxyRelayService proxyRelayService, GrpcChannelManager grpcChannelManager, Context ctx, String group, String clientId) { - super(proxyRelayService, null, new GrpcChannelId(group, clientId)); + super(proxyRelayService, null, new GrpcChannelId(group, clientId), + InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.REMOTE_ADDRESS), + InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LOCAL_ADDRESS)); this.grpcChannelManager = grpcChannelManager; - this.remoteAddress = InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.REMOTE_ADDRESS); - this.localAddress = InterceptorConstants.METADATA.get(ctx).get(InterceptorConstants.LOCAL_ADDRESS); this.group = group; this.clientId = clientId; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java new file mode 100644 index 0000000000..3e2ea4562a --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannel.java @@ -0,0 +1,168 @@ +/* + * 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.service.channel; + +import com.google.common.base.Strings; +import io.netty.channel.AbstractChannel; +import io.netty.channel.Channel; +import io.netty.channel.ChannelConfig; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelMetadata; +import io.netty.channel.ChannelOutboundBuffer; +import io.netty.channel.DefaultChannelPromise; +import io.netty.channel.EventLoop; +import io.netty.util.concurrent.GlobalEventExecutor; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; + +/** + * SimpleChannel is used to handle writeAndFlush situation in processor + * @see io.netty.channel.ChannelHandlerContext#writeAndFlush + * @see io.netty.channel.Channel#writeAndFlush + */ +public class SimpleChannel extends AbstractChannel { + protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + + protected final String remoteAddress; + protected final String localAddress; + + protected long lastAccessTime; + + /** + * Creates a new instance. + * + * @param parent the parent of this channel. {@code null} if there's no parent. + * @param remoteAddress Remote address + * @param localAddress Local address + */ + public SimpleChannel(Channel parent, String remoteAddress, String localAddress) { + super(parent); + lastAccessTime = System.currentTimeMillis(); + this.remoteAddress = remoteAddress; + this.localAddress = localAddress; + } + + public SimpleChannel(String remoteAddress, String localAddress) { + this(null, remoteAddress, localAddress); + } + + @Override + protected AbstractUnsafe newUnsafe() { + return null; + } + + @Override + protected boolean isCompatible(EventLoop loop) { + return false; + } + + private static SocketAddress parseSocketAddress(String address) { + if (Strings.isNullOrEmpty(address)) { + return null; + } + + String[] segments = address.split(":"); + if (2 == segments.length) { + return new InetSocketAddress(segments[0], Integer.parseInt(segments[1])); + } + + return null; + } + + @Override + protected SocketAddress localAddress0() { + return parseSocketAddress(localAddress); + } + + @Override + public SocketAddress localAddress() { + return localAddress0(); + } + + @Override + public SocketAddress remoteAddress() { + return remoteAddress0(); + } + + @Override + protected SocketAddress remoteAddress0() { + return parseSocketAddress(remoteAddress); + } + + @Override + protected void doBind(SocketAddress localAddress) throws Exception { + + } + + @Override + protected void doDisconnect() throws Exception { + + } + + @Override + public ChannelFuture close() { + DefaultChannelPromise promise = new DefaultChannelPromise(this, GlobalEventExecutor.INSTANCE); + promise.setSuccess(); + return promise; + } + + @Override + protected void doClose() throws Exception { + + } + + @Override + protected void doBeginRead() throws Exception { + + } + + @Override + protected void doWrite(ChannelOutboundBuffer in) throws Exception { + + } + + @Override + public ChannelConfig config() { + return null; + } + + @Override + public boolean isOpen() { + return true; + } + + @Override + public boolean isActive() { + return (System.currentTimeMillis() - lastAccessTime) <= 120L * 1000; + } + + @Override + public ChannelMetadata metadata() { + return null; + } + + @Override + public ChannelFuture writeAndFlush(Object msg) { + DefaultChannelPromise promise = new DefaultChannelPromise(this, GlobalEventExecutor.INSTANCE); + promise.setSuccess(); + return promise; + } +} \ No newline at end of file diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelHandlerContext.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelHandlerContext.java new file mode 100644 index 0000000000..801c62ee5f --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/channel/SimpleChannelHandlerContext.java @@ -0,0 +1,246 @@ +/* + * 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.service.channel; + +import io.netty.buffer.ByteBufAllocator; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPipeline; +import io.netty.channel.ChannelProgressivePromise; +import io.netty.channel.ChannelPromise; +import io.netty.util.Attribute; +import io.netty.util.AttributeKey; +import io.netty.util.concurrent.EventExecutor; +import java.net.SocketAddress; +import org.apache.commons.lang3.NotImplementedException; + +public class SimpleChannelHandlerContext implements ChannelHandlerContext { + + private final Channel channel; + + public SimpleChannelHandlerContext(Channel channel) { + this.channel = channel; + } + + @Override + public Channel channel() { + return channel; + } + + @Override + public EventExecutor executor() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public String name() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandler handler() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public boolean isRemoved() { + return false; + } + + @Override + public ChannelHandlerContext fireChannelRegistered() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelUnregistered() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelActive() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelInactive() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireExceptionCaught(Throwable cause) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireUserEventTriggered(Object evt) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelRead(Object msg) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelReadComplete() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext fireChannelWritabilityChanged() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture bind(SocketAddress localAddress) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture connect(SocketAddress remoteAddress) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture disconnect() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture close() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture deregister() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture disconnect(ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture close(ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture deregister(ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext read() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture write(Object msg) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture write(Object msg, ChannelPromise promise) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelHandlerContext flush() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) { + return channel.writeAndFlush(msg, promise); + } + + @Override + public ChannelFuture writeAndFlush(Object msg) { + return channel.writeAndFlush(msg); + } + + @Override + public ChannelPipeline pipeline() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ByteBufAllocator alloc() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelPromise newPromise() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelProgressivePromise newProgressivePromise() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture newSucceededFuture() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelFuture newFailedFuture(Throwable cause) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public ChannelPromise voidPromise() { + throw new NotImplementedException("Not implemented"); + } + + @Override + public Attribute attr(AttributeKey key) { + throw new NotImplementedException("Not implemented"); + } + + @Override + public boolean hasAttr(AttributeKey attributeKey) { + return false; + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java index 4db1a84540..9b356e788b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java @@ -21,6 +21,7 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.common.protocol.header.ConsumeMessageDirectlyResultRequestHeader; import org.apache.rocketmq.common.protocol.header.GetConsumerRunningInfoRequestHeader; +import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.remoting.protocol.RemotingCommand; /** @@ -29,13 +30,15 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class ClusterProxyRelayService implements ProxyRelayService { @Override - public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, + public CompletableFuture> processGetConsumerRunningInfo( + ProxyContext context, RemotingCommand command, GetConsumerRunningInfoRequestHeader header) { return null; } @Override public CompletableFuture> processConsumeMessageDirectly( - RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header) { + ProxyContext context, RemotingCommand command, + ConsumeMessageDirectlyResultRequestHeader header) { return null; } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java index 8ecb6988d2..c39cc3f262 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java @@ -23,6 +23,10 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.common.protocol.header.ConsumeMessageDirectlyResultRequestHeader; import org.apache.rocketmq.common.protocol.header.GetConsumerRunningInfoRequestHeader; +import org.apache.rocketmq.proxy.common.ContextVariable; +import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.service.channel.SimpleChannel; +import org.apache.rocketmq.proxy.service.channel.SimpleChannelHandlerContext; import org.apache.rocketmq.remoting.RemotingServer; import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract; import org.apache.rocketmq.remoting.protocol.RemotingCommand; @@ -36,8 +40,8 @@ public class LocalProxyRelayService implements ProxyRelayService { } @Override - public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, - GetConsumerRunningInfoRequestHeader header) { + public CompletableFuture> processGetConsumerRunningInfo( + ProxyContext context, RemotingCommand command, GetConsumerRunningInfoRequestHeader header) { CompletableFuture> future = new CompletableFuture<>(); future.thenAccept(proxyOutResult -> { if (proxyOutResult.getCode() == ResponseCode.SUCCESS && proxyOutResult.getResult() != null) { @@ -50,8 +54,8 @@ public class LocalProxyRelayService implements ProxyRelayService { ConsumerRunningInfo runningInfo = new ConsumerRunningInfo(); runningInfo.setJstack(consumerRunningInfo.getJstack()); remotingCommand.setBody(runningInfo.encode()); - - // nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(channelManager.createChannel(ctx)), remotingCommand); + SimpleChannel simpleChannel = new SimpleChannel(context.getVal(ContextVariable.REMOTE_ADDRESS), context.getVal(ContextVariable.LOCAL_ADDRESS)); + nettyRemotingAbstract.processResponseCommand(new SimpleChannelHandlerContext(simpleChannel), remotingCommand); } } }); @@ -59,7 +63,8 @@ public class LocalProxyRelayService implements ProxyRelayService { } @Override - public CompletableFuture> processConsumeMessageDirectly(RemotingCommand command, + public CompletableFuture> processConsumeMessageDirectly( + ProxyContext context, RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header) { return null; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java index 153ffb709f..157bcd63ba 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java @@ -41,22 +41,30 @@ import org.apache.rocketmq.common.protocol.header.ConsumeMessageDirectlyResultRe import org.apache.rocketmq.common.protocol.header.GetConsumerRunningInfoRequestHeader; import org.apache.rocketmq.logging.InternalLogger; import org.apache.rocketmq.logging.InternalLoggerFactory; +import org.apache.rocketmq.proxy.common.ContextVariable; +import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.service.transaction.TransactionId; import org.apache.rocketmq.remoting.protocol.RemotingCommand; public abstract class ProxyChannel extends AbstractChannel { private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + protected final String remoteAddress; + protected final String localAddress; protected final ProxyRelayService proxyRelayService; - protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent) { + protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, String remoteAddress, String localAddress) { super(parent); this.proxyRelayService = proxyRelayService; + this.remoteAddress = remoteAddress; + this.localAddress = localAddress; } - protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id) { + protected ProxyChannel(ProxyRelayService proxyRelayService, Channel parent, ChannelId id, String remoteAddress, String localAddress) { super(parent, id); this.proxyRelayService = proxyRelayService; + this.remoteAddress = remoteAddress; + this.localAddress = localAddress; } @Override @@ -65,6 +73,9 @@ public abstract class ProxyChannel extends AbstractChannel { try { if (msg instanceof RemotingCommand) { + ProxyContext context = ProxyContext.create() + .withVal(ContextVariable.REMOTE_ADDRESS, remoteAddress) + .withVal(ContextVariable.REMOTE_ADDRESS, localAddress); RemotingCommand command = (RemotingCommand) msg; switch (command.getCode()) { case RequestCode.CHECK_TRANSACTION_STATE: { @@ -77,15 +88,15 @@ public abstract class ProxyChannel extends AbstractChannel { } case RequestCode.GET_CONSUMER_RUNNING_INFO: { GetConsumerRunningInfoRequestHeader header = (GetConsumerRunningInfoRequestHeader) command.readCustomHeader(); - processFuture = this.processGetConsumerRunningInfo(command, header, - this.proxyRelayService.processGetConsumerRunningInfo(command, header)); + CompletableFuture> relayFuture = this.proxyRelayService.processGetConsumerRunningInfo(context, command, header); + processFuture = this.processGetConsumerRunningInfo(command, header, relayFuture); break; } case RequestCode.CONSUME_MESSAGE_DIRECTLY: { ConsumeMessageDirectlyResultRequestHeader header = (ConsumeMessageDirectlyResultRequestHeader) command.readCustomHeader(); MessageExt messageExt = MessageDecoder.decode(ByteBuffer.wrap(command.getBody()), true, false, false); processFuture = this.processConsumeMessageDirectly(command, header, messageExt, - this.proxyRelayService.processConsumeMessageDirectly(command, header)); + this.proxyRelayService.processConsumeMessageDirectly(context, command, header)); break; } default: diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java index 1e800206a3..b7284da1c0 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java @@ -21,16 +21,19 @@ import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo; import org.apache.rocketmq.common.protocol.header.ConsumeMessageDirectlyResultRequestHeader; import org.apache.rocketmq.common.protocol.header.GetConsumerRunningInfoRequestHeader; +import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.remoting.protocol.RemotingCommand; public interface ProxyRelayService { CompletableFuture> processGetConsumerRunningInfo( + ProxyContext context, RemotingCommand command, GetConsumerRunningInfoRequestHeader header ); CompletableFuture> processConsumeMessageDirectly( + ProxyContext context, RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header );