mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-24 16:04:00 +08:00
[ISSUE #3949] Interceptor
* Add GlobalExceptionInterceptor * Catch AclException in AuthenticationInterceptor
This commit is contained in:
zhouxiang
@@ -45,6 +45,7 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager;
|
||||
import org.apache.rocketmq.proxy.config.ProxyConfig;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.GlobalExceptionInterceptor;
|
||||
import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor;
|
||||
|
||||
public class GrpcServerBuilder {
|
||||
@@ -150,6 +151,7 @@ public class GrpcServerBuilder {
|
||||
}
|
||||
|
||||
this.serverBuilder
|
||||
.intercept(new GlobalExceptionInterceptor())
|
||||
.intercept(new ContextInterceptor())
|
||||
.intercept(new HeaderInterceptor());
|
||||
|
||||
|
||||
+25
-18
@@ -24,9 +24,12 @@ import io.grpc.Metadata;
|
||||
import io.grpc.ServerCall;
|
||||
import io.grpc.ServerCallHandler;
|
||||
import io.grpc.ServerInterceptor;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
import java.util.List;
|
||||
import org.apache.rocketmq.acl.AccessResource;
|
||||
import org.apache.rocketmq.acl.AccessValidator;
|
||||
import org.apache.rocketmq.acl.common.AclException;
|
||||
import org.apache.rocketmq.acl.common.MetadataHeader;
|
||||
import org.apache.rocketmq.acl.plain.PlainAccessResource;
|
||||
import org.apache.rocketmq.proxy.grpc.v2.adapter.RequestMapping;
|
||||
@@ -44,25 +47,29 @@ public class AuthenticationInterceptor implements ServerInterceptor {
|
||||
return new ForwardingServerCallListener.SimpleForwardingServerCallListener<R>(next.startCall(call, headers)) {
|
||||
@Override
|
||||
public void onMessage(R message) {
|
||||
GeneratedMessageV3 messageV3 = (GeneratedMessageV3) message;
|
||||
MetadataHeader metadataHeader = MetadataHeader.builder()
|
||||
.remoteAddress(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REMOTE_ADDRESS))
|
||||
.namespace(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.NAMESPACE_ID))
|
||||
.authorization(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.AUTHORIZATION))
|
||||
.datetime(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.DATE_TIME))
|
||||
.sessionToken(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.SESSION_TOKEN))
|
||||
.requestId(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REQUEST_ID))
|
||||
.language(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.LANGUAGE))
|
||||
.clientVersion(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.CLIENT_VERSION))
|
||||
.protocol(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.PROTOCOL_VERSION))
|
||||
.requestCode(RequestMapping.map(messageV3.getDescriptorForType().getFullName()))
|
||||
.build();
|
||||
for (AccessValidator accessValidator : accessValidatorList) {
|
||||
AccessResource accessResource = accessValidator.parse(messageV3, metadataHeader);
|
||||
accessValidator.validate(accessResource);
|
||||
addHeader(headers, messageV3, accessResource);
|
||||
try {
|
||||
GeneratedMessageV3 messageV3 = (GeneratedMessageV3) message;
|
||||
MetadataHeader metadataHeader = MetadataHeader.builder()
|
||||
.remoteAddress(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REMOTE_ADDRESS))
|
||||
.namespace(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.NAMESPACE_ID))
|
||||
.authorization(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.AUTHORIZATION))
|
||||
.datetime(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.DATE_TIME))
|
||||
.sessionToken(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.SESSION_TOKEN))
|
||||
.requestId(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REQUEST_ID))
|
||||
.language(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.LANGUAGE))
|
||||
.clientVersion(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.CLIENT_VERSION))
|
||||
.protocol(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.PROTOCOL_VERSION))
|
||||
.requestCode(RequestMapping.map(messageV3.getDescriptorForType().getFullName()))
|
||||
.build();
|
||||
for (AccessValidator accessValidator : accessValidatorList) {
|
||||
AccessResource accessResource = accessValidator.parse(messageV3, metadataHeader);
|
||||
accessValidator.validate(accessResource);
|
||||
addHeader(headers, messageV3, accessResource);
|
||||
}
|
||||
super.onMessage(message);
|
||||
} catch (AclException aclException) {
|
||||
throw new StatusRuntimeException(Status.PERMISSION_DENIED, headers);
|
||||
}
|
||||
super.onMessage(message);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
+128
@@ -0,0 +1,128 @@
|
||||
/*
|
||||
* 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.grpc.interceptor;
|
||||
|
||||
import io.grpc.ForwardingServerCall;
|
||||
import io.grpc.ForwardingServerCallListener;
|
||||
import io.grpc.Metadata;
|
||||
import io.grpc.ServerCall;
|
||||
import io.grpc.ServerCallHandler;
|
||||
import io.grpc.ServerInterceptor;
|
||||
import io.grpc.Status;
|
||||
import io.grpc.StatusRuntimeException;
|
||||
import org.apache.rocketmq.common.constant.LoggerName;
|
||||
import org.apache.rocketmq.logging.InternalLogger;
|
||||
import org.apache.rocketmq.logging.InternalLoggerFactory;
|
||||
|
||||
public class GlobalExceptionInterceptor implements ServerInterceptor {
|
||||
private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME);
|
||||
|
||||
@Override
|
||||
public <R, W> ServerCall.Listener<R> interceptCall(
|
||||
ServerCall<R, W> call,
|
||||
Metadata headers,
|
||||
ServerCallHandler<R, W> next
|
||||
) {
|
||||
final ServerCall<R, W> serverCall = new ClosableServerCall<>(call);
|
||||
ServerCall.Listener<R> delegate = next.startCall(serverCall, headers);
|
||||
return new ForwardingServerCallListener.SimpleForwardingServerCallListener<R>(delegate) {
|
||||
@Override
|
||||
public void onMessage(R message) {
|
||||
try {
|
||||
super.onMessage(message);
|
||||
} catch (Throwable e) {
|
||||
closeWithException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onHalfClose() {
|
||||
try {
|
||||
super.onHalfClose();
|
||||
} catch (Throwable e) {
|
||||
closeWithException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCancel() {
|
||||
try {
|
||||
super.onCancel();
|
||||
} catch (Throwable e) {
|
||||
closeWithException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onComplete() {
|
||||
try {
|
||||
super.onComplete();
|
||||
} catch (Throwable e) {
|
||||
closeWithException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onReady() {
|
||||
try {
|
||||
super.onReady();
|
||||
} catch (Throwable e) {
|
||||
closeWithException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private void closeWithException(Throwable t) {
|
||||
Metadata trailers = new Metadata();
|
||||
Status status = Status.INTERNAL.withDescription(t.getMessage());
|
||||
boolean printLog = true;
|
||||
|
||||
if (t instanceof StatusRuntimeException) {
|
||||
trailers = ((StatusRuntimeException) t).getTrailers();
|
||||
status = ((StatusRuntimeException) t).getStatus();
|
||||
// no error stack for permission denied.
|
||||
if (status.getCode().value() == Status.PERMISSION_DENIED.getCode().value()) {
|
||||
printLog = false;
|
||||
}
|
||||
}
|
||||
|
||||
if (printLog) {
|
||||
log.error("grpc server has exception. errorMsg:{}, e:", t.getMessage(), t);
|
||||
}
|
||||
|
||||
serverCall.close(status, trailers);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private static class ClosableServerCall<R, W> extends
|
||||
ForwardingServerCall.SimpleForwardingServerCall<R, W> {
|
||||
private boolean closeCalled = false;
|
||||
|
||||
ClosableServerCall(ServerCall<R, W> delegate) {
|
||||
super(delegate);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void close(final Status status, final Metadata trailers) {
|
||||
if (!closeCalled) {
|
||||
closeCalled = true;
|
||||
ClosableServerCall.super.close(status, trailers);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user