X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=policy-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fpolicyagent%2Ftasks%2FRicSupervision.java;h=228038afa10bac5f849ecfae35e250ba7ebc88d7;hb=6a39814272307d0207222c9229b0d765ac062bf0;hp=2666d6060f6f865c0cfae637fb734c4f9a7e459c;hpb=d375c04a9c1917a9bf98285fb1f30062cff7a6d6;p=nonrtric.git

diff --git a/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSupervision.java b/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSupervision.java
index 2666d606..228038af 100644
--- a/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSupervision.java
+++ b/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSupervision.java
@@ -24,6 +24,7 @@ import java.util.Collection;
 
 import org.oransc.policyagent.clients.A1Client;
 import org.oransc.policyagent.clients.A1ClientFactory;
+import org.oransc.policyagent.exceptions.ServiceException;
 import org.oransc.policyagent.repository.Lock.LockType;
 import org.oransc.policyagent.repository.Policies;
 import org.oransc.policyagent.repository.PolicyTypes;
@@ -58,6 +59,28 @@ public class RicSupervision {
     private final A1ClientFactory a1ClientFactory;
     private final Services services;
 
+    private static class SynchStartedException extends ServiceException {
+        private static final long serialVersionUID = 1L;
+
+        public SynchStartedException(String message) {
+            super(message);
+        }
+    }
+
+    private static class RicData {
+        RicData(Ric ric, A1Client a1Client) {
+            this.ric = ric;
+            this.a1Client = a1Client;
+        }
+
+        A1Client getClient() {
+            return a1Client;
+        }
+
+        final Ric ric;
+        private final A1Client a1Client;
+    }
+
     @Autowired
     public RicSupervision(Rics rics, Policies policies, A1ClientFactory a1ClientFactory, PolicyTypes policyTypes,
         Services services) {
@@ -80,30 +103,47 @@ public class RicSupervision {
     private Flux<RicData> createTask() {
         return Flux.fromIterable(rics.getRics()) //
             .flatMap(this::createRicData) //
-            .flatMap(this::checkOneRic) //
-            .onErrorResume(throwable -> Mono.empty());
-
+            .flatMap(this::checkOneRic);
     }
 
     private Mono<RicData> checkOneRic(RicData ricData) {
         return checkRicState(ricData) //
             .flatMap(x -> ricData.ric.getLock().lock(LockType.EXCLUSIVE)) //
+            .flatMap(notUsed -> setRicState(ricData)) //
             .flatMap(x -> checkRicPolicies(ricData)) //
-            .flatMap(x -> ricData.ric.getLock().unlock()) //
-            .doOnError(throwable -> ricData.ric.getLock().unlockBlocking()) //
             .flatMap(x -> checkRicPolicyTypes(ricData)) //
-            .doOnNext(x -> logger.debug("Ric: {} checked OK", ricData.ric.name())) //
-            .doOnError(t -> logger.debug("Ric: {} check Failed, exception: {}", ricData.ric.name(), t.getMessage()));
+            .doOnNext(x -> onRicCheckedOk(ricData)) //
+            .doOnError(t -> onRicCheckedError(t, ricData)) //
+            .onErrorResume(throwable -> Mono.empty());
     }
 
-    private static class RicData {
-        RicData(Ric ric, A1Client a1Client) {
-            this.ric = ric;
-            this.a1Client = a1Client;
+    private void onRicCheckedError(Throwable t, RicData ricData) {
+        logger.debug("Ric: {} check stopped, exception: {}", ricData.ric.name(), t.getMessage());
+        if (t instanceof SynchStartedException) {
+            // this is just a temporary state,
+            ricData.ric.setState(RicState.AVAILABLE);
+        } else {
+            ricData.ric.setState(RicState.UNAVAILABLE);
         }
+        ricData.ric.getLock().unlockBlocking();
+    }
 
-        final Ric ric;
-        final A1Client a1Client;
+    private void onRicCheckedOk(RicData ricData) {
+        logger.debug("Ric: {} checked OK", ricData.ric.name());
+        ricData.ric.setState(RicState.AVAILABLE);
+        ricData.ric.getLock().unlockBlocking();
+    }
+
+    @SuppressWarnings("squid:S2445") // Blocks should be synchronized on "private final" fields
+    private Mono<RicData> setRicState(RicData ric) {
+        synchronized (ric) {
+            if (ric.ric.getState() == RicState.CONSISTENCY_CHECK) {
+                logger.debug("Ric: {} is already being checked", ric.ric.getConfig().name());
+                return Mono.empty();
+            }
+            ric.ric.setState(RicState.CONSISTENCY_CHECK);
+            return Mono.just(ric);
+        }
     }
 
     private Mono<RicData> createRicData(Ric ric) {
@@ -116,7 +156,7 @@ public class RicSupervision {
         if (ric.ric.getState() == RicState.UNAVAILABLE) {
             return startSynchronization(ric) //
                 .onErrorResume(t -> Mono.empty());
-        } else if (ric.ric.getState() == RicState.SYNCHRONIZING) {
+        } else if (ric.ric.getState() == RicState.SYNCHRONIZING || ric.ric.getState() == RicState.CONSISTENCY_CHECK) {
             return Mono.empty();
         } else {
             return Mono.just(ric);
@@ -124,7 +164,7 @@ public class RicSupervision {
     }
 
     private Mono<RicData> checkRicPolicies(RicData ric) {
-        return ric.a1Client.getPolicyIdentities() //
+        return ric.getClient().getPolicyIdentities() //
             .flatMap(ricP -> validateInstances(ricP, ric));
     }
 
@@ -144,7 +184,7 @@ public class RicSupervision {
     }
 
     private Mono<RicData> checkRicPolicyTypes(RicData ric) {
-        return ric.a1Client.getPolicyTypeIdentities() //
+        return ric.getClient().getPolicyTypeIdentities() //
             .flatMap(ricTypes -> validateTypes(ricTypes, ric));
     }
 
@@ -163,7 +203,7 @@ public class RicSupervision {
     private Mono<RicData> startSynchronization(RicData ric) {
         RicSynchronizationTask synchronizationTask = createSynchronizationTask();
         synchronizationTask.run(ric.ric);
-        return Mono.error(new Exception("Syncronization started"));
+        return Mono.error(new SynchStartedException("Syncronization started"));
     }
 
     RicSynchronizationTask createSynchronizationTask() {