RateControl { protected long sleepToMaintainRate(long duration) throws InterruptedException, DatabusException { if (duration <= 0 ) { throw new DatabusException("Negative duration specified"); } long remainingTimeNSec = duration; long remainingTimeMSec = remainingTimeNSec / DbusConstants.NUM_NSECS_IN_MSEC; remainingTimeNSec -= (remainingTimeMSec * DbusConstants.NUM_NSECS_IN_MSEC); if (remainingTimeNSec > 0) remainingTimeMSec++; boolean needToSleep = (remainingTimeMSec > 0); if (needToSleep) { if (LOG.isDebugEnabled()) { LOG.debug("Sleeping for remainingTimeMSec = " + remainingTimeMSec); } _ra.sleep(remainingTimeMSec); _numSleeps++; } else { if (LOG.isDebugEnabled()) { LOG.debug("No need to sleep. remainingTimeMSec = " + remainingTimeMSec); } } return remainingTimeMSec; } RateControl(long numEventsPerSec, long throttleDurationInSecs); long incrementEventCount(); boolean isEnabled(); long getNumSleeps(); void resetNumSleeps(); static final String MODULE; static final Logger LOG; }
@Test public void testSleepToMaintainRate() throws Exception { RateControl rc = new RateControl(2,1); long duration = 10000; long msecToSleep = rc.sleepToMaintainRate(duration); Assert.assertEquals(msecToSleep, 1L); duration = 100000; msecToSleep = rc.sleepToMaintainRate(duration); Assert.assertEquals(msecToSleep, 1L); duration = 1000000; msecToSleep = rc.sleepToMaintainRate(duration); Assert.assertEquals(msecToSleep, 1L); duration = 100000000; msecToSleep = rc.sleepToMaintainRate(duration); Assert.assertEquals(msecToSleep, 100L); duration = 1000000000; msecToSleep = rc.sleepToMaintainRate(duration); Assert.assertEquals(msecToSleep, 1000L); }