Skip to content

Commit

Permalink
Remove awaitKeysQueueEmpty. The interrupt will do this using awaitTer…
Browse files Browse the repository at this point in the history 
…mination.
  • Loading branch information
@bernardladenthin
bernardladenthin committed Mar 25, 2024
1 parent ecdd290 commit 89731e9
Show file tree
Hide file tree
Showing 2 changed files with 21 additions and 121 deletions.
33 changes: 11 additions & 22 deletions src/main/java/net/ladenthin/bitcoinaddressfinder/ConsumerJava.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import com.google.common.annotations.VisibleForTesting;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
Expand Down Expand Up @@ -324,32 +325,20 @@ public void consumeKeys(PublicKeyBytes[] publicKeyBytes) throws InterruptedExcep
keysQueue.put(publicKeyBytes);
}

/**
* Returns if the consume was finished or.
* @return {@code true} if the keys queue is empty, otherwise {@code false}.
*/
@Deprecated
public boolean awaitKeysQueueEmpty(Duration maxWait) throws InterruptedException {
final long startTime = System.currentTimeMillis();
do {
if (keysQueue.isEmpty()) {
return true;
}
Thread.sleep(DURATION_CYCLIC_CHECK_KEYS_QUEUE_EMPTY);
} while (maxWait.minusMillis(System.currentTimeMillis() - startTime).isPositive());
return false;
}

@Override
public void interrupt() {
try {
// the result does not matter, just try to wait some seconds to empty the queue
awaitKeysQueueEmpty(AWAIT_DURATION_QUEUE_EMPTY);
} catch (InterruptedException ex) {
// do nothing, it is no problem
}
shouldRun.set(false);
scheduledExecutorService.shutdown();
consumeKeysExecutorService.shutdown();
try {
consumeKeysExecutorService.awaitTermination(AWAIT_DURATION_QUEUE_EMPTY.get(ChronoUnit.SECONDS), TimeUnit.SECONDS);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
}

@VisibleForTesting
int keysQueueSize() {
return keysQueue.size();
}
}
109 changes: 10 additions & 99 deletions src/test/java/net/ladenthin/bitcoinaddressfinder/ConsumerJavaTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@
import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
import net.ladenthin.bitcoinaddressfinder.configuration.CConsumerJava;
import net.ladenthin.bitcoinaddressfinder.configuration.CLMDBConfigurationReadOnly;
import net.ladenthin.bitcoinaddressfinder.configuration.CProducerJava;
Expand All @@ -57,7 +56,6 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.mockito.internal.matchers.StartsWith;
import org.slf4j.Logger;

@RunWith(DataProviderRunner.class)
Expand Down Expand Up @@ -125,25 +123,11 @@ public void startStatisticsTimer_invalidparameter_throwsException() throws IOExc
consumerJava.startStatisticsTimer();
}

@Test
public void waitTillKeysQueueEmpty_noKeysAdded_resultIsTrue() throws IOException, InterruptedException {
CConsumerJava cConsumerJava = new CConsumerJava();
ConsumerJava consumerJava = new ConsumerJava(cConsumerJava, keyUtility, persistenceUtils);
Logger logger = mock(Logger.class);
consumerJava.setLogger(logger);

// act
boolean result = consumerJava.awaitKeysQueueEmpty(Duration.ofSeconds(1L));

// assert
assertThat(result, is(equalTo(Boolean.TRUE)));
}

/**
* Attention, this is an await time test. This tests changes {@link ConsumerJava#AWAIT_DURATION_QUEUE_EMPTY}.
*/
@Test
public void interrupt_keysQueueNotEmpty_consumeNotRunningWaitedInternallyForTheDuration() throws IOException, InterruptedException, MnemonicException.MnemonicLengthException {
public void interrupt_keysQueueNotEmpty_consumerNotRunningWaitedInternallyForTheDuration() throws IOException, InterruptedException, MnemonicException.MnemonicLengthException {
// Attention: Change the duration.
ConsumerJava.AWAIT_DURATION_QUEUE_EMPTY = AwaitTimeTests.AWAIT_DURATION;

Expand All @@ -156,9 +140,17 @@ public void interrupt_keysQueueNotEmpty_consumeNotRunningWaitedInternallyForTheD
consumerJava.consumeKeys(createExamplePublicKeyBytesfromPrivateKey73());

// pre-assert, assert the keys queue is not empty
assertThat(consumerJava.awaitKeysQueueEmpty(Duration.ofMillis(1L)), is(equalTo(Boolean.FALSE)));
assertThat(consumerJava.keysQueueSize(), is(equalTo(1)));
assertThat(consumerJava.shouldRun.get(), is(equalTo(Boolean.TRUE)));

// add a pseudo thread to the executor to test its eecution duration
consumerJava.consumeKeysExecutorService.submit(() -> {
try {
Thread.sleep(ConsumerJava.AWAIT_DURATION_QUEUE_EMPTY);
} catch (InterruptedException ex) {
throw new RuntimeException(ex);
}
});
// act
long beforeAct = System.currentTimeMillis();
// the consume is not running and the interrupt must wait and release nevertheless
Expand Down Expand Up @@ -201,87 +193,6 @@ public void interrupt_statisticsTimerStarted_executerServiceShutdown() throws IO
assertThat(consumerJava.consumeKeysExecutorService.isShutdown(), is(equalTo(Boolean.TRUE)));
}

@Test
public void waitTillKeysQueueEmpty_keysAdded_resultIsFalse() throws IOException, InterruptedException, MnemonicException.MnemonicLengthException {
CConsumerJava cConsumerJava = new CConsumerJava();
ConsumerJava consumerJava = new ConsumerJava(cConsumerJava, keyUtility, persistenceUtils);
Logger logger = mock(Logger.class);
consumerJava.setLogger(logger);

// add keys
consumerJava.consumeKeys(createExamplePublicKeyBytesfromPrivateKey73());

// act
boolean result = consumerJava.awaitKeysQueueEmpty(Duration.ofMillis(1L));

// assert
assertThat(result, is(equalTo(Boolean.FALSE)));
}

@Test
@UseDataProvider(value = CommonDataProvider.DATA_PROVIDER_COMPRESSED_AND_STATIC_AMOUNT, location = CommonDataProvider.class)
public void waitTillKeysQueueEmpty_keysAddedDuringWait_resultIsTrue(boolean compressed, boolean useStaticAmount) throws IOException, InterruptedException, MnemonicException.MnemonicLengthException {
TestAddressesLMDB testAddressesLMDB = new TestAddressesLMDB();

TestAddressesFiles testAddresses = new TestAddressesFiles(compressed);
File lmdbFolderPath = testAddressesLMDB.createTestLMDB(folder, testAddresses, useStaticAmount, false);

CConsumerJava cConsumerJava = new CConsumerJava();
cConsumerJava.lmdbConfigurationReadOnly = new CLMDBConfigurationReadOnly();
cConsumerJava.lmdbConfigurationReadOnly.lmdbDirectory = lmdbFolderPath.getAbsolutePath();

ConsumerJava consumerJava = new ConsumerJava(cConsumerJava, keyUtility, persistenceUtils);
consumerJava.initLMDB();

Logger logger = mock(Logger.class);
consumerJava.setLogger(logger);

// add keys
consumerJava.consumeKeys(createExamplePublicKeyBytesfromPrivateKey73());

// key is in queue
boolean waitBefore = consumerJava.awaitKeysQueueEmpty(Duration.ofMillis(1L));
assertThat(waitBefore, is(equalTo(Boolean.FALSE)));

AtomicBoolean result = new AtomicBoolean();

AtomicBoolean waitStarted = new AtomicBoolean(false);

Thread threadWaitQueueEmpty = Thread.startVirtualThread( () -> {
waitStarted.set(true);
try {
// act
result.set(consumerJava.awaitKeysQueueEmpty(Duration.ofSeconds(2L)));
} catch (InterruptedException ex) {
// same value as before
result.set(false);
}
waitStarted.set(false);
});

Thread.sleep(Duration.ofSeconds(1L));

{
// assert the waitTillKeysQueueEmpty is running
assertThat(waitStarted.get(), is(equalTo(true)));
// assert the result is false
assertThat(result.get(), is(equalTo(Boolean.FALSE)));
}

// consume the keys now to empty the queue, the result must change to true
consumerJava.consumeKeys(createHash160ByteBuffer());

// wait for terminate
threadWaitQueueEmpty.join(Duration.ofSeconds(10L));

{
// assert the waitTillKeysQueueEmpty is not running
assertThat(waitStarted.get(), is(equalTo(false)));
// assert the result is true
assertThat(result.get(), is(equalTo(Boolean.TRUE)));
}
}

@Test
@UseDataProvider(value = CommonDataProvider.DATA_PROVIDER_COMPRESSED_AND_STATIC_AMOUNT, location = CommonDataProvider.class)
public void runProber_testAddressGiven_hitExpected(boolean compressed, boolean useStaticAmount) throws IOException, InterruptedException, MnemonicException.MnemonicLengthException {
Expand Down

0 comments on commit 89731e9

Please sign in to comment.