FINERACT-1744

- [x] Batch api support for idempotency
- [x] BatchFilter for generic batch filtering
- [x] LoanCOBApiFilter extend with BatchFilter support
- [x] FineractRequestContextHolder generic context holder with batch and http request supports
This commit is contained in:
Janos Haber
2022-12-14 13:06:20 +01:00
committed by Arnold Gálovics
parent be958b9aef
commit 995b1ea021
14 changed files with 644 additions and 114 deletions
@@ -18,6 +18,7 @@
*/
package org.apache.fineract.batch.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
@@ -32,6 +33,7 @@ import lombok.experimental.Accessors;
* @see BatchResponse
*/
@NoArgsConstructor
@AllArgsConstructor
@Data
@Accessors(chain = true)
public class Header {
@@ -22,7 +22,11 @@ import com.google.gson.Gson;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.UriInfo;
@@ -33,15 +37,22 @@ import org.apache.fineract.batch.command.CommandStrategy;
import org.apache.fineract.batch.command.CommandStrategyProvider;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
import org.apache.fineract.batch.domain.Header;
import org.apache.fineract.batch.exception.ClientDetailsNotFoundException;
import org.apache.fineract.batch.exception.ErrorHandler;
import org.apache.fineract.batch.exception.ErrorInfo;
import org.apache.fineract.batch.service.ResolutionHelper.BatchRequestNode;
import org.apache.fineract.infrastructure.core.domain.BatchRequestContextHolder;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.apache.fineract.infrastructure.core.exception.AbstractIdempotentCommandException;
import org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessFailedException;
import org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessSucceedException;
import org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessUnderProcessingException;
import org.apache.fineract.infrastructure.core.filters.BatchCallHandler;
import org.apache.fineract.infrastructure.core.filters.BatchFilter;
import org.springframework.dao.NonTransientDataAccessException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
@@ -49,7 +60,6 @@ import org.springframework.transaction.support.TransactionTemplate;
* CommandStrategy from CommandStrategyProvider.
*
* @author Rishabh Shukla
*
* @see org.apache.fineract.batch.domain.BatchRequest
* @see org.apache.fineract.batch.domain.BatchResponse
* @see org.apache.fineract.batch.command.CommandStrategyProvider
@@ -63,6 +73,10 @@ public class BatchApiServiceImpl implements BatchApiService {
private final ResolutionHelper resolutionHelper;
private final TransactionTemplate transactionTemplate;
private final List<BatchFilter> batchFilters;
private final FineractRequestContextHolder fineractRequestContextHolder;
/**
* Returns the response list by getting a proper {@link org.apache.fineract.batch.command.CommandStrategy}.
* execute() method of acquired commandStrategy is then provided with the separate Request.
@@ -84,6 +98,7 @@ public class BatchApiServiceImpl implements BatchApiService {
responseList.add(response);
return responseList;
}
for (BatchRequestNode rootNode : batchRequestNodes) {
final BatchRequest rootRequest = rootNode.getRequest();
final CommandStrategy commandStrategy = this.strategyProvider
@@ -100,9 +115,15 @@ public class BatchApiServiceImpl implements BatchApiService {
}
private BatchResponse safelyExecuteStrategy(CommandStrategy commandStrategy, BatchRequest request, UriInfo uriInfo) {
private BatchResponse safelyExecuteStrategy(CommandStrategy commandStrategy, BatchRequest request, UriInfo originalUriInfo) {
try {
return commandStrategy.execute(request, uriInfo);
BatchRequestContextHolder.setRequestAttributes(new HashMap<>(Optional.ofNullable(request.getHeaders())
.map(list -> list.stream().collect(Collectors.toMap(Header::getName, Header::getValue)))
.orElse(Collections.emptyMap())));
return new BatchCallHandler(this.batchFilters, commandStrategy::execute).serviceCall(request, originalUriInfo);
} catch (AbstractIdempotentCommandException idempotentException) {
return handleIdempotentRequests(idempotentException, request);
} catch (RuntimeException e) {
log.warn("Exception while executing batch strategy {}", commandStrategy.getClass().getSimpleName(), e);
@@ -114,13 +135,33 @@ public class BatchApiServiceImpl implements BatchApiService {
response.setStatusCode(ex.getStatusCode());
response.setBody(ex.getMessage());
return response;
} finally {
BatchRequestContextHolder.resetRequestAttributes();
}
}
private BatchResponse handleIdempotentRequests(AbstractIdempotentCommandException idempotentException, BatchRequest request) {
BatchResponse response = new BatchResponse();
response.setRequestId(request.getRequestId());
response.setHeaders(Optional.ofNullable(request.getHeaders()).orElse(new HashSet<>()));
response.getHeaders().add(new Header(AbstractIdempotentCommandException.IDEMPOTENT_CACHE_HEADER, "true"));
response.setBody(idempotentException.getResponse());
if (idempotentException instanceof IdempotentCommandProcessSucceedException) {
response.setStatusCode(200);
} else if (idempotentException instanceof IdempotentCommandProcessUnderProcessingException) {
response.setStatusCode(409);
} else if (idempotentException instanceof IdempotentCommandProcessFailedException) {
response.setStatusCode(((IdempotentCommandProcessFailedException) idempotentException).getStatusCode());
} else {
response.setStatusCode(500);
}
return response;
}
private List<BatchResponse> processChildRequests(final BatchRequestNode rootRequest, BatchResponse rootResponse, UriInfo uriInfo) {
final List<BatchResponse> childResponses = new ArrayList<>();
if (rootRequest.getChildRequests().size() > 0) {
if (!rootRequest.getChildRequests().isEmpty()) {
for (BatchRequestNode childNode : rootRequest.getChildRequests()) {
@@ -175,30 +216,25 @@ public class BatchApiServiceImpl implements BatchApiService {
public List<BatchResponse> handleBatchRequestsWithEnclosingTransaction(final List<BatchRequest> requestList, final UriInfo uriInfo) {
List<BatchResponse> responseList = new ArrayList<>();
try {
return this.transactionTemplate.execute(new TransactionCallback<List<BatchResponse>>() {
return this.transactionTemplate.execute(status -> {
try {
responseList.addAll(handleBatchRequests(requestList, uriInfo));
return responseList;
} catch (RuntimeException ex) {
@Override
public List<BatchResponse> doInTransaction(TransactionStatus status) {
try {
responseList.addAll(handleBatchRequests(requestList, uriInfo));
return responseList;
} catch (RuntimeException ex) {
ErrorInfo e = ErrorHandler.handler(ex);
BatchResponse errResponse = new BatchResponse();
errResponse.setStatusCode(e.getStatusCode());
errResponse.setBody(e.getMessage());
ErrorInfo e = ErrorHandler.handler(ex);
BatchResponse errResponse = new BatchResponse();
errResponse.setStatusCode(e.getStatusCode());
errResponse.setBody(e.getMessage());
List<BatchResponse> errResponseList = new ArrayList<>();
errResponseList.add(errResponse);
List<BatchResponse> errResponseList = new ArrayList<>();
errResponseList.add(errResponse);
status.setRollbackOnly();
return errResponseList;
}
status.setRollbackOnly();
return errResponseList;
}
});
} catch (TransactionException ex) {
} catch (TransactionException | NonTransientDataAccessException ex) {
ErrorInfo e = ErrorHandler.handler(ex);
BatchResponse errResponse = new BatchResponse();
errResponse.setStatusCode(e.getStatusCode());
@@ -213,21 +249,6 @@ public class BatchApiServiceImpl implements BatchApiService {
List<BatchResponse> errResponseList = new ArrayList<>();
errResponseList.add(errResponse);
return errResponseList;
} catch (final NonTransientDataAccessException ex) {
ErrorInfo e = ErrorHandler.handler(ex);
BatchResponse errResponse = new BatchResponse();
errResponse.setStatusCode(e.getStatusCode());
for (BatchResponse res : responseList) {
if (!res.getStatusCode().equals(200)) {
errResponse.setBody("Transaction is being rolled back. First erroneous request: \n" + new Gson().toJson(res));
break;
}
}
List<BatchResponse> errResponseList = new ArrayList<>();
errResponseList.add(errResponse);
return errResponseList;
}
}
@@ -21,29 +21,24 @@ package org.apache.fineract.commands.service;
import java.util.Optional;
import lombok.RequiredArgsConstructor;
import org.apache.fineract.commands.domain.CommandWrapper;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
@Component
@RequiredArgsConstructor
public class IdempotencyKeyResolver {
private final FineractRequestContextHolder fineractRequestContextHolder;
private final IdempotencyKeyGenerator idempotencyKeyGenerator;
private final FineractProperties fineractProperties;
public String resolve(CommandWrapper wrapper) {
return Optional.ofNullable(wrapper.getIdempotencyKey())
.orElseGet(() -> getHeaderAttribute().orElseGet(idempotencyKeyGenerator::create));
return Optional.ofNullable(wrapper.getIdempotencyKey()).orElseGet(() -> getAttribute().orElseGet(idempotencyKeyGenerator::create));
}
private Optional<String> getHeaderAttribute() {
return Optional.ofNullable(RequestContextHolder.getRequestAttributes()) //
.filter(ServletRequestAttributes.class::isInstance) //
.map(ServletRequestAttributes.class::cast) //
.map(ServletRequestAttributes::getRequest) //
.map(request -> request.getHeader(fineractProperties.getIdempotencyKeyHeaderName()));
private Optional<String> getAttribute() {
return Optional.ofNullable(fineractRequestContextHolder.getAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_ATTRIBUTE))
.map(String::valueOf);
}
}
@@ -29,7 +29,6 @@ import java.lang.reflect.Type;
import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -45,6 +44,7 @@ import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDoma
import org.apache.fineract.infrastructure.core.api.JsonCommand;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResultBuilder;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.apache.fineract.infrastructure.core.exception.AbstractIdempotentCommandException;
import org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessFailedException;
import org.apache.fineract.infrastructure.core.exception.IdempotentCommandProcessSucceedException;
@@ -59,8 +59,6 @@ import org.apache.fineract.useradministration.domain.AppUser;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
@Service
@Slf4j
@@ -68,6 +66,8 @@ import org.springframework.web.context.request.RequestContextHolder;
public class SynchronousCommandProcessingService implements CommandProcessingService {
public static final String IDEMPOTENCY_KEY_STORE_FLAG = "idempotencyKeyStoreFlag";
public static final String IDEMPOTENCY_KEY_ATTRIBUTE = "IdempotencyKeyAttribute";
public static final String COMMAND_SOURCE_ID = "commandSourceId";
private final PlatformSecurityContext context;
private final ApplicationContext applicationContext;
@@ -78,6 +78,8 @@ public class SynchronousCommandProcessingService implements CommandProcessingSer
private final IdempotencyKeyResolver idempotencyKeyResolver;
private final IdempotencyKeyGenerator idempotencyKeyGenerator;
private final CommandSourceService commandSourceService;
private final FineractRequestContextHolder fineractRequestContextHolder;
private final Gson gson = GoogleGsonSerializerHelper.createSimpleGson();
@Override
@@ -147,9 +149,8 @@ public class SynchronousCommandProcessingService implements CommandProcessingSer
saveCommandToRequest(savedCommandSource);
}
private static void saveCommandToRequest(CommandSource savedCommandSource) {
Optional.ofNullable(RequestContextHolder.getRequestAttributes()).ifPresent(requestAttributes -> requestAttributes
.setAttribute(COMMAND_SOURCE_ID, savedCommandSource.getId(), RequestAttributes.SCOPE_REQUEST));
private void saveCommandToRequest(CommandSource savedCommandSource) {
fineractRequestContextHolder.setAttribute(COMMAND_SOURCE_ID, savedCommandSource.getId());
}
private void publishHookErrorEvent(CommandWrapper wrapper, JsonCommand command, Throwable t) {
@@ -176,9 +177,7 @@ public class SynchronousCommandProcessingService implements CommandProcessingSer
}
private void setIdempotencyKeyStoreFlag(boolean flag) {
Optional.ofNullable(RequestContextHolder.getRequestAttributes()).ifPresent(
requestAttributes -> requestAttributes.setAttribute(IDEMPOTENCY_KEY_STORE_FLAG, flag, RequestAttributes.SCOPE_REQUEST));
fineractRequestContextHolder.setAttribute(IDEMPOTENCY_KEY_STORE_FLAG, flag);
}
@Transactional
@@ -0,0 +1,64 @@
/**
* 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.fineract.infrastructure.core.domain;
import java.util.Map;
import org.springframework.core.NamedThreadLocal;
public final class BatchRequestContextHolder {
private BatchRequestContextHolder() {}
private static final ThreadLocal<Map<String, Object>> batchAttributes = new NamedThreadLocal<>("BatchAttributesForProcessing");
/**
* True if the batch attributes are set
*
* @return true if the batch attributes are set
*/
public static boolean isBatchRequest() {
return batchAttributes.get() != null;
}
/**
* Set the batch attributes for the current thread.
*
* @param requestAttributes
* the new batch attributes
*/
public static void setRequestAttributes(Map<String, Object> requestAttributes) {
batchAttributes.set(requestAttributes);
}
/**
* Returns the batch attributes for the current thread.
*
* @return the batch attributes for the current thread, cna be null
*/
public static Map<String, Object> getRequestAttributes() {
return batchAttributes.get();
}
/**
* Reset the batch attributes for the current thread.
*/
public static void resetRequestAttributes() {
batchAttributes.remove();
}
}
@@ -0,0 +1,130 @@
/**
* 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.fineract.infrastructure.core.domain;
import java.util.Optional;
import javax.servlet.http.HttpServletRequest;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
/**
* Common context holder to store environment-specific request attributes. <br/>
* It will try to get the attribute from the following order:
*
* <pre>
* <ul>
* <li>{@link HttpServletRequest} (if the parameter not null)</li>
* <li>{@link BatchRequestContextHolder} if it's a batch request set int the current thread</li>
* <li>{@link RequestContextHolder} if it exists in the current thread</li>
* </ul>
* </pre>
*/
@Component
@NoArgsConstructor
@Slf4j
public final class FineractRequestContextHolder {
/**
* Get the attribute, the request not set using only thread bound context variables
*
* @param key
* attribute key
* @return attribute value or null if it not set in the current thread
*/
public Object getAttribute(String key) {
return getAttribute(key, null);
}
/**
* Get the attribute.
*
* The method will check the request attribute first, then the thread bound context variables
*
* TODO: in {@link org.springframework.context.ApplicationEvent} always return null
*
* @param key
* attribute key
* @param request
* {@link HttpServletRequest} object
* @return attribute value or null if it not set in the current thread or {@link HttpServletRequest} (if not null)
*/
public Object getAttribute(String key, HttpServletRequest request) {
if (request != null) {
return request.getAttribute(key);
} else if (isBatchRequest()) {
return Optional.ofNullable(BatchRequestContextHolder.getRequestAttributes()).map(attributes -> attributes.get(key))
.orElse(null);
} else if (RequestContextHolder.getRequestAttributes() != null) {
return Optional.ofNullable(RequestContextHolder.getRequestAttributes())
.map(r -> r.getAttribute(key, RequestAttributes.SCOPE_REQUEST)).orElse(null);
}
return null;
}
/**
* Set the attribute
*
* @param key
* attribute key
* @param value
* attribute value
*/
public void setAttribute(String key, Object value) {
setAttribute(key, value, null);
}
/**
* Set the attribute.
*
* If the request is not null, it will set the attribute in the request otherwise it will set it in the thread bound
* context variables
*
* TODO: in {@link org.springframework.context.ApplicationEvent} the attributes not set
*
* @param key
* attribute key
* @param value
* attribute value
* @param request
* {@link HttpServletRequest} object
*/
public void setAttribute(String key, Object value, HttpServletRequest request) {
if (request != null) {
request.setAttribute(key, value);
} else if (isBatchRequest()) {
Optional.ofNullable(BatchRequestContextHolder.getRequestAttributes()).ifPresent(attributes -> attributes.put(key, value));
} else if (RequestContextHolder.getRequestAttributes() != null) {
Optional.ofNullable(RequestContextHolder.getRequestAttributes())
.ifPresent(requestAttributes -> requestAttributes.setAttribute(key, value, RequestAttributes.SCOPE_REQUEST));
}
}
/**
* True if {@link BatchRequestContextHolder} will be set in the current thread
*
* @return true if the current request is a batch request
*/
public static boolean isBatchRequest() {
return BatchRequestContextHolder.isBatchRequest();
}
}
@@ -0,0 +1,46 @@
/**
* 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.fineract.infrastructure.core.filters;
import java.util.List;
import java.util.function.BiFunction;
import javax.ws.rs.core.UriInfo;
import lombok.RequiredArgsConstructor;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
@RequiredArgsConstructor
public class BatchCallHandler implements BatchFilterChain {
private final List<? extends BatchFilter> filters;
private final BiFunction<BatchRequest, UriInfo, BatchResponse> lastElement;
private int currentPosition = 0;
@Override
public BatchResponse serviceCall(BatchRequest batchRequest, UriInfo uriInfo) {
if (this.currentPosition == filters.size()) {
return lastElement.apply(batchRequest, uriInfo);
} else {
BatchFilter currentFilter = filters.get(this.currentPosition);
this.currentPosition++;
return currentFilter.doFilter(batchRequest, uriInfo, this);
}
}
}
@@ -0,0 +1,30 @@
/**
* 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.fineract.infrastructure.core.filters;
import javax.ws.rs.core.UriInfo;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
public interface BatchFilter {
BatchResponse doFilter(BatchRequest batchRequest, UriInfo uriInfo, BatchFilterChain chain);
}
@@ -0,0 +1,28 @@
/**
* 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.fineract.infrastructure.core.filters;
import javax.ws.rs.core.UriInfo;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
public interface BatchFilterChain {
BatchResponse serviceCall(BatchRequest batchRequest, UriInfo uriInfo);
}
@@ -18,6 +18,7 @@
*/
package org.apache.fineract.infrastructure.core.filters;
import io.vavr.collection.List;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Optional;
@@ -25,13 +26,19 @@ import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.UriInfo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.mutable.Mutable;
import org.apache.commons.lang3.mutable.MutableObject;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
import org.apache.fineract.batch.domain.Header;
import org.apache.fineract.commands.domain.CommandSourceRepository;
import org.apache.fineract.commands.service.CommandSourceService;
import org.apache.fineract.commands.service.SynchronousCommandProcessingService;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
@@ -40,11 +47,15 @@ import org.springframework.web.util.ContentCachingResponseWrapper;
@RequiredArgsConstructor
@Slf4j
@Component
public class IdempotencyStoreFilter extends OncePerRequestFilter {
public class IdempotencyStoreFilter extends OncePerRequestFilter implements BatchFilter {
private final CommandSourceRepository commandSourceRepository;
private final CommandSourceService commandSourceService;
private final FineractProperties fineractProperties;
private final FineractRequestContextHolder fineractRequestContextHolder;
@Override
protected void doFilterInternal(@NotNull HttpServletRequest request, @NotNull HttpServletResponse response,
@NotNull FilterChain filterChain) throws ServletException, IOException {
@@ -52,23 +63,34 @@ public class IdempotencyStoreFilter extends OncePerRequestFilter {
if (isAllowedContentTypeRequest(request)) {
wrapper.setValue(new ContentCachingResponseWrapper(response));
}
extractIdempotentKeyFromHttpServletRequest(request).ifPresent(idempotentKey -> fineractRequestContextHolder
.setAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_ATTRIBUTE, idempotentKey, request));
filterChain.doFilter(request, wrapper.getValue() != null ? wrapper.getValue() : response);
Optional<Long> commandId = getCommandId(request);
boolean isSuccessWithoutStored = isStoreIdempotencyKey(request) && commandId.isPresent() && isAllowedContentTypeResponse(response)
&& wrapper.getValue() != null;
if (isSuccessWithoutStored) {
commandSourceRepository.findById(commandId.get()).ifPresent(commandSource -> {
commandSource.setResultStatusCode(response.getStatus());
commandSource.setResult(new String(wrapper.getValue().getContentAsByteArray(), StandardCharsets.UTF_8));
commandSourceService.saveResult(commandSource);
});
storeCommandResult(response.getStatus(), new String(wrapper.getValue().getContentAsByteArray(), StandardCharsets.UTF_8),
commandId);
}
if (wrapper.getValue() != null) {
wrapper.getValue().copyBodyToResponse();
}
}
private void storeCommandResult(int response, String body, Optional<Long> commandId) {
commandSourceRepository.findById(commandId.get()).ifPresent(commandSource -> {
commandSource.setResultStatusCode(response);
commandSource.setResult(body);
commandSourceService.saveResult(commandSource);
});
}
private Optional<String> extractIdempotentKeyFromHttpServletRequest(HttpServletRequest request) {
return Optional.ofNullable(request.getHeader(fineractProperties.getIdempotencyKeyHeaderName()));
}
private boolean isAllowedContentTypeResponse(HttpServletResponse response) {
return Optional.ofNullable(response.getContentType()).map(String::toLowerCase).map(ct -> ct.contains("application/json"))
.orElse(false);
@@ -80,12 +102,37 @@ public class IdempotencyStoreFilter extends OncePerRequestFilter {
}
private boolean isStoreIdempotencyKey(HttpServletRequest request) {
return Optional.ofNullable(request.getAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_STORE_FLAG))
return Optional
.ofNullable(
fineractRequestContextHolder.getAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_STORE_FLAG, request))
.filter(Boolean.class::isInstance).map(Boolean.class::cast).orElse(false);
}
private Optional<Long> getCommandId(HttpServletRequest request) {
return Optional.ofNullable(request.getAttribute(SynchronousCommandProcessingService.COMMAND_SOURCE_ID))
return Optional
.ofNullable(fineractRequestContextHolder.getAttribute(SynchronousCommandProcessingService.COMMAND_SOURCE_ID, request))
.filter(Long.class::isInstance).map(Long.class::cast);
}
private Optional<String> extractIdempotentKeyFromBatchRequest(BatchRequest request) {
return Optional.ofNullable(request.getHeaders()) //
.map(List::ofAll) //
.flatMap(headers -> headers.find(header -> header.getName().equals(fineractProperties.getIdempotencyKeyHeaderName()))
.toJavaOptional()) //
.map(Header::getValue); //
}
@Override
public BatchResponse doFilter(BatchRequest batchRequest, UriInfo uriInfo, BatchFilterChain chain) {
extractIdempotentKeyFromBatchRequest(batchRequest).ifPresent(idempotentKey -> fineractRequestContextHolder
.setAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_ATTRIBUTE, idempotentKey));
BatchResponse result = chain.serviceCall(batchRequest, uriInfo);
Optional<Long> commandId = getCommandId(null);
boolean isSuccessWithoutStored = isStoreIdempotencyKey(null) && commandId.isPresent();
if (isSuccessWithoutStored) {
storeCommandResult(result.getStatusCode(), result.getBody(), commandId);
}
return result;
}
}
@@ -19,6 +19,7 @@
package org.apache.fineract.infrastructure.jobs.filter;
import com.google.common.base.Splitter;
import io.vavr.control.Either;
import java.io.IOException;
import java.math.BigDecimal;
import java.util.Collections;
@@ -32,11 +33,16 @@ import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.UriInfo;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
import org.apache.fineract.cob.service.InlineLoanCOBExecutorServiceImpl;
import org.apache.fineract.cob.service.LoanAccountLockService;
import org.apache.fineract.infrastructure.core.data.ApiGlobalErrorResponse;
import org.apache.fineract.infrastructure.core.filters.BatchFilter;
import org.apache.fineract.infrastructure.core.filters.BatchFilterChain;
import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
import org.apache.fineract.portfolio.loanaccount.domain.GLIMAccountInfoRepository;
import org.apache.fineract.portfolio.loanaccount.domain.GroupLoanIndividualMonitoringAccount;
@@ -49,7 +55,7 @@ import org.springframework.web.filter.OncePerRequestFilter;
@Component
@RequiredArgsConstructor
public class LoanCOBApiFilter extends OncePerRequestFilter {
public class LoanCOBApiFilter extends OncePerRequestFilter implements BatchFilter {
private final GLIMAccountInfoRepository glimAccountInfoRepository;
private final LoanAccountLockService loanAccountLockService;
@@ -68,39 +74,83 @@ public class LoanCOBApiFilter extends OncePerRequestFilter {
private static final Integer GLIM_STRING_INDEX_IN_URL = 2;
private static final String JOB_NAME = "INLINE_LOAN_COB";
private static class Reject {
private final String message;
private final Integer statusCode;
Reject(String message, Integer statusCode) {
this.message = message;
this.statusCode = statusCode;
}
public static Reject reject(Long loanId, int status) {
return new Reject(ApiGlobalErrorResponse.loanIsLocked(loanId).toJson(), status);
}
public void toServletResponse(HttpServletResponse response) throws IOException {
response.setStatus(statusCode);
response.getWriter().write(message);
}
public BatchResponse toBatchResponse(BatchRequest request) {
BatchResponse response = new BatchResponse();
response.setStatusCode(statusCode);
response.setBody(message);
response.setHeaders(request.getHeaders());
response.setRequestId(request.getRequestId());
return response;
}
}
@Override
protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
throws ServletException, IOException {
if (!isOnApiList(request) || isBypassUser(response)) {
if (!isOnApiList(request.getPathInfo(), request.getMethod())) {
proceed(filterChain, request, response);
} else {
Iterable<String> split = Splitter.on('/').split(request.getPathInfo());
Supplier<Stream<String>> streamSupplier = () -> StreamSupport.stream(split.spliterator(), false);
boolean isGlim = isGlim(streamSupplier);
Long loanIdFromRequest = getLoanId(isGlim, streamSupplier);
List<Long> loanIds = isGlim ? getGlimChildLoanIds(loanIdFromRequest) : Collections.singletonList(loanIdFromRequest);
if (isLoanHardLocked(loanIds)) {
reject(loanIdFromRequest, response, HttpStatus.SC_CONFLICT);
} else if (isLoanSoftLocked(loanIds)) {
executeInlineCob(loanIds);
Either<Reject, Boolean> bypassUser = isBypassUser();
if (bypassUser.isRight() && Boolean.TRUE.equals(bypassUser.get())) {
proceed(filterChain, request, response);
} else if (bypassUser.isLeft()) {
bypassUser.getLeft().toServletResponse(response);
} else {
proceed(filterChain, request, response);
Either<Reject, List<Long>> result = loanIdCalculation(request.getPathInfo());
if (result.isLeft()) {
result.getLeft().toServletResponse(response);
} else {
if (isLoanSoftLocked(result.get())) {
executeInlineCob(result.get());
}
proceed(filterChain, request, response);
}
}
}
}
private Either<Reject, List<Long>> loanIdCalculation(String pathInfo) {
Iterable<String> split = Splitter.on('/').split(pathInfo);
Supplier<Stream<String>> streamSupplier = () -> StreamSupport.stream(split.spliterator(), false);
boolean isGlim = isGlim(streamSupplier);
Long loanIdFromRequest = getLoanId(isGlim, streamSupplier);
List<Long> loanIds = isGlim ? getGlimChildLoanIds(loanIdFromRequest) : Collections.singletonList(loanIdFromRequest);
if (isLoanHardLocked(loanIds)) {
return Either.left(Reject.reject(loanIdFromRequest, HttpStatus.SC_CONFLICT));
} else {
return Either.right(loanIds);
}
}
private void executeInlineCob(List<Long> loanIds) {
inlineLoanCOBExecutorService.execute(loanIds, JOB_NAME);
}
private boolean isBypassUser(HttpServletResponse response) throws IOException {
private Either<Reject, Boolean> isBypassUser() {
try {
return context.authenticatedUser().isBypassUser();
return Either.right(context.authenticatedUser().isBypassUser());
} catch (UnAuthenticatedUserException e) {
reject(null, response, HttpStatus.SC_UNAUTHORIZED);
return Either.left(Reject.reject(null, HttpStatus.SC_UNAUTHORIZED));
}
return false;
}
private List<Long> getGlimChildLoanIds(Long loanIdFromRequest) {
@@ -131,12 +181,6 @@ public class LoanCOBApiFilter extends OncePerRequestFilter {
filterChain.doFilter(request, response);
}
private void reject(Long loanId, HttpServletResponse response, int status) throws IOException {
response.setStatus(status);
ApiGlobalErrorResponse errorResponse = ApiGlobalErrorResponse.loanIsLocked(loanId);
response.getWriter().write(errorResponse.toJson());
}
private Long getLoanId(boolean isGlim, Supplier<Stream<String>> streamSupplier) {
if (!isGlim) {
return streamSupplier.get().skip(LOAN_ID_INDEX_IN_URL).findFirst().map(Long::valueOf).orElse(null);
@@ -145,14 +189,41 @@ public class LoanCOBApiFilter extends OncePerRequestFilter {
}
}
private boolean isOnApiList(HttpServletRequest request) {
if (StringUtils.isBlank(request.getPathInfo())) {
private boolean isOnApiList(String pathInfo, String method) {
if (StringUtils.isBlank(pathInfo)) {
return false;
}
return HTTP_METHODS.contains(HttpMethod.valueOf(request.getMethod())) && URL_FUNCTION.test(request.getPathInfo());
return HTTP_METHODS.contains(HttpMethod.valueOf(method)) && URL_FUNCTION.test(pathInfo);
}
private boolean isGlim(Supplier<Stream<String>> streamSupplier) {
return streamSupplier.get().skip(GLIM_STRING_INDEX_IN_URL).findFirst().map(s -> s.equals("glimAccount")).orElse(false);
}
@Override
public BatchResponse doFilter(BatchRequest batchRequest, UriInfo uriInfo, BatchFilterChain chain) {
if (!isOnApiList("/" + batchRequest.getRelativeUrl(), batchRequest.getMethod())) {
return chain.serviceCall(batchRequest, uriInfo);
} else {
Either<Reject, Boolean> bypassUser = isBypassUser();
if (bypassUser.isRight() && Boolean.TRUE.equals(bypassUser.get())) {
return chain.serviceCall(batchRequest, uriInfo);
} else if (bypassUser.isLeft()) {
return bypassUser.getLeft().toBatchResponse(batchRequest);
} else {
Either<Reject, List<Long>> result = loanIdCalculation("/" + batchRequest.getRelativeUrl());
if (result.isLeft()) {
return result.getLeft().toBatchResponse(batchRequest);
} else {
if (!isLoanSoftLocked(result.get())) {
executeInlineCob(result.get());
}
return chain.serviceCall(batchRequest, uriInfo);
}
}
}
}
}
@@ -18,46 +18,47 @@
*/
package org.apache.fineract.commands.service;
import static org.mockito.Mockito.when;
import java.util.HashMap;
import org.apache.fineract.commands.domain.CommandWrapper;
import org.apache.fineract.infrastructure.core.config.FineractProperties;
import org.apache.fineract.infrastructure.core.domain.BatchRequestContextHolder;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.springframework.mock.web.MockHttpServletRequest;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import org.mockito.Spy;
public class IdempotencyKeyResolverTest {
@Mock
private IdempotencyKeyGenerator idempotencyKeyGenerator;
@Mock
private FineractProperties fineractProperties;
@InjectMocks
private IdempotencyKeyResolver underTest;
@Spy
private FineractRequestContextHolder fineractRequestContextHolder;
@BeforeEach
public void setup() {
MockitoAnnotations.openMocks(this);
BatchRequestContextHolder.setRequestAttributes(new HashMap<>());
}
@AfterEach
public void tearDown() {
BatchRequestContextHolder.resetRequestAttributes();
}
@Test
public void testIPKResolveFromRequest() {
String idkh = "foo";
String idk = "bar";
Mockito.when(fineractProperties.getIdempotencyKeyHeaderName()).thenReturn(idkh);
MockHttpServletRequest request = new MockHttpServletRequest();
request.addHeader(idkh, idk);
RequestAttributes requestAttributes = new ServletRequestAttributes(request);
RequestContextHolder.setRequestAttributes(requestAttributes);
fineractRequestContextHolder.setAttribute(SynchronousCommandProcessingService.IDEMPOTENCY_KEY_ATTRIBUTE, idk);
CommandWrapper wrapper = CommandWrapper.wrap("act", "ent", 1L, 1L);
String resolvedIdk = underTest.resolve(wrapper);
Assertions.assertEquals(idk, resolvedIdk);
@@ -66,8 +67,7 @@ public class IdempotencyKeyResolverTest {
@Test
public void testIPKResolveFromGenerate() {
String idk = "idk";
Mockito.when(idempotencyKeyGenerator.create()).thenReturn(idk);
RequestContextHolder.setRequestAttributes(null);
when(idempotencyKeyGenerator.create()).thenReturn(idk);
CommandWrapper wrapper = CommandWrapper.wrap("act", "ent", 1L, 1L);
String resolvedIdk = underTest.resolve(wrapper);
Assertions.assertEquals(idk, resolvedIdk);
@@ -75,7 +75,6 @@ public class IdempotencyKeyResolverTest {
@Test
public void testIPKResolveFromWrapper() {
RequestContextHolder.setRequestAttributes(null);
String idk = "idk";
CommandWrapper wrapper = new CommandWrapper(null, null, null, null, null, null, null, null, null, null, null, null, null, null,
null, null, null, idk);
@@ -31,6 +31,7 @@ import org.apache.fineract.commands.provider.CommandHandlerProvider;
import org.apache.fineract.infrastructure.configuration.domain.ConfigurationDomainService;
import org.apache.fineract.infrastructure.core.api.JsonCommand;
import org.apache.fineract.infrastructure.core.data.CommandProcessingResult;
import org.apache.fineract.infrastructure.core.domain.FineractRequestContextHolder;
import org.apache.fineract.infrastructure.core.serialization.ToApiJsonSerializer;
import org.apache.fineract.infrastructure.security.service.PlatformSecurityContext;
import org.apache.fineract.useradministration.domain.AppUser;
@@ -41,6 +42,7 @@ import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.mockito.Spy;
import org.springframework.context.ApplicationContext;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
@@ -66,6 +68,9 @@ public class SynchronousCommandProcessingServiceTest {
@Mock
private CommandSourceService commandSourceService;
@Spy
private FineractRequestContextHolder fineractRequestContextHolder;
@InjectMocks
private SynchronousCommandProcessingService underTest;
@@ -25,9 +25,13 @@ import io.restassured.specification.RequestSpecification;
import io.restassured.specification.ResponseSpecification;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.UUID;
import org.apache.fineract.batch.domain.BatchRequest;
import org.apache.fineract.batch.domain.BatchResponse;
import org.apache.fineract.batch.domain.Header;
import org.apache.fineract.infrastructure.core.exception.AbstractIdempotentCommandException;
import org.apache.fineract.integrationtests.common.BatchHelper;
import org.apache.fineract.integrationtests.common.ClientHelper;
import org.apache.fineract.integrationtests.common.CollateralManagementHelper;
@@ -146,4 +150,93 @@ public class BatchRequestsIntegrationTest {
Assertions.assertEquals(200L, (long) res.getStatusCode(), "Verify Status Code 200");
}
}
@Test
public void shouldReturnOkStatusWithIdempotencySupport() {
// Generate a random count of number of clients to be created
final Integer clientsCount = (int) Math.ceil(secureRandom.nextDouble() * 7) + 3;
final Integer[] clientIDs = new Integer[clientsCount];
// Create a new group and get its groupId
Integer groupID = GroupHelper.createGroup(this.requestSpec, this.responseSpec, true);
// Create new clients and add those to this group
for (Integer i = 0; i < clientsCount; i++) {
clientIDs[i] = ClientHelper.createClient(this.requestSpec, this.responseSpec);
groupID = GroupHelper.associateClient(this.requestSpec, this.responseSpec, groupID.toString(), clientIDs[i].toString());
LOG.info("client {} has been added to the group {}", clientIDs[i], groupID);
}
// Generate a random count of number of new loan products to be created
final Integer loansCount = (int) Math.ceil(secureRandom.nextDouble() * 4) + 1;
final Integer[] loanProducts = new Integer[loansCount];
// Create new loan Products
LoanTransactionHelper helper = new LoanTransactionHelper(this.requestSpec, this.responseSpec);
for (Integer i = 0; i < loansCount; i++) {
final String loanProductJSON = new LoanProductTestBuilder() //
.withPrincipal(String.valueOf(10000.00 + Math.ceil(secureRandom.nextDouble() * 1000000.00))) //
.withNumberOfRepayments(String.valueOf(2 + (int) Math.ceil(secureRandom.nextDouble() * 36))) //
.withRepaymentAfterEvery(String.valueOf(1 + (int) Math.ceil(secureRandom.nextDouble() * 3))) //
.withRepaymentTypeAsMonth() //
.withinterestRatePerPeriod(String.valueOf(1 + (int) Math.ceil(secureRandom.nextDouble() * 4))) //
.withInterestRateFrequencyTypeAsMonths() //
.withAmortizationTypeAsEqualPrincipalPayment() //
.withInterestTypeAsDecliningBalance() //
.currencyDetails("0", "100").build(null);
loanProducts[i] = helper.getLoanProductId(loanProductJSON);
}
// Select anyone of the loan products at random
final Integer loanProductID = loanProducts[(int) Math.floor(secureRandom.nextDouble() * (loansCount - 1))];
final List<BatchRequest> batchRequests = new ArrayList<>();
// Select a few clients from created group at random
Integer selClientsCount = (int) Math.ceil(secureRandom.nextDouble() * clientsCount) + 2;
for (int i = 0; i < selClientsCount; i++) {
final Integer collateralId = CollateralManagementHelper.createCollateralProduct(this.requestSpec, this.responseSpec);
Assertions.assertNotNull(collateralId);
final Integer clientCollateralId = CollateralManagementHelper.createClientCollateral(this.requestSpec, this.responseSpec,
String.valueOf(clientIDs[(int) Math.floor(secureRandom.nextDouble() * (clientsCount - 1))]), collateralId);
Assertions.assertNotNull(clientCollateralId);
BatchRequest br = BatchHelper.applyLoanRequest((long) selClientsCount, null, loanProductID, clientCollateralId);
br.setBody(br.getBody().replace("$.clientId",
String.valueOf(clientIDs[(int) Math.floor(secureRandom.nextDouble() * (clientsCount - 1))])));
br.setHeaders(new HashSet<>());
br.getHeaders().add(new Header("Idempotency-Key", UUID.randomUUID().toString()));
batchRequests.add(br);
}
// Send the request to Batch - API
final String jsonifiedRequest = BatchHelper.toJsonString(batchRequests);
final List<BatchResponse> response = BatchHelper.postBatchRequestsWithoutEnclosingTransaction(this.requestSpec, this.responseSpec,
jsonifiedRequest);
// Verify that each loan has been applied successfully
for (BatchResponse res : response) {
Assertions.assertFalse(io.vavr.collection.List.ofAll(res.getHeaders())
.find(header -> header.getName().equals(AbstractIdempotentCommandException.IDEMPOTENT_CACHE_HEADER)).isDefined(),
"First can not be cached!");
Assertions.assertEquals(200L, (long) res.getStatusCode(), "Verify Status Code 200");
}
final List<BatchResponse> secondResponse = BatchHelper.postBatchRequestsWithoutEnclosingTransaction(this.requestSpec,
this.responseSpec, jsonifiedRequest);
// Verify that each loan has been applied successfully
for (BatchResponse res : secondResponse) {
Assertions.assertEquals("true",
io.vavr.collection.List.ofAll(res.getHeaders())
.find(header -> header.getName().equals(AbstractIdempotentCommandException.IDEMPOTENT_CACHE_HEADER))
.map(Header::getValue).get(),
"Not cached by idempotency key!");
Assertions.assertEquals(200L, (long) res.getStatusCode(), "Verify Status Code 200");
}
}
}