Reworked initialize function of broker tester

README.md

@@ -1,9 +1,9 @@
 # AmqpRunner
 [![Build Status](https://travis-ci.org/mmichailidis/AmqpRunner.svg?branch=master)](https://travis-ci.org/mmichailidis/AmqpRunner)
-
 [![codecov](https://codecov.io/gh/mmichailidis/AmqpRunner/branch/master/graph/badge.svg)](https://codecov.io/gh/mmichailidis/AmqpRunner)
-
 [![Maven Central](https://img.shields.io/maven-central/v/gr.mmichailidis/amqprunner.svg?label=Maven%20Central)](https://search.maven.org/search?q=g:%22gr.mmichailidis%22%20AND%20a:%22amqprunner%22)
+[![Codacy Badge](https://api.codacy.com/project/badge/Grade/a26f2d10a3384dec9832bec64fd082fa)](https://www.codacy.com/app/mmichailidis/AmqpRunner?utm_source=github.com&utm_medium=referral&utm_content=mmichailidis/AmqpRunner&utm_campaign=Badge_Grade)
+[![License](https://img.shields.io/github/license/mmichailidis/amqprunner.svg)](https://opensource.org/licenses/Apache-2.0)
 
 AmqpRunner is a JUnit runner that provides a fluent api for creating in 
 memory amqp and validating the packets that reached it. It extends 

src/main/java/gr/mmichaildis/amqprunner/BrokerManager.java

@@ -21,14 +21,19 @@
 import gr.mmichaildis.amqprunner.util.PortExtractingLauncherListener;
 import io.vavr.collection.Stream;
 import io.vavr.control.Option;
+import io.vavr.control.Try;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.qpid.server.SystemLauncher;
 import org.apache.qpid.server.SystemLauncherListener.DefaultSystemLauncherListener;
 
 import java.io.File;
 import java.net.URL;
+import java.util.Map;
 import java.util.*;
 
+import static gr.mmichaildis.amqprunner.util.StreamHelpers.not;
+import static gr.mmichaildis.amqprunner.util.StreamHelpers.replaceWith;
+import static io.vavr.API.*;
 import static java.lang.Thread.sleep;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
@@ -47,7 +52,11 @@ public class BrokerManager {
  private static final Long SLEEP_STEP = 100L;
  private final String introduction;
 
- private static ReferenceHolder refHolder;
+ /**
+ * The path that contains the configuration file for qpid initialization.
+ */
+ private static final String INITIAL_CONFIG_PATH = "amqp.json";
+ private static final ReferenceHolder refHolder = new ReferenceHolder();
 
  private final String username;
  private final String password;
@@ -82,8 +91,9 @@ public BrokerManager(final String username,
  this.requestedAmqpPort = requestedAmqpPort;
  this.requestedWorkPath = requestedWorkPath;
  this.requestedLogPath = requestedLogPath;
- refHolder = new ReferenceHolder();
- refHolder.setCleanUpList(Collections.synchronizedList(new LinkedList<>()));
+
+ refHolder.setQueueCleanUpList(Collections.synchronizedList(new LinkedList<>()));
+ refHolder.setExchangeCleanUpList(Collections.synchronizedList(new LinkedList<>()));
 
  introduction = "[BrokerManager" + (name.isEmpty() ? "" : "-" + name) + "] ";
  // this.systemLauncher = new SystemLauncher();
@@ -97,11 +107,6 @@ public BrokerManager(final String username,
  this.uuid = UUID.randomUUID();
  }
 
- /**
- * The path that contains the configuration file for qpid initialization.
- */
- private static final String INITIAL_CONFIG_PATH = "amqp.json";
- private static final String INITIAL_CONFIG_PATH_NETWORK = "amqpNetwork.json";
 
  /**
  * Start the broker with the properties that was initialized with.
@@ -160,15 +165,15 @@ public void stopBroker() {
  systemLauncher.shutdown();
  log.info("SystemLauncher shutdown complete. Cleaning up.");
 
- File db = new File(requestedWorkPath + uuid);
- File log = new File(requestedLogPath + uuid);
+ final File db = new File(requestedWorkPath + uuid);
+ final File log = new File(requestedLogPath + uuid);
 
  Stream.of(db, log)
  .filter(File::exists)
  .forEach(BrokerManager::deleteFolder);
 
  try {
- Thread.sleep(5000);
+ sleep(5000);
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
@@ -200,22 +205,27 @@ private static void deleteFolder(File folder) {
 
  private static void deleteFile(File file, Integer retryStep) {
  try {
- Thread.sleep(2500 * retryStep);
+ sleep(2500 * retryStep);
  } catch (InterruptedException e) {
  e.printStackTrace();
  }
- if (!file.delete() && retryStep < 3) {
- deleteFile(file, retryStep + 1);
- } else {
- log.error("File {} failed to be deleted after {} retries", file, retryStep);
- }
+ Try.run(() -> sleep(2500 * retryStep))
+ .filter(ignore -> retryStep < 3)
+ .map(ignore -> file.delete())
+ .filter(not(Boolean::booleanValue))
+ .map(replaceWith(retryStep < 3))
+ .forEach(hasMoreSteps -> Match(hasMoreSteps).of(
+ Case($(true), run(() -> deleteFile(file, retryStep + 1))),
+ Case($(false), run(() -> log.error("File {} failed to be deleted after {} retries", file, retryStep)))
+ ));
  }
 
  /**
  * Cleans up all the queues and exchanges in the broker.
  */
  public void cleanUp() {
- refHolder.getCleanUpList().forEach(r -> r.apply(null));
+ refHolder.getQueueCleanUpList().forEach(r -> r.apply(null));
+ refHolder.getExchangeCleanUpList().forEach(r -> r.apply(null));
  }
 
  private Map<String, Object> createSystemConfig() {

src/main/java/gr/mmichaildis/amqprunner/broker/AssertionTime.java

@@ -42,9 +42,9 @@ final class AssertionTime {
  * set the target to {@code totalMillis} and reset the counter.
  *
  * @param totalMillis The millis that will be either setup as the new target or will put the current target ahead.
- * @param override {@code true} to override the target, {@code false} to push ahead.
+ * @param override  {@code true} to override the target, {@code false} to push ahead.
  */
- void updateTotalSeconds(Long totalMillis, Boolean override) {
+ protected void updateTotalSeconds(Long totalMillis, Boolean override) {
  if (override) {
  this.totalSeconds.set(totalMillis);
  currentMillis.set(0L);
@@ -60,7 +60,7 @@ void updateTotalSeconds(Long totalMillis, Boolean override) {
  * @param sleepStep the amount of time to update the counter.
  * @return The difference between the counter and the target.
  */
- Long updateCurrentMillis(Long sleepStep) {
+ protected Long updateCurrentMillis(Long sleepStep) {
  return totalSeconds.get() - currentMillis.updateAndGet(curr -> curr + sleepStep);
  }
 }

src/main/java/gr/mmichaildis/amqprunner/broker/AssertionVerification.java

@@ -69,8 +69,8 @@ public boolean isValid() {
  * the {@link AssertionError} found in all the registered assertions.
  *
  * @return the {@link List} which contains all the {@link AssertionError} found.
- *  If the {@link AssertionVerification#isValid()} was not called before
- *  this then the {@link List} is empty.
+ * If the {@link AssertionVerification#isValid()} was not called before
+ * this then the {@link List} is empty.
  */
  public List<AssertionError> getErrors() {
  return assertionErrors;
@@ -82,7 +82,7 @@ public List<AssertionError> getErrors() {
  * @param identifier The assertions identifier.
  * @param valid The valid status of the identified assertion.
  */
- void setValid(String identifier, boolean valid) {
+ protected void setValid(String identifier, boolean valid) {
  final AssertionPair pair = this.response.get(identifier);
  pair.setValid(valid);
  this.response.put(identifier, pair);
@@ -94,7 +94,7 @@ void setValid(String identifier, boolean valid) {
  * @param identifier The assertions identifier.
  * @param e The {@link AssertionError} that will be registered for the identified assertion.
  */
- void setAssertionError(String identifier, AssertionError e) {
+ protected void setAssertionError(String identifier, AssertionError e) {
  final AssertionPair pair = this.response.get(identifier);
  pair.setAssertionError(e);
  this.response.put(identifier, pair);
@@ -106,7 +106,7 @@ void setAssertionError(String identifier, AssertionError e) {
  *
  * @param identifier The identifier that will identify each assertion.
  */
- void register(String identifier, String userProvidedIdentifier) {
+ protected void register(String identifier, String userProvidedIdentifier) {
  response.put(identifier, AssertionPair.initial(userProvidedIdentifier));
  block.put(identifier, new AssertionTime(0));
  }
@@ -117,9 +117,9 @@ void register(String identifier, String userProvidedIdentifier) {
  *
  * @param nextName The assertion name which will be used to create the unique identifier.
  * @return The unique identifier which should be used for registration and identification on
- *  the main {@link Thread}.
+ * the main {@link Thread}.
  */
- String getNextName(String nextName) {
+ protected String getNextName(String nextName) {
  Integer identifier = nameProvider.getOrDefault(nextName, 0);
  identifier += 1;
  nameProvider.put(nextName, identifier);
@@ -131,7 +131,7 @@ String getNextName(String nextName) {
  *
  * @param identifier The assertions identifier.
  */
- void clearAssertionError(String identifier) {
+ protected void clearAssertionError(String identifier) {
  final AssertionPair pair = response.get(identifier);
  pair.setAssertionError(null);
  response.put(identifier, pair);
@@ -143,7 +143,7 @@ void clearAssertionError(String identifier) {
  * @param identifier The identifier that will have it sleep increased.
  * @param sleepTime The N <b>seconds</b> that will be increase.
  */
- void updateBlock(String identifier, Integer sleepTime) {
+ protected void updateBlock(String identifier, Integer sleepTime) {
  this.block.get(identifier)
  .updateTotalSeconds(sleepTime * SECOND_TO_MILLIS, false);
  }
@@ -166,7 +166,7 @@ void setBlock(String identifier, Integer sleepTime) {
  *
  * @param sleepStep the step in time
  * @return {@code true} if at least one assertion requires more time. {@code false} if and only if all
- *  the assertions are completed and responds they require no more time.
+ * the assertions are completed and responds they require no more time.
  */
  public boolean shouldBlock(Long sleepStep) {
  return block.values().stream()