opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-24 16:04:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] Do refactor some code for readability.

Browse Source
This commit is contained in:
Jixiang.jjx
2022-07-13 11:29:16 +08:00
committed by zhouxiang
parent d051fe9d7d
commit 041bce0d4a
7 changed files with 17 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/channel/ChannelManager.java
+4 -6
View File
@@ -17,7 +17,6 @@
package org.apache.rocketmq.proxy.channel;
import com.google.common.base.Strings;
import io.grpc.Context;
import java.util.ArrayList;
import java.util.Collections;
@@ -28,9 +27,10 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Supplier;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.common.Cleaner;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.adapter.channel.GrpcClientChannel;
import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants;
import org.slf4j.Logger;
@@ -54,14 +54,12 @@ public class ChannelManager {
}
public <T extends SimpleChannel> T createChannel(String clientId, Supplier<T> creator, Class<T> clazz) {
if (Strings.isNullOrEmpty(clientId)) {
if (StringUtils.isBlank(clientId)) {
log.warn("ClientId is unexpected null or empty");
return creator.get();
}
if (!clientIdChannelMap.containsKey(clientId)) {
clientIdChannelMap.putIfAbsent(clientId, creator.get());
}
clientIdChannelMap.computeIfAbsent(clientId, key -> creator.get());
T channel = clazz.cast(clientIdChannelMap.get(clientId));
channel.updateLastAccessTime();
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/GrpcConverter.java
+3 -3
View File
@@ -678,10 +678,10 @@ public class GrpcConverter {
return consumeMessageDirectlyResult;
}
public static Resource buildResource(String resourceNameWithNamespace) {
public static Resource buildResource(String resourceStr) {
return Resource.newBuilder()
.setResourceNamespace(NamespaceUtil.getNamespaceFromResource(resourceNameWithNamespace))
.setName(NamespaceUtil.withoutNamespace(resourceNameWithNamespace))
.setResourceNamespace(NamespaceUtil.getNamespaceFromResource(resourceStr))
.setName(NamespaceUtil.withoutNamespace(resourceStr))
.build();
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/InvocationContext.java
+2 -2
View File
@@ -17,8 +17,8 @@
package org.apache.rocketmq.proxy.grpc.adapter;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
public class InvocationContext<R, W> {
private final R request;
@@ -31,7 +31,7 @@ public class InvocationContext<R, W> {
}
public boolean expired(long expiredTimeSec) {
return System.currentTimeMillis() - timestamp >= TimeUnit.SECONDS.toMillis(expiredTimeSec);
return System.currentTimeMillis() - timestamp >= Duration.ofSeconds(expiredTimeSec).toMillis();
}
public R getRequest() {
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/ResponseWriter.java
+2 -11
View File
@@ -44,27 +44,18 @@ public class ResponseWriter {
}
}
public static <T> void writeException(StreamObserver<?> observer, final Throwable e) {
public static <T> void writeException(StreamObserver<T> observer, final Throwable e) {
if (observer instanceof ServerCallStreamObserver) {
final ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver<T>) observer;
if (null == e) {
return;
}
final ServerCallStreamObserver<T> serverCallStreamObserver = (ServerCallStreamObserver<T>) observer;
if (serverCallStreamObserver.isCancelled()) {
log.warn("Client has cancelled the request. Exception to write", e);
return;
}
// if (e instanceof CompletionException) {
// if (e.getCause() instanceof ProxyException) {
// ProxyException proxyException = (ProxyException) e.getCause();
// serverCallStreamObserver.onNext(ResponseBuilder.buildCommon(proxyException.getCode(), proxyException.getMessage()));
// serverCallStreamObserver.onCompleted();
// return;
// }
// }
log.debug("Start to write error response", e);
serverCallStreamObserver.onError(e);
serverCallStreamObserver.onCompleted();
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/adapter/channel/GrpcClientChannel.java
+2 -3
View File
@@ -53,7 +53,7 @@ public class GrpcClientChannel extends SimpleChannel {
this.manager = manager;
}
public void addClientObserver(CompletableFuture<PollCommandResponse> future) {
public void setClientObserver(CompletableFuture<PollCommandResponse> future) {
this.pollCommandResponseFutureRef.set(future);
}
@@ -131,8 +131,7 @@ public class GrpcClientChannel extends SimpleChannel {
break;
}
case RequestCode.GET_CONSUMER_RUNNING_INFO: {
final GetConsumerRunningInfoRequestHeader requestHeader =
(GetConsumerRunningInfoRequestHeader) command.decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class);
final GetConsumerRunningInfoRequestHeader requestHeader = command.decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class);
if (!requestHeader.isJstackEnable()) {
break;
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/LocalGrpcService.java
+2 -2
View File
@@ -420,7 +420,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo
.build());
break;
}
producerChannel.addClientObserver(future);
producerChannel.setClientObserver(future);
break;
case CONSUMER_GROUP:
Resource consumerGroup = request.getConsumerGroup();
@@ -432,7 +432,7 @@ public class LocalGrpcService extends AbstractStartAndShutdown implements GrpcFo
.build());
break;
}
consumerChannel.addClientObserver(future);
consumerChannel.setClientObserver(future);
break;
default:
break;
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ForwardClientService.java
+2 -2
View File
@@ -145,7 +145,7 @@ public class ForwardClientService extends BaseService {
if (producerChannel == null) {
future.complete(noopCommandResponse);
} else {
producerChannel.addClientObserver(future);
producerChannel.setClientObserver(future);
}
break;
case CONSUMER_GROUP:
@@ -155,7 +155,7 @@ public class ForwardClientService extends BaseService {
if (consumerChannel == null) {
future.complete(noopCommandResponse);
} else {
consumerChannel.addClientObserver(future);
consumerChannel.setClientObserver(future);
}
break;
default: