I am new to spring integration.
I have a flow on which I need to perform an http or a tcp call depending on some conditions.
The problem I am focused on is related to the http call.
The rest endpoint called needs an accessToken as header parameter for authentication that is provided by a spring service that has 2 methods getCurrentAccessToken() and refreshAccessToken(). I want to call the method refresh accessToken only when the currentAccessToken is expired.
What I would like to do is to add the following logic when performing the call to the rest api:
If the token is expired the rest endpoint returns a 401 and I would like to intercept in the flow this error and retry the request by adding a refreshed access token.
@Bean
public IntegrationFlow clientIn(AbstractServerConnectionFactory server,
AbstractClientConnectionFactory client, LogService logService) {
return IntegrationFlows.from(Tcp.inboundAdapter(client)
.enrichHeaders(t -> t.headerFunction(IpHeaders.CONNECTION_ID, message -> this.client, true))
.log(msg -> "client: " + logService.log(msg))
.<byte[], Boolean>route(this::shouldForwardToHttp,
mapping -> mapping.subFlowMapping(true, sf -> sf
.enrichHeaders(t -> t.header("Content-Type", MimeTypeUtils.APPLICATION_JSON_VALUE))
.<byte[], RequestMessage>transform(this::buildRequestFromMessage)
.<RequestMessage, HttpEntity>transform(this::getHttpEntity)
.handle(Http.outboundGateway(restUrl).httpMethod(HttpMethod.POST)
.expectedResponseType(ResponseMessage.class))
.<ResponseMessage, byte[]>transform(p -> this.transformResponse(p))
.handle(Tcp.outboundAdapter(client))).subFlowMapping(false,
t -> t.handle(Tcp.outboundAdapter(server).retryInterval(1000))))
.get();
}
HttpEntity getHttpEntity(RequestMessage request) {
MultiValueMap<String, String> mv = new HttpHeaders();
mv.add("accessToken", tokenProvider.getCurrentAccessToken());
HttpEntity entity = new HttpEntity(request, mv);
return entity;
}
I have tried by adding a requestHandlerRetry advice and redirecting it to a recoveryChannel, but I was not able to return something to the caller flow in order to get the response with the status code and retry the call with the new accessToken.
Any idea on how I can implement this?
I don't think you need a retry advice since you definitely are simulating it via catching that 401 exception and calling a service back with refreshed token. Sounds more like recursion. To achieve it properly I would suggest to take a look into an ExpressionEvaluatingRequestHandlerAdvice: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#message-handler-advice-chain. Yes, it is similar to the retry one and it also has that failureChannel, but there is no built-in retry since we are going to simulate it calling the same endpoint again and again when necessary.
To simplify a recursion logic, I would extract that .handle(Http.outboundGateway(restUrl).httpMethod(HttpMethod.POST) .expectedResponseType(ResponseMessage.class)) into a separate flow and use a gateway() with an input channel for that flow in the main flow instead.
A failureChannel sub-flow should re-route its message back to the input of the gateway flow.
What is the most important part in this logic is to carry on all the original request message headers which includes a required for the gateway logic replyChannel.
See more docs about gateways: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#gateway.
When an ExpressionEvaluatingRequestHandlerAdvice sends a message to the failureChannel, it comes as an ErrorMessage with a MessageHandlingExpressionEvaluatingAdviceException as a payload. The message which causes a failure and has all the required headers is there in the getFailedMessage() property. So, you take that message, request for fresh token, add it into headers of a new message based on that original. In the end you send this new message to the input channel of the IntegrationFlow for an HTTP request. When all is good, the result of the HTTP call is going to be forwarded to the mentioned replyChannel from headers and in therefore to the main flow for next steps.