[ISSUE #3949] Add LocalProxyRelayService implementation

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";
+}

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<StreamObserver<TelemetryCommand>> 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;
     }

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;
+    }
+}

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 <T> Attribute<T> attr(AttributeKey<T> key) {
+        throw new NotImplementedException("Not implemented");
+    }
+
+    @Override
+    public <T> boolean hasAttr(AttributeKey<T> attributeKey) {
+        return false;
+    }
+}

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<ProxyRelayResult<ConsumerRunningInfo>> processGetConsumerRunningInfo(RemotingCommand command,
+    public CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> processGetConsumerRunningInfo(
+        ProxyContext context, RemotingCommand command,
         GetConsumerRunningInfoRequestHeader header) {
         return null;
     }
 
     @Override public CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
-        RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header) {
+        ProxyContext context, RemotingCommand command,
+        ConsumeMessageDirectlyResultRequestHeader header) {
         return null;
     }
 }

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<ProxyRelayResult<ConsumerRunningInfo>> processGetConsumerRunningInfo(RemotingCommand command,
-        GetConsumerRunningInfoRequestHeader header) {
+    public CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> processGetConsumerRunningInfo(
+        ProxyContext context, RemotingCommand command, GetConsumerRunningInfoRequestHeader header) {
         CompletableFuture<ProxyRelayResult<ConsumerRunningInfo>> 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<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(RemotingCommand command,
+    public CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
+        ProxyContext context, RemotingCommand command,
         ConsumeMessageDirectlyResultRequestHeader header) {
         return null;
     }

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<ProxyRelayResult<ConsumerRunningInfo>> 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:

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<ProxyRelayResult<ConsumerRunningInfo>> processGetConsumerRunningInfo(
+        ProxyContext context,
         RemotingCommand command,
         GetConsumerRunningInfoRequestHeader header
     );
 
     CompletableFuture<ProxyRelayResult<ConsumeMessageDirectlyResult>> processConsumeMessageDirectly(
+        ProxyContext context,
         RemotingCommand command,
         ConsumeMessageDirectlyResultRequestHeader header
     );