package forklift;

import static org.kohsuke.args4j.ExampleMode.ALL;

import consul.Consul;
import forklift.connectors.ActiveMQConnector;
import forklift.consumer.ConsumerDeploymentEvents;
import forklift.consumer.LifeCycleMonitors;
import forklift.deployment.DeploymentWatch;
import forklift.replay.ReplayLogger;
import forklift.retry.RetryHandler;
import org.kohsuke.args4j.CmdLineException;
import org.kohsuke.args4j.CmdLineParser;

import java.io.File;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Start Forklift as a server.
 * @author zdavep
 */
public final class ForkliftServer {
    // Lock Waits
    private static final AtomicBoolean running = new AtomicBoolean(false);

    // Logging
    private static org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(ForkliftServer.class);

    // Consumer deployment interval
    private static int SLEEP_INTERVAL = 10000; // 10 seconds

    /**
     * Launch a Forklift server instance.
     */
    public static void main(String[] args) throws Throwable {
        final ForkliftOpts opts = new ForkliftOpts();
        final CmdLineParser argParse = new CmdLineParser(opts);
        try {
            argParse.parseArgument(args);
        } catch( CmdLineException e ) {
            // if there's a problem in the command line,
            // you'll get this exception. this will report
            // an error message.
            System.err.println(e.getMessage());
            System.err.println("java SampleMain [options...] arguments...");
            // print the list of available options
            argParse.printUsage(System.err);
            System.err.println();

            // print option sample. This is useful some time
            System.err.println("  Example: java SampleMain" + argParse.printExample(ALL));

            return;
        }

        String brokerUrl = opts.getBrokerUrl();
        if (brokerUrl.startsWith("consul.") && brokerUrl.length() > "consul.".length()) {
            log.info("Building failover url using consul");

            final Consul c = new Consul("http://" + opts.getConsulHost(), 8500);

            // Build the connection string.
            final String serviceName = brokerUrl.split("\\.")[1];

            brokerUrl = "failover:(" +
                c.catalog().service(serviceName).getProviders().stream()
                    .filter(srvc -> !srvc.isCritical())
                    .map(srvc -> "tcp://" + srvc.getAddress() + ":" + srvc.getPort())
                    .reduce("", (a, b) -> a + "," + b) +
                ")";

            c.shutdown();

            brokerUrl = brokerUrl.replaceAll("failover:\\(,", "failover:(");

            log.info("url {}", brokerUrl);
            if (brokerUrl.equals("failover:()")) {
                log.error("No brokers found");
                System.exit(-1);
            }
        }

        // Start a forklift server w/ specified connector.
        final Forklift forklift = new Forklift();
        final ConsumerDeploymentEvents deploymentEvents = new ConsumerDeploymentEvents(forklift);

        final DeploymentWatch deploymentWatch;
        if (opts.getConsumerDir() != null)
            deploymentWatch = new DeploymentWatch(new java.io.File(opts.getConsumerDir()), deploymentEvents);
        else
            deploymentWatch = null;

        final DeploymentWatch propsWatch;
        if (opts.getPropsDir() != null)
            propsWatch = new DeploymentWatch(new java.io.File(opts.getPropsDir()), deploymentEvents);
        else
            propsWatch = null;

        forklift.start(new ActiveMQConnector(brokerUrl));
        if (!forklift.isRunning()) {
            throw new RuntimeException("Unable to start Forklift");
        }

        log.info("Registering LifeCycleMonitors");
        LifeCycleMonitors.register(new RetryHandler(forklift.getConnector(), new File(opts.getRetryDir())));
        LifeCycleMonitors.register(new ReplayLogger(new File(opts.getReplayDir())));

        log.info("Connected to broker on " + brokerUrl);
        log.info("Scanning for Forklift consumers at " + opts.getConsumerDir());
        log.info("Scanning for Forklift consumers at " + opts.getPropsDir());

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                // End the deployment watcher.
                running.set(false);
                synchronized (running) {
                    running.notify();
                }

                if (deploymentWatch != null)
                    deploymentWatch.shutdown();

                if (propsWatch != null)
                    propsWatch.shutdown();

                forklift.shutdown();
            }
        });

        running.set(true);
        while (running.get()) {
            log.debug("Scanning for new deployments...");

            try {
                if (deploymentWatch != null)
                    deploymentWatch.run();
            } catch (Throwable e) {
                log.error("", e);
            }

            try {
                if (propsWatch != null)
                    propsWatch.run();
            } catch (Throwable e) {
                log.error("", e);
            }

            synchronized (running) {
                running.wait(SLEEP_INTERVAL);
            }
        }
    }
}

@@ -342,16 +342,66 @@
 Handler;
+%0A%0Aimport org.apache.activemq.broker.BrokerService;
 %0Aimport 
@@ -951,16 +951,65 @@
 econds%0A%0A
+    private static BrokerService broker = null;%0A%0A
     /**%0A
@@ -2942,32 +2942,345 @@
      %7D%0A        %7D
+ else if (brokerUrl.startsWith(%22embed%22)) %7B%0A            brokerUrl = %22tcp://localhost:61616%22;%0A            broker = new BrokerService();%0A%0A            // configure the broker%0A            broker.addConnector(brokerUrl);%0A            broker.addConnector(%22stomp://localhost:61613%22);%0A%0A            broker.start();%0A        %7D
 %0A%0A        // Sta
@@ -5054,32 +5054,32 @@
 ch.shutdown();%0A%0A
-
                 
@@ -5094,24 +5094,178 @@
 .shutdown();
+%0A%0A                if (broker != null)%0A                    try %7B%0A                        broker.stop();%0A                    %7D catch (Exception ignored) %7B %7D
 %0A           

Add embedded broker for quickstart.