-
Notifications
You must be signed in to change notification settings - Fork 434
/
GRpcExceptionHandlerInterceptor.java
105 lines (78 loc) · 3.46 KB
/
GRpcExceptionHandlerInterceptor.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
package org.lognet.springboot.grpc.recovery;
import io.grpc.*;
import org.lognet.springboot.grpc.FailureHandlingSupport;
import org.lognet.springboot.grpc.MessageBlockingServerCallListener;
import org.lognet.springboot.grpc.autoconfigure.GRpcServerProperties;
import org.springframework.core.Ordered;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
public class GRpcExceptionHandlerInterceptor implements ServerInterceptor, Ordered {
private final GRpcExceptionHandlerMethodResolver methodResolver;
private final FailureHandlingSupport failureHandlingSupport;
private Integer order;
public GRpcExceptionHandlerInterceptor(GRpcExceptionHandlerMethodResolver methodResolver, FailureHandlingSupport failureHandlingSupport, GRpcServerProperties serverProperties) {
this.methodResolver = methodResolver;
this.failureHandlingSupport = failureHandlingSupport;
this.order = Optional.ofNullable(serverProperties.getRecovery())
.map(GRpcServerProperties.RecoveryProperties::getInterceptorOrder)
.orElse(Ordered.HIGHEST_PRECEDENCE);
}
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
final AtomicBoolean callIsClosed = new AtomicBoolean(false);
if (!methodResolver.hasErrorHandlers()) {
return next.startCall(call, headers);
}
final ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT> errorHandlingCall = new ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
@Override
public void close(Status status, Metadata trailers) {
if( callIsClosed.compareAndSet(false,true)){
super.close(status, trailers);
}
}
@Override
public void sendMessage(RespT message) {
try {
super.sendMessage(message);
} catch (RuntimeException e) {
failureHandlingSupport.closeCall(e, this, headers, b -> b.response(message));
}
}
};
final ServerCall.Listener<ReqT> listener;
try {
listener = next.startCall( errorHandlingCall, headers);
} catch (RuntimeException e) {
failureHandlingSupport.closeCall(e, errorHandlingCall, headers);
return new ServerCall.Listener<ReqT>() {
};
}
return new MessageBlockingServerCallListener<ReqT>(listener) {
private ReqT request;
@Override
public void onMessage(ReqT message) {
try {
request = message;
super.onMessage(message);
} catch (RuntimeException e) {
blockMessage();
failureHandlingSupport.closeCall(e, errorHandlingCall, headers, b -> b.request(request));
}
}
@Override
public void onHalfClose() {
try {
if(!callIsClosed.get()) {
super.onHalfClose();
}
} catch (RuntimeException e) {
failureHandlingSupport.closeCall(e, errorHandlingCall, headers, b -> b.request(request));
}
}
};
}
@Override
public int getOrder() {
return order;
}
}