1 首先定義ClientInterceptor
public HeaderClientInterceptor(String clientId, String accessToken) {
this.clientId = clientId;
this.accessToken = accessToken;
}
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
public void start(Listener<RespT> responseListener, Metadata headers) {
headers.put(CLIENT_ID, clientId);
if (StringUtils.isNotEmpty(accessToken)) {
headers.put(ACCESS_TOKEN, accessToken);
}
super.start(new SimpleForwardingClientCallListener<RespT>(responseListener) {
/**
* {@inheritDoc}
* @see io.grpc.ForwardingClientCallListener#onClose(Status, Metadata)
*/
@Override
public void onClose(Status status, Metadata trailers) {
CLOG.debug("關(guān)閉原因:{}", status.getCode().toString());
CLOG.debug("獲取請求返回頭部信息,{}", trailers.toString());
String isexpire = trailers.get(IS_EXPIRE);
if (Boolean.getBoolean(isexpire)) {
OAuthService oAuthService = SpringContextUtils.getBean(OAuthService.class);
String accessToken = oAuthService.oauth();
CLOG.debug("refresh Token,{}", accessToken);
}
super.onClose(status, trailers);
}
@Override
public void onHeaders(Metadata headers) {
CLOG.debug("獲取請求返回頭部信息,{}", headers.toString());
super.onHeaders(headers);
}
}, headers);
}
};
}
2 Auth,項(xiàng)目中做成了切面。
public String verifying() {
ManagedChannel channel = channelPool.getManagedChannel();
OAuthBlockingStub blockingStub = OAuthGrpc.newBlockingStub(channel)
.withInterceptors(new HeaderClientInterceptor(clientProp.getClientId(), null));
OAuthRequest oAuthRequest = OAuthRequest.newBuilder().setClientId(clientProp.getClientId())
.setClientSecret(clientProp.getSecret()).setGrantType(clientProp.getGrant()).build();
CLOG.info("OAuthRequest: " + JsonUtils.toJson(oAuthRequest), tagMap);
OAuthResponse oauthResponse = null;
try {
oauthResponse = blockingStub.getAccessToken(oAuthRequest);
RedisHelper.setHNATokenAndEffectTime(oauthResponse.getAccessToken(), new Date());
} catch (StatusRuntimeException e) {
CLOG.info(e, tagMap);
return "";
} finally {
if (null != channel) {
channelPool.returnObject(channel);
}
}
return oauthResponse.getAccessToken();
}
3 具體clientStub調(diào)用
public AirLowFareSearchRS airLowFareSearch(AirLowFareSearchRQ airLowFareSearchRQ) {
ManagedChannel channel = channelPool.getManagedChannel();
Map<String, String> tagMap = this.tagMap(airLowFareSearchRQ);
AirLowFareSearchBlockingStub blockingStub = AirLowFareSearchGrpc.newBlockingStub(channel)
.withInterceptors(new HeaderClientInterceptor(clientProp.getClientId(), RedisHelper.getToken()));
CLOG.info("AirLowFareSearchRQ :" + JsonUtils.toJson(airLowFareSearchRQ), tagMap);
AirLowFareSearchRS airLowFareSearchRS = null;
try {
airLowFareSearchRS = blockingStub.airLowFareSearch(airLowFareSearchRQ);
CLOG.info("AirLowFareSearchRS :" + JsonUtils.toJson(airLowFareSearchRS), tagMap);
} catch (Exception e) {
CLOG.info(e, tagMap);
ExceptionHandler.tokenUselessToEmpty(e);
} finally {
if (null != channel) {
channelPool.returnObject(channel);
}
}
return airLowFareSearchRS;
}
4 深入源碼 (調(diào)了半天有點(diǎn)暈,下次再續(xù))
/**
* Executes a unary call and returns a {@link ListenableFuture} to the response.
*
* @return a future for the single response message.
*/
public static <ReqT, RespT> ListenableFuture<RespT> futureUnaryCall(
ClientCall<ReqT, RespT> call,
ReqT param) {
GrpcFuture<RespT> responseFuture = new GrpcFuture<RespT>(call);
asyncUnaryRequestCall(call, param, new UnaryStreamToFuture<RespT>(responseFuture), false);
return responseFuture;
}
最后編輯于 :2018.05.04 16:04:41
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者 【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成,瀏覽時(shí)請結(jié)合常識與多方信息審慎甄別。 平臺聲明:文章內(nèi)容(如有圖片或視頻亦包括在內(nèi))由作者上傳并發(fā)布,文章內(nèi)容僅代表作者本人觀點(diǎn),簡書系信息發(fā)布平臺,僅提供信息存儲服務(wù)。