Merge pull request #114 from catenax-ng/CX_PI12_S5_04

[feat|sde-backend] : PCF exchange service implementation code updated.
eclipse-tractusx · Apr 9, 2024 · 309a884 · 309a884
2 parents 69f7639 + 18cb31a
commit 309a884
Show file tree

Hide file tree

Showing 5 changed files with 675 additions and 1 deletion.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -10,7 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Updated supported sub-model implementation classes.
 - EDC asset update refactored in supported submodels.
 - Support for pcf v6.0.0 submodel.
-- Added module for pcf exchange service.
+- Updated PCF exchange service implementation code.
 
 ## [2.3.6] - 2024-03-06
 ### Fixed

diff --git a/.../org/eclipse/tractusx/sde/pcfexchange/service/impl/AsyncPushPCFDataForApproveRequest.java b/.../org/eclipse/tractusx/sde/pcfexchange/service/impl/AsyncPushPCFDataForApproveRequest.java
@@ -0,0 +1,142 @@
+/********************************************************************************
+ * Copyright (c) 2024 T-Systems International GmbH
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License, Version 2.0 which is available at
+ * https://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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.tractusx.sde.pcfexchange.service.impl;
+
+import java.util.List;
+
+import org.eclipse.tractusx.sde.common.entities.PolicyModel;
+import org.eclipse.tractusx.sde.common.exception.NoDataFoundException;
+import org.eclipse.tractusx.sde.common.model.PagingResponse;
+import org.eclipse.tractusx.sde.common.utils.PolicyOperationUtil;
+import org.eclipse.tractusx.sde.pcfexchange.enums.PCFRequestStatusEnum;
+import org.eclipse.tractusx.sde.pcfexchange.enums.PCFTypeEnum;
+import org.eclipse.tractusx.sde.pcfexchange.request.PcfRequestModel;
+import org.eclipse.tractusx.sde.submodels.pcf.entity.PcfEntity;
+import org.eclipse.tractusx.sde.submodels.pcf.service.PcfService;
+import org.springframework.stereotype.Component;
+
+import com.google.gson.JsonObject;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@SuppressWarnings("unchecked")
+public class AsyncPushPCFDataForApproveRequest {
+
+	private final PCFRepositoryService pcfRepositoryService;
+
+	private final PcfService pcfService;
+
+	private final ProxyRequestInterface proxyRequestInterface;
+
+
+	public void pushPCFDataForApproveRequest(String processId, PolicyModel policy) {
+
+		List<String> accessBPNList = PolicyOperationUtil.getAccessBPNList(policy);
+
+		List<String> productList = pcfService.readCreatedTwins(processId).stream().map(PcfEntity::getProductId)
+				.toList();
+
+		markedPCFDataForPendingProviderRequestAsRequested(productList);
+
+		PagingResponse pcfData = pcfRepositoryService.getPcfData(
+				List.of(PCFRequestStatusEnum.PUSHED, PCFRequestStatusEnum.PUSHED_UPDATED_DATA), PCFTypeEnum.PROVIDER, 0,
+				1000);
+		List<PcfRequestModel> requestList = (List<PcfRequestModel>) pcfData.getItems();
+
+		if (!requestList.isEmpty()) {
+
+			requestList.forEach(request -> {
+
+				if (productList.contains(request.getProductId()) && accessBPNList.contains(request.getBpnNumber())) {
+
+					String msg = "";
+
+					try {
+						request.setStatus(PCFRequestStatusEnum.PUSHING_UPDATED_DATA);
+
+						JsonObject calculatedPCFValue = pcfService
+								.readCreatedTwinsDetailsByProductId(request.getProductId()).get("json")
+								.getAsJsonObject();
+
+						PCFRequestStatusEnum status = pcfRepositoryService.identifyRunningStatus(request.getRequestId(),
+								request.getStatus());
+
+						// push api call
+						Runnable runnable = () -> proxyRequestInterface.sendNotificationToConsumer(status,
+								calculatedPCFValue, request.getProductId(), request.getBpnNumber(),
+								request.getRequestId());
+
+						new Thread(runnable).start();
+
+						msg = "PCF request '" + request.getStatus()
+								+ "' and asynchronously sending notification to consumer";
+
+					} catch (NoDataFoundException e) {
+						msg = "Unable to take action on PCF request becasue pcf calculated value does not exist, please provide/upload PCF value for "
+								+ request.getProductId() + ", requestId " + request.getRequestId();
+						log.error("Async PushPCFDataForApproveRequest" + msg);
+						throw new NoDataFoundException(msg);
+					} catch (Exception e) {
+						pcfRepositoryService.savePcfStatus(request.getRequestId(), PCFRequestStatusEnum.FAILED);
+					}
+				}
+			});
+		} else {
+			log.debug("Async PushPCFDataForApproveRequest - No APPROVED request found");
+		}
+
+	}
+
+	public void markedPCFDataForPendingProviderRequestAsRequested(List<String> productList) {
+
+		PagingResponse pcfData = pcfRepositoryService.getPcfData(List.of(PCFRequestStatusEnum.PENDING_DATA_FROM_PROVIDER),
+				PCFTypeEnum.PROVIDER, 0, 1000);
+		List<PcfRequestModel> requestList = (List<PcfRequestModel>) pcfData.getItems();
+
+		if (!requestList.isEmpty()) {
+			requestList.forEach(request -> {
+				if (productList.contains(request.getProductId())) {
+					String msg = "";
+					try {
+						pcfService.readCreatedTwinsDetailsByProductId(request.getProductId()).get("json")
+								.getAsJsonObject();
+						pcfRepositoryService.savePcfStatus(request.getRequestId(), PCFRequestStatusEnum.REQUESTED);
+
+					} catch (NoDataFoundException e) {
+						msg = "Unable to markedPCFDataForPendingProviderRequestAsRequested becasue pcf calculated value does not exist  "
+								+ request.getProductId() + ", requestId " + request.getRequestId();
+						log.error("Async PushPCFDataForApproveRequest" + msg);
+						throw new NoDataFoundException(msg);
+					} catch (Exception e) {
+						pcfRepositoryService.savePcfStatus(request.getRequestId(), PCFRequestStatusEnum.FAILED);
+					}
+				}
+			});
+		} else {
+			log.debug("Async PushPCFDataForApproveRequest - No 'PENDING_DATA_FROM_PROVIDER' request found");
+		}
+
+	}
+}
diff --git a/...src/main/java/org/eclipse/tractusx/sde/pcfexchange/service/impl/PCFRepositoryService.java b/...src/main/java/org/eclipse/tractusx/sde/pcfexchange/service/impl/PCFRepositoryService.java
@@ -0,0 +1,149 @@
+/********************************************************************************
+ * Copyright (c) 2024 T-Systems International GmbH
+ * Copyright (c) 2024 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License, Version 2.0 which is available at
+ * https://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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ********************************************************************************/
+
+package org.eclipse.tractusx.sde.pcfexchange.service.impl;
+
+import java.time.Instant;
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+import org.eclipse.tractusx.sde.common.model.PagingResponse;
+import org.eclipse.tractusx.sde.pcfexchange.entity.PcfRequestEntity;
+import org.eclipse.tractusx.sde.pcfexchange.enums.PCFRequestStatusEnum;
+import org.eclipse.tractusx.sde.pcfexchange.enums.PCFTypeEnum;
+import org.eclipse.tractusx.sde.pcfexchange.mapper.PcfExchangeMapper;
+import org.eclipse.tractusx.sde.pcfexchange.repository.PcfRequestRepository;
+import org.eclipse.tractusx.sde.pcfexchange.request.PcfRequestModel;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.stereotype.Component;
+
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class PCFRepositoryService {
+
+	private static final String SUCCESS = "SUCCESS";
+	private final PcfRequestRepository pcfRequestRepository;
+	private final PcfExchangeMapper pcfMapper;
+
+	public PcfRequestModel savePcfRequestData(String requestId, String productId, String bpnNumber, String message,
+			PCFTypeEnum type, PCFRequestStatusEnum status, String remark) {
+
+		PcfRequestModel pojo = PcfRequestModel.builder().requestId(requestId).productId(productId).bpnNumber(bpnNumber)
+				.status(status).message(message).type(type).requestedTime(Instant.now().getEpochSecond())
+				.lastUpdatedTime(Instant.now().getEpochSecond()).remark(remark).build();
+
+		PcfRequestEntity entity = pcfMapper.mapFrom(pojo);
+
+		return pcfMapper.mapFrom(pcfRequestRepository.save(entity));
+	}
+
+	public PCFRequestStatusEnum updatePCFPushStatus(PCFRequestStatusEnum status, String requestId, String sendNotificationStatus) {
+
+		if ((PCFRequestStatusEnum.APPROVED.equals(status) || PCFRequestStatusEnum.PUSHING_DATA.equals(status))
+				&& SUCCESS.equalsIgnoreCase(sendNotificationStatus)) {
+			status = PCFRequestStatusEnum.PUSHED;
+			sendNotificationStatus ="PCF data successfuly pushed";
+		} else if (PCFRequestStatusEnum.PUSHING_UPDATED_DATA.equals(status)
+				&& SUCCESS.equalsIgnoreCase(sendNotificationStatus)) {
+			status = PCFRequestStatusEnum.PUSHED_UPDATED_DATA;
+			sendNotificationStatus ="PCF updated data successfuly pushed";
+		} else if (PCFRequestStatusEnum.REJECTED.equals(status) && SUCCESS.equalsIgnoreCase(sendNotificationStatus)) {
+			status = PCFRequestStatusEnum.REJECTED;
+			sendNotificationStatus ="PCF request rejected successfuly";
+		} else if (PCFRequestStatusEnum.APPROVED.equals(status)
+				|| PCFRequestStatusEnum.FAILED_TO_PUSH_DATA.equals(status)
+				|| PCFRequestStatusEnum.PUSHING_DATA.equals(status)
+				|| PCFRequestStatusEnum.PUSHING_UPDATED_DATA.equals(status)) {
+			status = PCFRequestStatusEnum.FAILED_TO_PUSH_DATA;
+		} else if (PCFRequestStatusEnum.REJECTED.equals(status)
+				|| PCFRequestStatusEnum.FAILED_TO_SEND_REJECT_NOTIFICATION.equals(status))
+			status = PCFRequestStatusEnum.FAILED_TO_SEND_REJECT_NOTIFICATION;
+		else {
+			status = PCFRequestStatusEnum.FAILED;
+		}
+
+		savePcfStatus(requestId, status, sendNotificationStatus);
+
+		return status;
+	}
+
+	public PCFRequestStatusEnum identifyRunningStatus(String requestId, PCFRequestStatusEnum status) {
+
+		boolean isApproval = PCFRequestStatusEnum.APPROVED.equals(status)
+				|| PCFRequestStatusEnum.FAILED_TO_PUSH_DATA.equals(status);
+
+		boolean isRejection = PCFRequestStatusEnum.REJECTED.equals(status)
+				|| PCFRequestStatusEnum.FAILED_TO_SEND_REJECT_NOTIFICATION.equals(status);
+
+		if (isApproval) {
+			status = PCFRequestStatusEnum.PUSHING_DATA;
+		} else if (isRejection) {
+			status = PCFRequestStatusEnum.SENDING_REJECT_NOTIFICATION;
+		}
+
+		savePcfStatus(requestId, status);
+
+		return status;
+	}
+
+	@SneakyThrows
+	public PcfRequestEntity savePcfStatus(String requestId, PCFRequestStatusEnum status) {
+		return savePcfStatus(requestId, status, null);
+	}
+
+	@SneakyThrows
+	public PcfRequestEntity savePcfStatus(String requestId, PCFRequestStatusEnum status, String remark) {
+
+		PcfRequestEntity pcfRequestEntity = pcfRequestRepository.getReferenceById(requestId);
+		pcfRequestEntity.setLastUpdatedTime(Instant.now().getEpochSecond());
+		pcfRequestEntity.setStatus(status);
+
+		if(StringUtils.isNotBlank(remark))
+			pcfRequestEntity.setRemark(remark);
+
+		log.info("'" + pcfRequestEntity.getProductId() + "' pcf request saved in the database successfully as {}",
+				status);
+		pcfRequestRepository.save(pcfRequestEntity);
+		return pcfRequestEntity;
+
+	}
+
+	public PagingResponse getPcfData(List<PCFRequestStatusEnum> status, PCFTypeEnum type, Integer page, Integer pageSize) {
+
+		Page<PcfRequestEntity> result = null;
+		if (status == null || status.isEmpty()) {
+			result = pcfRequestRepository.findByType(PageRequest.of(page, pageSize), type);
+		} else {
+			result = pcfRequestRepository.findByTypeAndStatusIn(PageRequest.of(page, pageSize), type, status);
+		}
+
+		List<PcfRequestModel> requestList = result.stream().map(pcfMapper::mapFrom).toList();
+
+		return PagingResponse.builder().items(requestList).pageSize(result.getSize()).page(result.getNumber())
+				.totalItems(result.getTotalElements()).build();
+	}
+
+}