X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=policy-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fpolicyagent%2Ftasks%2FRicSynchronizationTask.java;h=a42135ef335e594696c45206ae9e27cf5c758fe2;hb=d7432fc246557dbda44b67d8f5693841a2eaba1b;hp=bcfda484dc124061a400790e8f2ec4b968966961;hpb=b66dcce5210e25b2571036becb6f0e7b0c23e1b2;p=nonrtric.git diff --git a/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSynchronizationTask.java b/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSynchronizationTask.java index bcfda484..a42135ef 100644 --- a/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSynchronizationTask.java +++ b/policy-agent/src/main/java/org/oransc/policyagent/tasks/RicSynchronizationTask.java @@ -22,13 +22,14 @@ package org.oransc.policyagent.tasks; import static org.oransc.policyagent.repository.Ric.RicState; -import java.util.Collection; import java.util.Vector; import org.oransc.policyagent.clients.A1Client; import org.oransc.policyagent.clients.A1ClientFactory; import org.oransc.policyagent.clients.AsyncRestClient; import org.oransc.policyagent.repository.ImmutablePolicyType; +import org.oransc.policyagent.repository.Lock; +import org.oransc.policyagent.repository.Lock.LockType; import org.oransc.policyagent.repository.Policies; import org.oransc.policyagent.repository.Policy; import org.oransc.policyagent.repository.PolicyType; @@ -43,13 +44,18 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** - * Synchronizes the content of a RIC with the content in the repository. - * This means: - * - load all policy types - * - send all policy instances to the RIC - * --- if that fails remove all policy instances - * - Notify subscribing services + * Synchronizes the content of a RIC with the content in the repository. This + * means: + *
+ * load all policy types + *
+ * send all policy instances to the RIC + *
+ * if that fails remove all policy instances + *
+ * Notify subscribing services */ +@SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally public class RicSynchronizationTask { private static final Logger logger = LoggerFactory.getLogger(RicSynchronizationTask.class); @@ -67,7 +73,7 @@ public class RicSynchronizationTask { this.services = services; } - @SuppressWarnings("squid:S2629") + @SuppressWarnings("squid:S2445") // Blocks should be synchronized on "private final" fields public void run(Ric ric) { logger.debug("Handling ric: {}", ric.getConfig().name()); @@ -78,7 +84,10 @@ public class RicSynchronizationTask { } ric.setState(RicState.SYNCHRONIZING); } - this.a1ClientFactory.createA1Client(ric)// + + ric.getLock().lock(LockType.EXCLUSIVE) // Make sure no NBI updates are running + .flatMap(Lock::unlock) // + .flatMap(lock -> this.a1ClientFactory.createA1Client(ric)) // .flatMapMany(client -> startSynchronization(ric, client)) // .subscribe(x -> logger.debug("Synchronize: {}", x), // throwable -> onSynchronizationError(ric, throwable), // @@ -87,19 +96,14 @@ public class RicSynchronizationTask { private Flux