public class DownloadException extends RuntimeException {

  public DownloadException(String msg) {
    super(msg);
  }

  public DownloadException(String msg, Throwable cause) {
    super(msg, cause);
  }
} public interface Downloader extends Runnable {

  @Override
  void run() throws DownloadException;

  abstract class Decorator implements Downloader {

    final Downloader innerDownloader;

    public Decorator(Downloader innerDownloader) {
      this.innerDownloader = innerDownloader;
    }
  }
} public class DownloaderFactory {

  EmbeddedRabbitMqConfig config;

  public DownloaderFactory(EmbeddedRabbitMqConfig config) {
    this.config = config;
  }

  /**
   * @return an appropriate instance depending on the given configuration.
   */
  public Downloader getNewInstance() {
    Downloader downloader = new BasicDownloader(config);
    if (config.shouldCachedDownload()) {
      downloader = new CachedDownloader(downloader, config);
    }
    return downloader;
  }

} public class ExtractionException extends RuntimeException {

  public ExtractionException(String message) {
    super(message);
  }

  public ExtractionException(String message, Throwable cause) {
    super(message, cause);
  }
} public interface Extractor extends Runnable {

  @Override
  void run() throws ExtractionException;

  abstract class Decorator implements Extractor {

    protected final Extractor innerExtractor;

    public Decorator(Extractor innerExtractor) {
      this.innerExtractor = innerExtractor;
    }
  }
} public class ExtractorFactory {

  private EmbeddedRabbitMqConfig config;

  public ExtractorFactory(EmbeddedRabbitMqConfig config) {
    this.config = config;
  }

  /**
   * Returns an Extractor instance appropriate based on the given configuration.
   */
  public Extractor getNewInstance() {
    Extractor extractor = new BasicExtractor(config);
    if (config.shouldCachedDownload()) {
      extractor = new CachedExtractor(extractor, config);
    }
    return extractor;
  }

} public class ErlangVersionChecker {

  private static final Logger LOGGER = LoggerFactory.getLogger(ErlangVersionChecker.class);

  private final ErlangShell erlangShell;
  private final String minErlangVersion;

  public ErlangVersionChecker(EmbeddedRabbitMqConfig config) {
    this(config.getVersion().getMinimumErlangVersion(), new ErlangShell(config));
  }

  public ErlangVersionChecker(String minErlangVersion, ErlangShell erlangShell) {
    this.minErlangVersion = minErlangVersion;
    this.erlangShell = erlangShell;
  }

  /**
   * Retrieves the current system's Erlang version to compare it to the minimum required version.
   * <p>
   * The system's Erlang version is always retrieved, but the comparison might be skipped if the RabbitMQ version
   * doesn't specify a minimum required version.
   *
   * @throws ErlangVersionException if the minimum required version is not met or if it can't be determined.
   */
  public void check() throws ErlangVersionException {
    String erlangVersion;
    try {
      erlangVersion = erlangShell.getErlangVersion();
      LOGGER.debug("Erlang version installed in this system: {}", erlangVersion);
    } catch (ErlangShellException e) {
      throw new ErlangVersionException("Could not determine Erlang version. Ensure Erlang is correctly installed.", e);
    }

    if (minErlangVersion == null) {
      LOGGER.debug("RabbitMQ version to execute doesn't specify a minimum Erlang version. Will skip this check.");
      return;
    } else {
      LOGGER.debug("RabbitMQ version to execute requires Erlang version {} or above.", minErlangVersion);
    }

    int[] expected;
    int[] actual;
    try {
      expected = parse(minErlangVersion);
      actual = parse(erlangVersion);
    } catch (RuntimeException e) {
      LOGGER.warn("Error parsing Erlang version: " + minErlangVersion + " or " + erlangVersion + ". Ignoring check...");
      return;
    }

    for (int i = 0; i < actual.length; i++) {
      if (actual[i] > expected[i]) {
        break;
      }
      if (actual[i] < expected[i]) {
        throw new ErlangVersionException(
            String.format("Minimum required Erlang version not found. Expected '%s' or higher. Actual is: '%s'",
                minErlangVersion, erlangVersion));
      }
    }
  }

  /**
   * @return a numeric value useful for comparing versions.
   */
  static int[] parse(String erlangVersion) {
    int[] version = {0, 0, 0, 0, 0};
    if (erlangVersion.startsWith("r") || erlangVersion.startsWith("R") ) {
      erlangVersion = erlangVersion.substring(1);                     // "R15B03-1" -> "15B03-1"
      String[] components = erlangVersion.split("\\D", 2);            // "15B03-1" -> ["15", "03-1"]
      version[0] = Integer.parseInt(components[0]);                        // "15" -> 15
      version[1] = erlangVersion.toUpperCase(Locale.US)
                                .replaceAll("[^A-Z]", "").charAt(0);       // "15B03-1-" -> "B" -> 66
      if (components.length >= 2
          && !components[1].isEmpty()
          && components[1].indexOf("-") != 0) {
        version[2] = Integer.parseInt(components[1].split("-", 2)[0]);     // "03-1" -> "03" -> 3
      }
    } else {
      String[] components = erlangVersion.split("\\.", 5);
      for (int i = 0; i < components.length; i++) {
        version[i] = Integer.parseInt(components[i]);
      }
    }
    return version;
  }
} public class ErlangVersionException extends RuntimeException {

  public ErlangVersionException(String message) {
    super(message);
  }

  public ErlangVersionException(String message, Throwable cause) {
    super(message, cause);
  }
} public class ShutDownException extends RabbitMqCommandException {

  public ShutDownException(String msg, Throwable cause) {
    super(msg, cause);
  }
} public class ShutdownHelper implements Runnable {

  private static final Logger LOGGER = LoggerFactory.getLogger(ShutdownHelper.class);

  private final EmbeddedRabbitMqConfig config;
  private final Future<ProcessResult> rabbitMqProcess;
  private final long timeoutDuration;
  private final TimeUnit timeoutUnit;

  /**
   * Constructs a new instance that will be used to shut down the given RabbitMQ server process.
   */
  public ShutdownHelper(EmbeddedRabbitMqConfig config, Future<ProcessResult> rabbitMqProcess) {
    this.config = config;
    this.rabbitMqProcess = rabbitMqProcess;
    this.timeoutDuration = config.getDefaultRabbitMqCtlTimeoutInMillis();
    this.timeoutUnit = TimeUnit.MILLISECONDS;
  }

  @Override
  public void run() throws ShutDownException {
    submitShutdownRequest();
    confirmShutdown();
  }

  private void submitShutdownRequest() throws ShutDownException {
    Future<ProcessResult> resultFuture;
    try {
      resultFuture = new RabbitMqCtl(config).stop();
    } catch (RabbitMqCommandException e) {
      throw new ShutDownException("Could not successfully execute command to stop RabbitMQ Server", e);
    }

    int exitValue;
    try {
      ProcessResult rabbitMqCtlProcessResult = resultFuture.get(timeoutDuration, timeoutUnit);
      exitValue = rabbitMqCtlProcessResult.getExitValue();
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
      throw new ShutDownException("Error while waiting " + timeoutDuration + " " + timeoutUnit + " for command "
          + "to shut down RabbitMQ Server to finish", e);
    }

    if (exitValue == 0) {
      LOGGER.debug("Successfully commanded RabbitMQ Server to stop.");
    } else {
      LOGGER.warn("Command to stop RabbitMQ Sever failed with exit value: " + exitValue);
    }
  }

  private void confirmShutdown() throws ShutDownException {
    int exitValue;
    try {
      ProcessResult rabbitMqProcessResult = rabbitMqProcess.get(timeoutDuration, TimeUnit.MILLISECONDS);
      exitValue = rabbitMqProcessResult.getExitValue();
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
      throw new ShutDownException("Error while waiting " + timeoutDuration + " " + timeoutUnit + "for "
          + "RabbitMQ Server to shut down", e);
    }

    if (exitValue == 0) {
      LOGGER.debug("RabbitMQ Server stopped successfully.");
    } else {
      LOGGER.warn("RabbitMQ Server stopped with exit value: " + exitValue);
    }
  }
} public class StartupException extends RabbitMqCommandException {

  public StartupException(String msg) {
    super(msg);
  }

  public StartupException(String msg, Throwable cause) {
    super(msg, cause);
  }
} public class StartupHelper implements Callable<Future<ProcessResult>> {

  public static final String BROKER_STARTUP_COMPLETED = ".*completed with \\d+ plugins.*";
  private final EmbeddedRabbitMqConfig config;

  public StartupHelper(EmbeddedRabbitMqConfig config) {
    this.config = config;
  }

  /**
   * Starts the RabbitMQ Server and blocks the current thread until the server is confirmed to have started.
   * <p>
   * This is useful to ensure no other interactions happen with the RabbitMQ Server until it's safe to do so
   *
   * @return an unfinished future representing the eventual result of the {@code rabbitmq-server} process running in "foreground".
   * @throws StartupException if anything fails while attempting to start and confirm successful initialization.
   * @see ShutdownHelper
   */
  @Override
  public Future<ProcessResult> call() throws StartupException {
    PatternFinderOutputStream initializationWatcher = new PatternFinderOutputStream(BROKER_STARTUP_COMPLETED);

    // Inform the initializationWatcher if the process ends before the expected output is produced.
    PublishingProcessListener rabbitMqProcessListener = new PublishingProcessListener();
    rabbitMqProcessListener.addSubscriber(initializationWatcher);

    Future<ProcessResult> resultFuture = startProcess(initializationWatcher, rabbitMqProcessListener);
    waitForConfirmation(initializationWatcher);

    return resultFuture;
  }

  private Future<ProcessResult> startProcess(PatternFinderOutputStream initializationWatcher,
                                             PublishingProcessListener rabbitMqProcessListener) {
    Future<ProcessResult> resultFuture;
    try {
      resultFuture = new RabbitMqServer(config)
          .writeOutputTo(initializationWatcher)
          .listeningToEventsWith(rabbitMqProcessListener)
          .start();
    } catch (RabbitMqCommandException e) {
      throw new StartupException("Could not start RabbitMQ Server", e);
    }
    return resultFuture;
  }

  private void waitForConfirmation(PatternFinderOutputStream initializationWatcher) {
    long timeout = config.getRabbitMqServerInitializationTimeoutInMillis();
    boolean match = initializationWatcher.waitForMatch(timeout, TimeUnit.MILLISECONDS);

    if (!match) {
      throw new StartupException(
          "Could not confirm RabbitMQ Server initialization completed successfully within " + timeout + "ms");
    }
  }

  /**
   * Notifies subscribers of process termination so they don't have to rely on blocking {@link Future#get()} of
   * {@link ProcessResult}s, which is returned by {@link RabbitMqCommand}s.
   */
  static class PublishingProcessListener extends ProcessListener {

    interface Subscriber {
      void processFinished(int exitValue);
    }

    private final List<Subscriber> subscribers;

    public PublishingProcessListener(Subscriber... subscribers) {
      this.subscribers = new ArrayList<>(Arrays.asList(subscribers));
    }

    @Override
    public void afterFinish(Process process, ProcessResult result) {
      super.afterFinish(process, result);
      for (Subscriber subscriber : subscribers) {
        subscriber.processFinished(result.getExitValue());
      }
    }

    public void addSubscriber(Subscriber subscriber) {
      this.subscribers.add(subscriber);
    }

  }

  /**
   * An output stream that compares each line with a given pattern.
   *
   * This class offers the ability to wait until the pattern is found or the given amount of time has passed.
   */
  static class PatternFinderOutputStream extends LogOutputStream implements PublishingProcessListener.Subscriber {

    private static final Logger LOGGER = LoggerFactory.getLogger(PatternFinderOutputStream.class);

    private final Pattern pattern;
    private final Semaphore lock;
    private boolean matchFound;

    public PatternFinderOutputStream(String initializationMarkerPattern) {
      this(Pattern.compile(initializationMarkerPattern, Pattern.CASE_INSENSITIVE));
    }

    public PatternFinderOutputStream(Pattern initializationMarkerPattern) {
      try {
        lock = new Semaphore(1);
        lock.acquire();
      } catch (InterruptedException e) {
        throw new IllegalStateException("Could not acquire a lock we create right above?", e);
      }
      pattern = initializationMarkerPattern;
      matchFound = false;
    }

    @Override
    protected void processLine(String line) {
      if (pattern.matcher(line).matches()) {
        LOGGER.trace("Pattern '{}' found in line: {}", pattern, line);
        matchFound = true;
        lock.release();
      }
      LOGGER.trace("Pattern '{}' NOT found in line: {}", pattern, line);
    }

    @Override
    public void processFinished(int exitValue) {
      LOGGER.debug("No more output is expected since process finished (exit code: {})", exitValue);
      lock.release();
    }

    public boolean waitForMatch(long duration, TimeUnit timeUnit) {
      try {
        boolean acquired = lock.tryAcquire(duration, timeUnit);
        if (!acquired) {
          LOGGER.info("Waited for {} {} for pattern '{}' to appear but it didn't.", duration, timeUnit, pattern );
        }
      } catch (InterruptedException e) {
        LOGGER.warn("Error while waiting for process output that matches the pattern '{}'", pattern);
      }
      return isMatchFound();
    }

    public boolean isMatchFound() {
      return matchFound;
    }
  }
}
import io.arivera.oss.embedded.rabbitmq.download.DownloadException;
import io.arivera.oss.embedded.rabbitmq.download.Downloader;
import io.arivera.oss.embedded.rabbitmq.download.DownloaderFactory;
import io.arivera.oss.embedded.rabbitmq.extract.ExtractionException;
import io.arivera.oss.embedded.rabbitmq.extract.Extractor;
import io.arivera.oss.embedded.rabbitmq.extract.ExtractorFactory;
import io.arivera.oss.embedded.rabbitmq.helpers.ErlangVersionChecker;
import io.arivera.oss.embedded.rabbitmq.helpers.ErlangVersionException;
import io.arivera.oss.embedded.rabbitmq.helpers.ShutDownException;
import io.arivera.oss.embedded.rabbitmq.helpers.ShutdownHelper;
import io.arivera.oss.embedded.rabbitmq.helpers.StartupException;
import io.arivera.oss.embedded.rabbitmq.helpers.StartupHelper;
import org.zeroturnaround.exec.ProcessResult;
import java.util.concurrent.Future;
package io.arivera.oss.embedded.rabbitmq;




/**
 * This is the main class to interact with RabbitMQ.
 * <p>
 * Example use:
 * <pre>
 * {@code
 *   EmbeddedRabbitMqConfig config = new EmbeddedRabbitMqConfig.Builder().build();
 *   EmbeddedRabbitMq rabbitMq = new EmbeddedRabbitMq(config);
 *   rabbitMq.start();
 *   // ...
 *   rabbitMq.stop();
 * }
 * </pre>
 *
 * @see EmbeddedRabbitMqConfig
 */
public class EmbeddedRabbitMq {

  private EmbeddedRabbitMqConfig config;
  private Future<ProcessResult> rabbitMqProcess;

  public EmbeddedRabbitMq(EmbeddedRabbitMqConfig config) {
    this.config = config;
  }

  /**
   * Starts the RabbitMQ server process and blocks the current thread until the initialization is completed.
   *
   * @throws ErlangVersionException when there's an issue with the system's Erlang version
   * @throws DownloadException    when there's an issue downloading the appropriate artifact
   * @throws ExtractionException  when there's an issue extracting the files from the downloaded artifact
   * @throws StartupException     when there's an issue starting the RabbitMQ server
   */
  public void start() throws ErlangVersionException, DownloadException, ExtractionException, StartupException {