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=6110c58c4371223fb7d8d9cf98d0df221ecdae75;hb=61fcee9bad6d96005c403fdfd3f273430d4f8641;hp=d75999111532c9fd916a3cb76a4884c31b9b9d93;hpb=9013ed7ad46ce6927fbf69890487e8df61b7d7ee;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 d7599911..6110c58c 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 @@ -28,6 +28,7 @@ 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; @@ -54,6 +55,7 @@ import reactor.core.publisher.Mono; *
* Notify subscribing services */ +@SuppressWarnings("squid:S2629") // Invoke method(s) only conditionally public class RicSynchronizationTask { private static final Logger logger = LoggerFactory.getLogger(RicSynchronizationTask.class); @@ -71,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()); @@ -82,9 +84,10 @@ public class RicSynchronizationTask { } ric.setState(RicState.SYNCHRONIZING); } - ric.getLock().lockBlocking(LockType.EXCLUSIVE); // Make sure no NBI updates are running - ric.getLock().unlock(); - 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), // @@ -92,16 +95,15 @@ public class RicSynchronizationTask { } private Flux