2016-07 Cool Stuff

plugins {
    id "us.kirchmeier.capsule" version "1.0.2"
}

task fatCapsule(type: FatCapsule) {
    archiveName = "grabbit-cli-${version}"
    applicationClass "com.twcable.grabbit.tools.cli.GrabbitCli"

    capsuleManifest {
        minJavaVersion = defaults.compatibilityVersion
    }

    manifest = jar.manifest
    reallyExecutable
}

assemble.dependsOn fatCapsule

public class HostAndJobIds {
    URI uri;
    Publisher<Long> jobIds;

    @java.beans.ConstructorProperties({"uri", "jobIds"})
    public HostAndJobIds(URI uri, Publisher<Long> jobIds) {
        this.uri = uri;
        this.jobIds = jobIds;
    }

    public URI uri() {
        return this.uri;
    }

    public Publisher<Long> jobIds() {
        return this.jobIds;
    }

    public boolean equals(Object o) {
        if (o == this) return true;
        if (!(o instanceof HostAndJobIds)) return false;
        final HostAndJobIds other = (HostAndJobIds)o;
        final Object this$uri = this.uri;
        final Object other$uri = other.uri;
        if (this$uri == null ? other$uri != null :
            !this$uri.equals(other$uri)) return false;
        final Object this$jobIds = this.jobIds;
        final Object other$jobIds = other.jobIds;
        if (this$jobIds == null ? other$jobIds != null :
            !this$jobIds.equals(other$jobIds)) return false;
        return true;
    }

    public int hashCode() {
        final int PRIME = 59;
        int result = 1;
        final Object $uri = this.uri;
        result = result * PRIME + ($uri == null ? 0 : $uri.hashCode());
        final Object $jobIds = this.jobIds;
        result = result * PRIME + ($jobIds == null ? 0 : $jobIds.hashCode());
        return result;
    }

    public String toString() {
        return "com.twcable.grabbit.tools.jobstarter.HostAndJobIds(uri=" +
            this.uri + ", jobIds=" + this.jobIds + ")";
    }
}

public static JobStatus fromJson(URI uri, @Nullable String jsonStr) {
    val json = (jsonStr == null || jsonStr.trim().isEmpty()) ? "{}" : jsonStr;

    val map = (Map<String, Object>)new Yaml().loadAs(json, Map.class);
    val transactionId = (Long)map.getOrDefault("transactionID", -1L);
    val jobExecutionId = (Long)map.getOrDefault("jobExecutionId", -1L);
    val startTimeStr = (String)map.getOrDefault("startTime", DATE_TIME_FORMATTER.format(now()));
    val startTime = OffsetDateTime.parse(startTimeStr, DATE_TIME_FORMATTER);
    val endTimeStr = (String)map.get("endTime");
    val endTime = (endTimeStr != null) ?
        OffsetDateTime.parse(endTimeStr, DATE_TIME_FORMATTER) : (OffsetDateTime)null;
    val path = (String)map.getOrDefault("path", "/MISSING_PATH");
    val existStatusMap = (Map<String, Object>)map.getOrDefault("exitStatus", emptyMap());
    val exitDescription = (String)existStatusMap.getOrDefault("exitDescription", "");
    val exitCode = (String)existStatusMap.getOrDefault("exitCode", "UNKNOWN");
    val running = (Boolean)existStatusMap.getOrDefault("running", Boolean.FALSE);
    val timeTaken = Long.valueOf(map.getOrDefault("timeTaken", -1L).toString());
    val jcrNodesWritten = Long.parseLong(map.getOrDefault("jcrNodesWritten", -1L).toString());

    return new JobStatus(uri, transactionId, jobExecutionId, startTime, endTime, path, timeTaken,
        jcrNodesWritten, exitDescription, exitCode, running);
}

public interface Stream<T> {
    <R> Stream<R> map(Function<T, R> mapper);
    <R> Stream<R> flatMap(Function<T, Stream<R>> mapper);
}

public @Nullable HostJobState put(HostJobState entry) {
    val existing = entries().stream().
        filter(e -> match(e, entry)).
        findAny().orElse(null);

    if (existing == null || !existing.state().equals(entry.state())) {
        if (existing == null) { // new entry
            this.hostJobStates = new ArrayList<>(entries());
            hostJobStates.add(entry);
        }
        else { // modified entry
            hostJobStates = entries().stream().
                map(e -> match(e, entry) ? entry : e).
                collect(Collectors.toList());
        }
    }

    return existing;
}

private void parseStartJobsOutput(String startJobsOutput,
                                  Subscriber<Long> jobIdSub) {
    // the output from starting a job looks like "[123,125]"
    val matcher = JOB_IDS_PATTERN.matcher(startJobsOutput);
    if (matcher.matches()) {
        val jobIdsStr = (@NonNull String)matcher.group("jobIds");
        Arrays.stream(jobIdsStr.split(",")).
            map(String::trim).
            map(Long::valueOf).
            forEach(jobId -> jobIdSub.onNext(jobId));
        jobIdSub.onComplete();
    }
    else {
        jobIdSub.onError(new IllegalStateException("Could not parse job " +
            "ids from: " + startJobsOutput));
    }
}

default Stream<E> stream() {
    return StreamSupport.stream(spliterator(), false);
}

default Stream<E> parallelStream() {
    return StreamSupport.stream(spliterator(), true);
}

/**
 * Convert the String into an Optional that only has value if the
 * input was both non-null and non-empty.
 */
public static Optional<String> opt(String val) {
    return Optional.ofNullable(val).filter(s -> !s.isEmpty());
}

public String getIssuesUrl() {
    return opt(issuesUrl).orElseGet(() -> getSiteUrl() + "/issues");
}

public String getBintrayPkg() {
    return opt(bintrayPkg).orElse(project.getName());
}

try {
    // ...
}
catch (IOException e) {
    throw new RuntimeException(e);
}

/**
 * Remove checked-ness from the exception. The same exception is returned
 * (checked or unchecked), but this removes the compiler's checks.
 */
public static <T extends Throwable> RuntimeException softened(T exp) {
    return Utils.<RuntimeException>uncheck(exp);
}

private static <T extends Throwable> T uncheck(Throwable throwable) throws T {
    throw (T)throwable;
}

protected void writeFile() {
    Writer fileWriter = null;
    try {
        // ...
    }
    catch (IOException e) {
        throw softened(e);
    }
    finally {
        Utils.flushAndClose(fileWriter);
    }
}

/**
 * An either/or (XOR) value represented as Left/Right.
 *
 * By convention, Left is an error value whereas Right is a success value.
 */
public interface Either<L, R> extends Supplier<R> {
    boolean isLeft();

    L getLeft();

    boolean isRight();

    R get();

    static <L, R> Left<L, R> left(L value) {
        return new Left<>(value);
    }

    static <L, R> Right<L, R> right(R value) {
        return new Right<>(value);
    }
    // ...
}

final class CliOptions {
    /**
     * Parse the arguments.
     *
     * @param args the command line arguments
     * @return Right if successfully parsed; otherwise Left with
     *         the error message/help
     */
    public static Either<String, CliOptions> create(String[] args) {
        // ...
    }
}

public static void main(String[] args) {
    val cliOptions = CliOptions.create(args);

    if (cliOptions.isRight()) {
        run(cliOptions.get());
        System.exit(0);
    }
    else {
        System.err.println(cliOptions.getLeft());
        System.exit(1);
    }
}

public interface Subscriber<T> {
    /**
     * Invoked after calling Publisher#subscribe(Subscriber).
     *
     * No data will start flowing until Subscription#request(long) is invoked.
     *
     * It is the responsibility of this Subscriber instance to call Subscription#request(long) whenever more data is wanted.
     *
     * The Publisher will send notifications only in response to Subscription#request(long).
     *
     * @param s Subscription that allows requesting data via Subscription#request(long)
     */
    public void onSubscribe(Subscription s);

    /**
     * Data notification sent by the Publisher in response to requests to Subscription#request(long).
     *
     * @param t the element signaled
     */
    public void onNext(T t);

    /**
     * Failed terminal state.
     *
     * No further events will be sent even if Subscription#request(long) is invoked again.
     *
     * @param t the throwable signaled
     */
    public void onError(Throwable t);

    /**
     * Successful terminal state.
     *
     * No further events will be sent even if Subscription#request(long) is invoked again.
     */
    public void onComplete();
}

public interface Publisher<T> {
    /**
     * Request Publisher to start streaming data.
     *
     * This is a "factory method" and can be called multiple times,
     * each time starting a new Subscription.
     *
     * Each Subscription will work for only a single Subscriber.
     *
     * A Subscriber should only subscribe once to a single Publisher.
     *
     * If the Publisher rejects the subscription attempt or otherwise fails it
     * will signal the error via Subscriber#onError.
     *
     * @param s the Subscriber that will consume signals from this Publisher
     */
    public void subscribe(Subscriber<? super T> s);
}

public interface Subscription {
    /**
     * No events will be sent by a Publisher until demand is signaled via this method.
     *
     * It can be called however often and whenever needed - but the outstanding
     * cumulative demand must never exceed Long.MAX_VALUE. An outstanding
     * cumulative demand of Long.MAX_VALUE may be treated by the Publisher as
     * "effectively unbounded".
     *
     * Whatever has been requested can be sent by the Publisher so only signal
     * demand for what can be safely handled.
     *
     * A Publisher can send less than is requested if the stream ends but then
     * must emit either Subscriber#onError(Throwable) or Subscriber#onComplete().
     *
     * @param n the strictly positive number of elements to requests to the upstream Publisher
     */
    public void request(long n);

    /**
     * Request the Publisher to stop sending data and clean up resources.
     *
     * Data may still be sent to meet previously signalled demand after calling cancel as
     * this request is asynchronous.
     */
    public void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

/**
 * Start the jobs on the Grabbit clients, returning the hosts
 * and the jobs ids.
 */
public Publisher<HostAndJobIds> startJobs() {
    return Flux.fromIterable(hosts).
        map(this::startJobsForHost);
}

private HostAndJobIds startJobsForHost(HostInfo hostInfo) {
    Publisher<Long> jobIds = startJobsOnHost(hostInfo.baseUri(),
                                             hostInfo.credentials());
    return new HostAndJobIds(hostInfo.baseUri(), jobIds);
}

/**
 * Print the started jobs to the PrintStream, waiting up to 30
 * minutes for the jobs to finish starting across all the hosts.
 */
static void printStartedJobs(Publisher<HostAndJobIds> startedJobs,
                             PrintStream out) {
    Flux.from(startedJobs).
        flatMap(GrabbitCli::hostAndJobIdsToStrings).
        doOnNext(out::println).
        then().block(Duration.ofMinutes(30));
}

/**
 * Start a process on {@link #executorService} that connects to the Grabbit client at `baseUri` and
 * publishes the job ids.
 */
private Publisher<Long> startJobsOnHost(final URI baseUri, UsernameAndPassword credentials) {
    final Processor<Long, Long> processor = WorkQueueProcessor.share(executorService);
    executorService.execute(() -> startJobsOnHostWithSubscriber(processor, baseUri, credentials));
    return processor;
}

private void startJobsOnHostWithSubscriber(Subscriber<Long> jobIdSubscriber, URI baseUri,
                                           UsernameAndPassword credentials) {
    // ...
}

private void parseStartJobsOutput(String startJobsOutput, Subscriber<Long> jobIdSub) {
    // the output from starting a job looks like "[123,125]"
    val matcher = JOB_IDS_PATTERN.matcher(startJobsOutput);
    if (matcher.matches()) {
        val jobIdsStr = (@NonNull String)matcher.group("jobIds");
        Arrays.stream(jobIdsStr.split(",")).
            map(String::trim).
            map(Long::valueOf).
            forEach(jobId -> jobIdSub.onNext(jobId));
        jobIdSub.onComplete();
    }
    else {
        jobIdSub.onError(new IllegalStateException("Could not parse job ids from: " + startJobsOutput));
    }
}

/**
 * Entry point for the command line arguments.
 *
 * [plantuml]
 * ....
 * start
 * :CliOptions.create(argv);
 * if (Either(error message, CliOptions)) then (error message)
 *     :print error message to STDERR;
 *     end
 * else (CliOptions)
 *     :run(cliOptions);
 *     stop
 * endif
 * ....
 *
 * @param args CLI arguments
 */
public static void main(String[] args) {
    // ...
}

/**
 * Runs the appropriate process(es) for the command line options.
 *
 * Output from the processes are sent to STDOUT.
 *
 * [plantuml]
 * ....
 * :CliOptions.create(argv);
 * if (start) then (true)
 *     if (monitor) then (true)
 *         :startWithMonitor(..);
 *     else (false)
 *         :startWithNoMonitor(..);
 *     endif
 * else (false)
 *     :monitorJobs(..);
 * endif
 * ....
 *
 * @see #startWithMonitor(String, String, String, PrintStream)
 * @see #startWithNoMonitor(String, String, String, PrintStream)
 * @see #monitorJobs(String, String, String, PrintStream)
 */
public static void run(CliOptions options) throws IOException {
    // ...
}

public static PollingJobMonitor.Builder builder() {
    return new PollingJobMonitor.Builder();
}
public static class Builder {
    // ...
    public PollingJobMonitor build() throws IllegalStateException {
        // ...
    }
    public Builder jobStatusCache(JobStatusCache jobStatusCache) {
        this.jobStatusCache = jobStatusCache; return this;
    }
    public Builder executor(ExecutorService executorService) {
        this.executorService = executorService; return this;
    }
    public Builder environment(Environment environment) {
        this.environment = environment; return this;
    }
    public Builder poller(JobStatusPoller jobStatusPoller) {
        this.jobStatusPoller = jobStatusPoller; return this;
    }
    public Builder sleep(long sleepMs) {
        this.sleepMs = sleepMs; return this;
    }
}

• You only need to set ExecutorService or Environment, but at least one of them must be set.
• JobStatusCache is required, but JobStatusPoller and sleep are optional.
• If build() is called without all the requirements being met, it throws an IllegalStateException.

public static B.WithJobStatusCache builder() {
    return new B.Builder();
}
public interface B { // namespacing
    class Builder implements Build, ExecutorOrEnvironment, WithJobStatusCache {
        //... virtually identical to before
    }
    interface Build {
        PollingJobMonitor build();
        Build poller(JobStatusPoller jobStatusPoller);
        Build sleep(long sleepMs);
    }
    interface WithExecutor {
        Build executor(ExecutorService executorService);
    }
    interface WithEnvironment {
        Build environment(Environment environment);
    }
    interface ExecutorOrEnvironment extends WithExecutor, WithEnvironment {
    }
    interface WithJobStatusCache {
        ExecutorOrEnvironment jobStatusCache(JobStatusCache jobStatusCache);
    }
}

• The type system is acting as a compile-time state machine, ensuring that it’s impossible to call build() in a bad state.