refactor: Decouple services that needed for monitoring (alarms) and based...

refactor: Decouple services that needed for monitoring (alarms) and based common policy services. Didn't change any of th existing implementation.

src/policy/src/main/java/org/etsi/tfs/policy/policy/AddPolicyDeviceImpl.java

+package org.etsi.tfs.policy.policy;
+
+import static org.etsi.tfs.policy.common.ApplicationProperties.INVALID_MESSAGE;
+import static org.etsi.tfs.policy.common.ApplicationProperties.VALIDATED_POLICYRULE_STATE;
+
+import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.groups.UniJoin;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.List;
+import org.etsi.tfs.policy.context.ContextService;
+import org.etsi.tfs.policy.exception.ExternalServiceFailureException;
+import org.etsi.tfs.policy.policy.model.PolicyRule;
+import org.etsi.tfs.policy.policy.model.PolicyRuleBasic;
+import org.etsi.tfs.policy.policy.model.PolicyRuleDevice;
+import org.etsi.tfs.policy.policy.model.PolicyRuleState;
+import org.etsi.tfs.policy.policy.model.PolicyRuleStateEnum;
+import org.etsi.tfs.policy.policy.model.PolicyRuleTypeDevice;
+import org.etsi.tfs.policy.policy.service.PolicyRuleConditionValidator;
+
+@ApplicationScoped
+public class AddPolicyDeviceImpl {
+
+    @Inject private PolicyRuleConditionValidator policyRuleConditionValidator;
+
+    @Inject private CommonPolicyServiceImpl commonPolicyServiceImpl;
+    @Inject private CommonAlarmService commonAlarmService;
+
+    @Inject private ContextService contextService;
+
+    public Uni<List<Boolean>> returnInvalidDeviceIds(List<String> deviceIds) {
+        UniJoin.Builder<Boolean> builder = Uni.join().builder();
+        for (String deviceId : deviceIds) {
+            final var validatedDeviceId = policyRuleConditionValidator.isDeviceIdValid(deviceId);
+            builder.add(validatedDeviceId);
+        }
+        return builder.joinAll().andFailFast();
+    }
+
+    public Uni<PolicyRuleState> areDeviceOnContext(
+            List<Boolean> areDevices,
+            PolicyRuleDevice policyRuleDevice,
+            PolicyRuleBasic policyRuleBasic) {
+        if (areDevices.contains(false)) {
+            var policyRuleState =
+                    new PolicyRuleState(
+                            PolicyRuleStateEnum.POLICY_FAILED,
+                            String.format(
+                                    INVALID_MESSAGE, policyRuleDevice.getPolicyRuleBasic().getPolicyRuleId()));
+
+            return Uni.createFrom().item(policyRuleState);
+        }
+
+        final var policyRuleTypeDevice = new PolicyRuleTypeDevice(policyRuleDevice);
+        final var policyRule = new PolicyRule(policyRuleTypeDevice);
+
+        final var alarmDescriptorList = commonPolicyServiceImpl.createAlarmDescriptorList(policyRule);
+        if (alarmDescriptorList.isEmpty()) {
+            var policyRuleState =
+                    new PolicyRuleState(
+                            PolicyRuleStateEnum.POLICY_FAILED,
+                            String.format(
+                                    "Invalid PolicyRuleConditions in PolicyRule with ID: %s",
+                                    policyRuleBasic.getPolicyRuleId()));
+            return Uni.createFrom().item(policyRuleState);
+        }
+
+        return contextService
+                .setPolicyRule(policyRule)
+                .onFailure()
+                .transform(failure -> new ExternalServiceFailureException(failure.getMessage()))
+                .onItem()
+                .transform(
+                        policyId -> {
+                            commonAlarmService.startMonitoringBasedOnAlarmDescriptors(
+                                    policyId, policyRuleDevice, alarmDescriptorList);
+                            return VALIDATED_POLICYRULE_STATE;
+                        });
+    }
+}

src/policy/src/main/java/org/etsi/tfs/policy/policy/AddPolicyServiceImpl.java

+package org.etsi.tfs.policy.policy;
+
+import static org.etsi.tfs.policy.common.ApplicationProperties.INVALID_MESSAGE;
+import static org.etsi.tfs.policy.common.ApplicationProperties.VALIDATED_POLICYRULE_STATE;
+
+import io.smallrye.mutiny.Uni;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.List;
+import org.etsi.tfs.policy.context.ContextService;
+import org.etsi.tfs.policy.context.model.ServiceId;
+import org.etsi.tfs.policy.exception.NewException;
+import org.etsi.tfs.policy.monitoring.model.AlarmDescriptor;
+import org.etsi.tfs.policy.policy.model.PolicyRule;
+import org.etsi.tfs.policy.policy.model.PolicyRuleBasic;
+import org.etsi.tfs.policy.policy.model.PolicyRuleService;
+import org.etsi.tfs.policy.policy.model.PolicyRuleState;
+import org.etsi.tfs.policy.policy.model.PolicyRuleStateEnum;
+import org.etsi.tfs.policy.policy.model.PolicyRuleTypeService;
+import org.jboss.logging.Logger;
+
+@ApplicationScoped
+public class AddPolicyServiceImpl {
+    private static final Logger LOGGER = Logger.getLogger(AddPolicyServiceImpl.class);
+
+    @Inject private CommonPolicyServiceImpl commonPolicyService;
+    @Inject private CommonAlarmService commonAlarmService;
+    @Inject private ContextService contextService;
+
+    public Uni<PolicyRuleState> constructPolicyStateBasedOnCriteria(
+            Boolean isService,
+            ServiceId serviceId,
+            PolicyRuleService policyRuleService,
+            PolicyRuleBasic policyRuleBasic) {
+
+        if (!isService) {
+            var policyRuleState =
+                    new PolicyRuleState(
+                            PolicyRuleStateEnum.POLICY_FAILED, String.format(INVALID_MESSAGE, serviceId));
+
+            return Uni.createFrom().item(policyRuleState);
+        }
+
+        final var policyRuleTypeService = new PolicyRuleTypeService(policyRuleService);
+        final var policyRule = new PolicyRule(policyRuleTypeService);
+        final var alarmDescriptorList = commonPolicyService.createAlarmDescriptorList(policyRule);
+
+        if (alarmDescriptorList.isEmpty()) {
+            var policyRuleState =
+                    new PolicyRuleState(
+                            PolicyRuleStateEnum.POLICY_FAILED,
+                            String.format(
+                                    "Invalid PolicyRuleConditions in PolicyRule with ID: %s",
+                                    policyRuleBasic.getPolicyRuleId()));
+            return Uni.createFrom().item(policyRuleState);
+        }
+
+        return setPolicyRuleOnContextAndReturnState(policyRule, policyRuleService, alarmDescriptorList);
+    }
+
+    private Uni<PolicyRuleState> setPolicyRuleOnContextAndReturnState(
+            PolicyRule policyRule,
+            PolicyRuleService policyRuleService,
+            List<AlarmDescriptor> alarmDescriptorList) {
+        return contextService
+                .setPolicyRule(policyRule)
+                .onFailure()
+                .transform(failure -> new NewException(failure.getMessage()))
+                .onItem()
+                .transform(
+                        policyId -> {
+                            commonAlarmService.startMonitoringBasedOnAlarmDescriptors(
+                                    policyId, policyRuleService, alarmDescriptorList);
+
+                            return VALIDATED_POLICYRULE_STATE;
+                        });
+    }
+}

src/policy/src/main/java/org/etsi/tfs/policy/policy/CommonAlarmService.java

+package org.etsi.tfs.policy.policy;
+
+import static org.etsi.tfs.policy.common.ApplicationProperties.PROVISIONED_POLICYRULE_STATE;
+import static org.etsi.tfs.policy.common.ApplicationProperties.VALIDATED_POLICYRULE_STATE;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.subscription.Cancellable;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.List;
+import org.etsi.tfs.policy.monitoring.MonitoringService;
+import org.etsi.tfs.policy.monitoring.model.AlarmDescriptor;
+import org.etsi.tfs.policy.monitoring.model.AlarmResponse;
+import org.etsi.tfs.policy.monitoring.model.AlarmSubscription;
+import org.etsi.tfs.policy.policy.model.PolicyRuleDevice;
+import org.etsi.tfs.policy.policy.model.PolicyRuleService;
+import org.jboss.logging.Logger;
+
+@ApplicationScoped
+public class CommonAlarmService {
+    private static final Logger LOGGER = Logger.getLogger(CommonAlarmService.class);
+
+    @Inject private CommonPolicyServiceImpl commonPolicyServiceImpl;
+    @Inject private MonitoringService monitoringService;
+
+    public void startMonitoringBasedOnAlarmDescriptors(
+            String policyId,
+            PolicyRuleDevice policyRuleDevice,
+            List<AlarmDescriptor> alarmDescriptorList) {
+        commonPolicyServiceImpl.setPolicyRuleDeviceToContext(
+                policyRuleDevice, VALIDATED_POLICYRULE_STATE);
+        commonPolicyServiceImpl.noAlarms = 0;
+
+        List<Uni<String>> alarmIds = createAlarmList(alarmDescriptorList);
+
+        List<Multi<AlarmResponse>> alarmResponseStreamList =
+                transformAlarmIds(alarmIds, policyRuleDevice);
+
+        // Merge the promised alarms into one stream (Multi Object)
+        final var multi = Multi.createBy().merging().streams(alarmResponseStreamList);
+        commonPolicyServiceImpl.setPolicyRuleDeviceToContext(
+                policyRuleDevice, PROVISIONED_POLICYRULE_STATE);
+
+        commonPolicyServiceImpl
+                .getSubscriptionList()
+                .put(policyId, monitorAlarmResponseForDevice(multi));
+
+        // TODO: Resubscribe to the stream, if it has ended
+
+        // TODO: Redesign evaluation of action
+        // evaluateAction(policyRule, alarmDescriptorList, multi);
+    }
+
+    public void startMonitoringBasedOnAlarmDescriptors(
+            String policyId,
+            PolicyRuleService policyRuleService,
+            List<AlarmDescriptor> alarmDescriptorList) {
+        commonPolicyServiceImpl.setPolicyRuleServiceToContext(
+                policyRuleService, VALIDATED_POLICYRULE_STATE);
+        commonPolicyServiceImpl.noAlarms = 0;
+
+        List<Uni<String>> alarmIds =
+                createAlarmList(alarmDescriptorList); // setAllarmtomonitoring get back alarmid
+
+        List<Multi<AlarmResponse>> alarmResponseStreamList =
+                transformAlarmIds(alarmIds, policyRuleService);
+
+        // Merge the promised alarms into one stream (Multi Object)
+        final var multi = Multi.createBy().merging().streams(alarmResponseStreamList);
+        commonPolicyServiceImpl.setPolicyRuleServiceToContext(
+                policyRuleService, PROVISIONED_POLICYRULE_STATE);
+
+        commonPolicyServiceImpl
+                .getSubscriptionList()
+                .put(policyId, monitorAlarmResponseForService(multi));
+
+        // TODO: Resubscribe to the stream, if it has ended
+
+        // TODO: Redesign evaluation of action
+        // evaluateAction(policyRule, alarmDescriptorList, multi);
+    }
+
+    /**
+    * Transform the alarmIds into promised alarms returned from the getAlarmResponseStream
+    *
+    * @param alarmIds the list of alarm ids
+    * @param policyRuleService the policy rule service
+    * @return
+    */
+    private List<Multi<AlarmResponse>> transformAlarmIds(
+            List<Uni<String>> alarmIds, PolicyRuleService policyRuleService) {
+        List<Multi<AlarmResponse>> alarmResponseStreamList = new ArrayList<>();
+        for (Uni<String> alarmId : alarmIds) {
+            Multi<AlarmResponse> alarmResponseStream =
+                    alarmId.onItem().transformToMulti(id -> setPolicyMonitor(policyRuleService, id));
+
+            alarmResponseStreamList.add(alarmResponseStream);
+        }
+        return alarmResponseStreamList;
+    }
+
+    private List<Multi<AlarmResponse>> transformAlarmIds(
+            List<Uni<String>> alarmIds, PolicyRuleDevice policyRuleDevice) {
+        // Transform the alarmIds into promised alarms returned from the
+        // getAlarmResponseStream
+        List<Multi<AlarmResponse>> alarmResponseStreamList = new ArrayList<>();
+        for (Uni<String> alarmId : alarmIds) {
+            alarmResponseStreamList.add(
+                    alarmId.onItem().transformToMulti(id -> setPolicyMonitor(policyRuleDevice, id)));
+        }
+        return alarmResponseStreamList;
+    }
+
+    private Multi<AlarmResponse> setPolicyMonitor(PolicyRuleService policyRuleService, String id) {
+        commonPolicyServiceImpl.getAlarmPolicyRuleServiceMap().put(id, policyRuleService);
+
+        // TODO: Create infinite subscription
+        var alarmSubscription = new AlarmSubscription(id, 259200, 5000);
+        return monitoringService.getAlarmResponseStream(alarmSubscription);
+    }
+
+    private Multi<AlarmResponse> setPolicyMonitor(PolicyRuleDevice policyRuleDevice, String id) {
+        commonPolicyServiceImpl.getAlarmPolicyRuleDeviceMap().put(id, policyRuleDevice);
+
+        // TODO: Create infinite subscription
+        var alarmSubscription = new AlarmSubscription(id, 259200, 5000);
+        return monitoringService.getAlarmResponseStream(alarmSubscription);
+    }
+
+    /**
+    * Create an alarmIds list that contains the promised ids returned from setKpiAlarm
+    *
+    * @param alarmDescriptorList the list of alarm descriptors
+    * @return the list of alarm descriptors
+    */
+    public List<Uni<String>> createAlarmList(List<AlarmDescriptor> alarmDescriptorList) {
+        List<Uni<String>> alarmIds = new ArrayList<Uni<String>>();
+        for (AlarmDescriptor alarmDescriptor : alarmDescriptorList) {
+            LOGGER.infof("alarmDescriptor:");
+            LOGGER.infof(alarmDescriptor.toString());
+            alarmIds.add(monitoringService.setKpiAlarm(alarmDescriptor));
+        }
+        return alarmIds;
+    }
+
+    private Cancellable monitorAlarmResponseForService(Multi<AlarmResponse> multi) {
+        return multi
+                .subscribe()
+                .with(
+                        alarmResponse -> {
+                            LOGGER.infof("**************************Received Alarm!**************************");
+                            LOGGER.infof("alarmResponse:");
+                            LOGGER.info(alarmResponse);
+                            LOGGER.info(alarmResponse.getAlarmId());
+                            commonPolicyServiceImpl.applyActionService(alarmResponse.getAlarmId());
+                        });
+    }
+
+    private Cancellable monitorAlarmResponseForDevice(Multi<AlarmResponse> multi) {
+        return multi
+                .subscribe()
+                .with(
+                        alarmResponse -> {
+                            LOGGER.infof("**************************Received Alarm!**************************");
+                            LOGGER.infof("alarmResponse:");
+                            LOGGER.info(alarmResponse);
+                            LOGGER.info(alarmResponse.getAlarmId());
+                            commonPolicyServiceImpl.applyActionDevice(alarmResponse.getAlarmId());
+                        });
+    }
+}

src/policy/src/main/java/org/etsi/tfs/policy/policy/DeletePolicyImpl.java

+package org.etsi.tfs.policy.policy;
+
+public class DeletePolicyImpl {}