diff options
Diffstat (limited to 'integrity-audit')
3 files changed, 115 insertions, 117 deletions
diff --git a/integrity-audit/src/main/java/org/onap/policy/common/ia/AuditThread.java b/integrity-audit/src/main/java/org/onap/policy/common/ia/AuditThread.java index f1839b12..58ed8b99 100644 --- a/integrity-audit/src/main/java/org/onap/policy/common/ia/AuditThread.java +++ b/integrity-audit/src/main/java/org/onap/policy/common/ia/AuditThread.java @@ -25,10 +25,7 @@ import java.util.Comparator; import java.util.Date; import java.util.List; import java.util.Properties; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; - import org.onap.policy.common.ia.jpa.IntegrityAuditEntity; import org.onap.policy.common.logging.eelf.MessageCodes; import org.onap.policy.common.logging.flexlogger.FlexLogger; @@ -108,18 +105,6 @@ public class AuditThread extends Thread { private IntegrityAudit integrityAudit; /** - * A latch is taken from this queue before starting an audit. May be {@code null}. Used by JUnit - * tests. - */ - private BlockingQueue<CountDownLatch> auditLatchQueue; - - /** - * Latch to be decremented when the next audit completes. May be {@code null}. Used by JUnit - * tests to wait for an audit to complete. - */ - private CountDownLatch auditCompletionLatch = null; - - /** * AuditThread constructor. * * @param resourceName the resource name @@ -133,7 +118,7 @@ public class AuditThread extends Thread { int integrityAuditPeriodSeconds, IntegrityAudit integrityAudit) throws IntegrityAuditException { this(resourceName, persistenceUnit, properties, TimeUnit.SECONDS.toMillis(integrityAuditPeriodSeconds), - integrityAudit, null); + integrityAudit); } /** @@ -148,13 +133,12 @@ public class AuditThread extends Thread { * @throws IntegrityAuditException if an error occurs */ public AuditThread(String resourceName, String persistenceUnit, Properties properties, long integrityAuditMillis, - IntegrityAudit integrityAudit, BlockingQueue<CountDownLatch> queue) throws IntegrityAuditException { + IntegrityAudit integrityAudit) throws IntegrityAuditException { this.resourceName = resourceName; this.persistenceUnit = persistenceUnit; this.properties = properties; this.integrityAuditPeriodMillis = integrityAuditMillis; this.integrityAudit = integrityAudit; - this.auditLatchQueue = queue; /* * The DbDAO Constructor registers this node in the IntegrityAuditEntity table. Each @@ -174,14 +158,8 @@ public class AuditThread extends Thread { logger.info("AuditThread.run: Entering"); try { - /* - * For JUnit testing: wait for the first latch, decrement it to indicate that the thread - * has started, and then wait for the next latch, before we actually start doing - * anything. These simply return if there is no latch queue defined. - */ - getNextLatch(); - decrementLatch(); - getNextLatch(); + // for junit testing + runStarted(); /* * Triggers change in designation, unless no other viable candidate. @@ -284,11 +262,8 @@ public class AuditThread extends Thread { * property, otherwise just sleep the normal interval. */ if (auditCompleted) { - // indicate that an audit has completed - decrementLatch(); - - // don't start the next audit cycle until a latch has been provided - getNextLatch(); + // for junit testing: indicate that an audit has completed + auditCompleted(); if (logger.isDebugEnabled()) { logger.debug("AuditThread.run: Audit completed; resourceName=" + this.resourceName @@ -342,29 +317,6 @@ public class AuditThread extends Thread { } /** - * Gets the next audit-completion latch from the queue. Blocks, if the queue is empty. - * - * @throws InterruptedException if interrupted while waiting - */ - private void getNextLatch() throws InterruptedException { - BlockingQueue<CountDownLatch> queue = this.auditLatchQueue; - if (queue != null) { - this.auditCompletionLatch = queue.take(); - } - } - - /** - * Decrements the current audit-completion latch, if any. - */ - private void decrementLatch() { - CountDownLatch latch = this.auditCompletionLatch; - if (latch != null) { - this.auditCompletionLatch = null; - latch.countDown(); - } - } - - /** * Determines if an exception is an InterruptedException or was caused by an * InterruptedException. * @@ -788,6 +740,26 @@ public class AuditThread extends Thread { } /** + * Indicates that the {@link #run()} method has started. This method simply returns, + * and may overridden by junit tests. + * + * @throws InterruptedException + */ + public void runStarted() throws InterruptedException { + // does nothing + } + + /** + * Indicates that an audit has completed. This method simply returns, and may + * overridden by junit tests. + * + * @throws InterruptedException + */ + public void auditCompleted() throws InterruptedException { + // does nothing + } + + /** * Adjusts the thread-sleep-interval to be used when an audit has <i>not</i> been completed. * Used by JUnit tests. * diff --git a/integrity-audit/src/main/java/org/onap/policy/common/ia/IntegrityAudit.java b/integrity-audit/src/main/java/org/onap/policy/common/ia/IntegrityAudit.java index 9abdbe52..b3330faf 100644 --- a/integrity-audit/src/main/java/org/onap/policy/common/ia/IntegrityAudit.java +++ b/integrity-audit/src/main/java/org/onap/policy/common/ia/IntegrityAudit.java @@ -21,9 +21,6 @@ package org.onap.policy.common.ia; import java.util.Properties; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CountDownLatch; - import org.onap.policy.common.ia.IntegrityAuditProperties.NodeTypeEnum; import org.onap.policy.common.logging.flexlogger.FlexLogger; import org.onap.policy.common.logging.flexlogger.Logger; @@ -217,37 +214,20 @@ public class IntegrityAudit { * @throws IntegrityAuditException if an error occurs */ public void startAuditThread() throws IntegrityAuditException { - startAuditThread(null); - } - - /** - * Starts the audit thread. - * - * @param queue the queue - * @return {@code true} if the thread was started, {@code false} otherwise - * @throws IntegrityAuditException if an error occurs - */ - protected boolean startAuditThread(BlockingQueue<CountDownLatch> queue) throws IntegrityAuditException { - logger.info("startAuditThread: Entering"); - boolean success = false; - if (integrityAuditPeriodMillis >= 0) { - this.auditThread = new AuditThread(this.resourceName, this.persistenceUnit, this.properties, - integrityAuditPeriodMillis, this, queue); + this.auditThread = makeAuditThread(this.resourceName, this.persistenceUnit, this.properties, integrityAuditPeriodMillis); logger.info("startAuditThread: Audit started and will run every " + integrityAuditPeriodMillis / 1000 + " seconds"); this.auditThread.start(); - success = true; + } else { logger.info("startAuditThread: Suppressing integrity audit, integrityAuditPeriodSeconds=" + integrityAuditPeriodMillis / 1000); } logger.info("startAuditThread: Exiting"); - - return success; } /** @@ -299,4 +279,29 @@ public class IntegrityAudit { return !this.auditThread.isAlive(); } } + + /** + * + * @return {@code true} if an audit thread exists, {@code false} otherwise + */ + protected boolean haveAuditThread() { + return (this.auditThread != null); + } + + /** + * Creates an audit thread. May be overridden by junit tests. + * + * @param resourceName2 + * @param persistenceUnit2 + * @param properties2 + * @param integrityAuditPeriodMillis2 + * + * @return a new audit thread + * @throws IntegrityAuditException + */ + protected AuditThread makeAuditThread(String resourceName2, String persistenceUnit2, Properties properties2, + long integrityAuditPeriodMillis2) throws IntegrityAuditException { + + return new AuditThread(resourceName2, persistenceUnit2, properties2, integrityAuditPeriodMillis2, this); + } } diff --git a/integrity-audit/src/test/java/org/onap/policy/common/ia/IntegrityAuditTestBase.java b/integrity-audit/src/test/java/org/onap/policy/common/ia/IntegrityAuditTestBase.java index afbcc452..c9179908 100644 --- a/integrity-audit/src/test/java/org/onap/policy/common/ia/IntegrityAuditTestBase.java +++ b/integrity-audit/src/test/java/org/onap/policy/common/ia/IntegrityAuditTestBase.java @@ -22,10 +22,6 @@ package org.onap.policy.common.ia; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; - -import ch.qos.logback.classic.Level; -import ch.qos.logback.classic.Logger; - import java.io.FileOutputStream; import java.io.IOException; import java.util.ArrayList; @@ -34,21 +30,19 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Properties; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; - import javax.persistence.EntityManager; import javax.persistence.EntityManagerFactory; import javax.persistence.EntityTransaction; import javax.persistence.Persistence; - import org.onap.policy.common.utils.jpa.EntityMgrCloser; import org.onap.policy.common.utils.jpa.EntityMgrFactoryCloser; import org.onap.policy.common.utils.jpa.EntityTransCloser; import org.onap.policy.common.utils.test.log.logback.ExtractAppender; import org.slf4j.LoggerFactory; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; /** * All JUnits are designed to run in the local development environment where they have write @@ -464,26 +458,26 @@ public class IntegrityAuditTestBase { protected void runAudit(MyIntegrityAudit... auditors) throws InterruptedException { // start an audit cycle on each auditor - List<CountDownLatch> latches = new ArrayList<>(auditors.length); + List<Semaphore> semaphores = new ArrayList<>(auditors.length); for (MyIntegrityAudit p : auditors) { - latches.add(p.startAudit()); + semaphores.add(p.startAudit()); } // wait for each auditor to complete its cycle - for (CountDownLatch latch : latches) { - waitLatch(latch); + for (Semaphore sem : semaphores) { + waitSem(sem); } } /** - * Waits for a latch to reach zero. + * Waits for a semaphore to be released. * - * @param latch the latch to wait for + * @param sem the semaphore for which to wait * @throws InterruptedException if the thread is interrupted * @throws AssertionError if the latch did not reach zero in the allotted time */ - protected void waitLatch(CountDownLatch latch) throws InterruptedException { - assertTrue(latch.await(WAIT_MS, TimeUnit.SECONDS)); + protected void waitSem(Semaphore sem) throws InterruptedException { + assertTrue(sem.tryAcquire(WAIT_MS, TimeUnit.SECONDS)); } /** @@ -520,11 +514,16 @@ public class IntegrityAuditTestBase { * Manages audits by inserting latches into a queue for the AuditThread to count. */ protected class MyIntegrityAudit extends IntegrityAudit { - + + /** + * Semaphore on which the audit thread should wait. + */ + private Semaphore auditSem = null; + /** - * Queue from which the AuditThread will take latches. + * Semaphore on which the junit management thread should wait. */ - private BlockingQueue<CountDownLatch> queue = null; + private Semaphore junitSem = null; /** * Constructs an auditor and starts the AuditThread. @@ -550,16 +549,14 @@ public class IntegrityAuditTestBase { } /** - * Triggers an audit by adding a latch to the queue. + * Triggers an audit by releasing the audit thread's semaphore. * - * @return the latch that was added + * @return the semaphore on which to wait * @throws InterruptedException if the thread is interrupted */ - public CountDownLatch startAudit() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - queue.add(latch); - - return latch; + public Semaphore startAudit() throws InterruptedException { + auditSem.release(); + return junitSem; } /** @@ -567,25 +564,23 @@ public class IntegrityAuditTestBase { */ @Override public final void startAuditThread() throws IntegrityAuditException { - if (queue != null) { - // queue up a bogus latch, in case a thread is still running - queue.add(new CountDownLatch(1) { - @Override - public void countDown() { - throw new RuntimeException("auditor has multiple threads"); - } - }); + if (auditSem != null) { + // release a bunch of semaphores, in case a thread is still running + auditSem.release(1000); } + + auditSem = new Semaphore(0); + junitSem = new Semaphore(0); + + super.startAuditThread(); - queue = new LinkedBlockingQueue<>(); + if (haveAuditThread()) { + // tell the thread it can run + auditSem.release(); - if (super.startAuditThread(queue)) { // wait for the thread to start - CountDownLatch latch = new CountDownLatch(1); - queue.add(latch); - try { - waitLatch(latch); + waitSem(junitSem); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -605,5 +600,31 @@ public class IntegrityAuditTestBase { assertTrue(waitThread(this)); } + + @Override + protected AuditThread makeAuditThread(String resourceName2, String persistenceUnit2, Properties properties2, + long integrityAuditPeriodMillis2) throws IntegrityAuditException { + + return new AuditThread(resourceName2, persistenceUnit2, properties2, integrityAuditPeriodMillis2, this) { + + private Semaphore auditSem = MyIntegrityAudit.this.auditSem; + private Semaphore junitSem = MyIntegrityAudit.this.junitSem; + + @Override + public void runStarted() throws InterruptedException { + auditSem.acquire(); + + junitSem.release(); + auditSem.acquire(); + } + + @Override + public void auditCompleted() throws InterruptedException { + junitSem.release(); + auditSem.acquire(); + } + + }; + } } } |