X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=policy-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Foransc%2Fpolicyagent%2Ftasks%2FRicSynchronizationTask.java;h=9d25fa34f25b05c230739ea294e3b870f2cac019;hb=refs%2Fchanges%2F65%2F2665%2F3;hp=bcfda484dc124061a400790e8f2ec4b968966961;hpb=842b9d220588fba7fc17df0cf9c094f91005118b;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..9d25fa34 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,13 @@ 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.LockType; import org.oransc.policyagent.repository.Policies; import org.oransc.policyagent.repository.Policy; import org.oransc.policyagent.repository.PolicyType; @@ -43,12 +43,16 @@ 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 */ public class RicSynchronizationTask { @@ -78,7 +82,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 -> lock.unlock()) // + .flatMap(lock -> this.a1ClientFactory.createA1Client(ric)) // .flatMapMany(client -> startSynchronization(ric, client)) // .subscribe(x -> logger.debug("Synchronize: {}", x), // throwable -> onSynchronizationError(ric, throwable), // @@ -87,13 +94,9 @@ public class RicSynchronizationTask { private Flux