/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.distributedlog.util;

import com.google.common.base.Objects;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nullable;

import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.meta.ZkVersion;
import org.apache.bookkeeper.versioning.Versioned;
import org.apache.distributedlog.DistributedLogConstants;
import org.apache.distributedlog.ZooKeeperClient;
import org.apache.distributedlog.common.concurrent.FutureUtils;
import org.apache.distributedlog.common.functions.VoidFunctions;
import org.apache.distributedlog.exceptions.BKTransmitException;
import org.apache.distributedlog.exceptions.DLInterruptedException;
import org.apache.distributedlog.exceptions.UnexpectedException;
import org.apache.distributedlog.exceptions.ZKException;
import org.apache.distributedlog.io.AsyncAbortable;
import org.apache.distributedlog.io.AsyncCloseable;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

/**
 * Basic Utilities.
 */
@Slf4j
public class Utils {

    /**
     * Current time from some arbitrary time base in the past, counting in
     * nanoseconds, and not affected by settimeofday or similar system clock
     * changes. This is appropriate to use when computing how much longer to
     * wait for an interval to expire.
     *
     * @return current time in nanoseconds.
     */
    public static long nowInNanos() {
        return System.nanoTime();
    }

    /**
     * Current time from some fixed base time - so useful for cross machine comparison.
     *
     * @return current time in milliseconds.
     */
    public static long nowInMillis() {
        return System.currentTimeMillis();
    }

    /**
     * Milliseconds elapsed since the time specified, the input is nanoTime
     * the only conversion happens when computing the elapsed time.
     *
     * @param startMsecTime the start of the interval that we are measuring
     * @return elapsed time in milliseconds.
     */
    public static long elapsedMSec(long startMsecTime) {
        return (System.currentTimeMillis() - startMsecTime);
    }

    public static boolean randomPercent(double percent) {
        return (Math.random() * 100.0) <= percent;
    }

    /**
     * Synchronously create zookeeper path recursively and optimistically.
     *
     * @see #zkAsyncCreateFullPathOptimistic(ZooKeeperClient, String, byte[], List, CreateMode)
     * @param zkc Zookeeper client
     * @param path Zookeeper full path
     * @param data Zookeeper data
     * @param acl Acl of the zk path
     * @param createMode Create mode of zk path
     * @throws ZooKeeperClient.ZooKeeperConnectionException
     * @throws KeeperException
     * @throws InterruptedException
     */
    public static void zkCreateFullPathOptimistic(
        ZooKeeperClient zkc,
        String path,
        byte[] data,
        final List<ACL> acl,
        final CreateMode createMode) throws IOException, KeeperException {
        try {
            FutureUtils.result(zkAsyncCreateFullPathOptimistic(zkc, path, data, acl, createMode));
        } catch (ZooKeeperClient.ZooKeeperConnectionException zkce) {
            throw zkce;
        } catch (KeeperException ke) {
            throw ke;
        } catch (InterruptedException ie) {
            throw new DLInterruptedException("Interrupted on create zookeeper path " + path, ie);
        } catch (RuntimeException rte) {
            throw rte;
        } catch (Exception exc) {
            throw new RuntimeException("Unexpected Exception", exc);
        }
    }

    /**
     * Asynchronously create zookeeper path recursively and optimistically.
     *
     * @param zkc Zookeeper client
     * @param pathToCreate  Zookeeper full path
     * @param parentPathShouldNotCreate The recursive creation should stop if this path doesn't exist
     * @param data Zookeeper data
     * @param acl Acl of the zk path
     * @param createMode Create mode of zk path
     * @param callback Callback
     * @param ctx Context object
     */
    public static void zkAsyncCreateFullPathOptimisticRecursive(
        final ZooKeeperClient zkc,
        final String pathToCreate,
        final Optional<String> parentPathShouldNotCreate,
        final byte[] data,
        final List<ACL> acl,
        final CreateMode createMode,
        final AsyncCallback.StringCallback callback,
        final Object ctx) {
        try {
            zkc.get().create(pathToCreate, data, acl, createMode, new AsyncCallback.StringCallback() {
                @Override
                public void processResult(int rc, String path, Object ctx, String name) {

                    if (rc != KeeperException.Code.NONODE.intValue()) {
                        callback.processResult(rc, path, ctx, name);
                        return;
                    }

                    // Since we got a nonode, it means that my parents may not exist
                    // ephemeral nodes can't have children so Create mode is always
                    // persistent parents
                    int lastSlash = pathToCreate.lastIndexOf('/');
                    if (lastSlash <= 0) {
                        callback.processResult(rc, path, ctx, name);
                        return;
                    }
                    String parent = pathToCreate.substring(0, lastSlash);
                    if (parentPathShouldNotCreate.isPresent()
                            && Objects.equal(parentPathShouldNotCreate.get(), parent)) {
                        // we should stop here
                        callback.processResult(rc, path, ctx, name);
                        return;
                    }
                    zkAsyncCreateFullPathOptimisticRecursive(zkc, parent, parentPathShouldNotCreate, new byte[0], acl,
                            CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
                                @Override
                                public void processResult(int rc, String path, Object ctx, String name) {
                                    if (rc == KeeperException.Code.OK.intValue()
                                            || rc == KeeperException.Code.NODEEXISTS.intValue()) {
                                        // succeeded in creating the parent, now create the original path
                                        zkAsyncCreateFullPathOptimisticRecursive(zkc, pathToCreate,
                                                parentPathShouldNotCreate, data, acl, createMode, callback, ctx);
                                    } else {
                                        callback.processResult(rc, path, ctx, name);
                                    }
                                }
                            }, ctx);
                }
            }, ctx);
        } catch (ZooKeeperClient.ZooKeeperConnectionException zkce) {
            callback.processResult(DistributedLogConstants.ZK_CONNECTION_EXCEPTION_RESULT_CODE,
                    zkce.getMessage(), ctx, pathToCreate);
        } catch (InterruptedException ie) {
            callback.processResult(DistributedLogConstants.DL_INTERRUPTED_EXCEPTION_RESULT_CODE,
                    ie.getMessage(), ctx, pathToCreate);
        }
    }

    /**
     * Asynchronously create zookeeper path recursively and optimistically.
     *
     * @param zkc Zookeeper client
     * @param pathToCreate  Zookeeper full path
     * @param data Zookeeper data
     * @param acl Acl of the zk path
     * @param createMode Create mode of zk path
     */
    public static CompletableFuture<Void> zkAsyncCreateFullPathOptimistic(
        final ZooKeeperClient zkc,
        final String pathToCreate,
        final byte[] data,
        final List<ACL> acl,
        final CreateMode createMode) {
        Optional<String> parentPathShouldNotCreate = Optional.absent();
        return zkAsyncCreateFullPathOptimistic(
                zkc,
                pathToCreate,
                parentPathShouldNotCreate,
                data,
                acl,
                createMode);
    }

    /**
     * Asynchronously create zookeeper path recursively and optimistically.
     *
     * @param zkc Zookeeper client
     * @param pathToCreate  Zookeeper full path
     * @param parentPathShouldNotCreate zookeeper parent path should not be created
     * @param data Zookeeper data
     * @param acl Acl of the zk path
     * @param createMode Create mode of zk path
     */
    public static CompletableFuture<Void> zkAsyncCreateFullPathOptimistic(
        final ZooKeeperClient zkc,
        final String pathToCreate,
        final Optional<String> parentPathShouldNotCreate,
        final byte[] data,
        final List<ACL> acl,
        final CreateMode createMode) {
        final CompletableFuture<Void> result = new CompletableFuture<Void>();

        zkAsyncCreateFullPathOptimisticRecursive(zkc, pathToCreate, parentPathShouldNotCreate,
                data, acl, createMode, new AsyncCallback.StringCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, String name) {
                handleKeeperExceptionCode(rc, path, result);
            }
        }, result);

        return result;
    }

    /**
     * Asynchronously create zookeeper path recursively and optimistically.
     *
     * @param zkc Zookeeper client
     * @param pathToCreate  Zookeeper full path
     * @param data Zookeeper data
     * @param acl Acl of the zk path
     * @param createMode Create mode of zk path
     */
    public static CompletableFuture<Void> zkAsyncCreateFullPathOptimisticAndSetData(
        final ZooKeeperClient zkc,
        final String pathToCreate,
        final byte[] data,
        final List<ACL> acl,
        final CreateMode createMode) {
        final CompletableFuture<Void> result = new CompletableFuture<Void>();

        try {
            zkc.get().setData(pathToCreate, data, -1, new AsyncCallback.StatCallback() {
                @Override
                public void processResult(int rc, String path, Object ctx, Stat stat) {
                    if (rc != KeeperException.Code.NONODE.intValue()) {
                        handleKeeperExceptionCode(rc, path, result);
                        return;
                    }

                    Optional<String> parentPathShouldNotCreate = Optional.absent();
                    zkAsyncCreateFullPathOptimisticRecursive(zkc, pathToCreate, parentPathShouldNotCreate,
                            data, acl, createMode, new AsyncCallback.StringCallback() {
                        @Override
                        public void processResult(int rc, String path, Object ctx, String name) {
                            handleKeeperExceptionCode(rc, path, result);
                        }
                    }, result);
                }
            }, result);
        } catch (Exception exc) {
            result.completeExceptionally(exc);
        }

        return result;
    }

    private static void handleKeeperExceptionCode(int rc, String pathOrMessage, CompletableFuture<Void> result) {
        if (KeeperException.Code.OK.intValue() == rc) {
            result.complete(null);
        } else if (DistributedLogConstants.ZK_CONNECTION_EXCEPTION_RESULT_CODE == rc) {
            result.completeExceptionally(new ZooKeeperClient.ZooKeeperConnectionException(pathOrMessage));
        } else if (DistributedLogConstants.DL_INTERRUPTED_EXCEPTION_RESULT_CODE == rc) {
            result.completeExceptionally(new DLInterruptedException(pathOrMessage));
        } else {
            result.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc), pathOrMessage));
        }
    }

    public static CompletableFuture<Versioned<byte[]>> zkGetData(ZooKeeperClient zkc, String path, boolean watch) {
        ZooKeeper zk;
        try {
            zk = zkc.get();
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            return FutureUtils.exception(zkException(e, path));
        } catch (InterruptedException e) {
            return FutureUtils.exception(zkException(e, path));
        }
        return zkGetData(zk, path, watch);
    }

    /**
     * Retrieve data from zookeeper <code>path</code>.
     *
     * @param path
     *          zookeeper path to retrieve data
     * @param watch
     *          whether to watch the path
     * @return future representing the versioned value. null version or null value means path doesn't exist.
     */
    public static CompletableFuture<Versioned<byte[]>> zkGetData(ZooKeeper zk, String path, boolean watch) {
        final CompletableFuture<Versioned<byte[]>> promise = new CompletableFuture<Versioned<byte[]>>();
        zk.getData(path, watch, new AsyncCallback.DataCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    if (null == stat) {
                        promise.complete(new Versioned<byte[]>(null, null));
                    } else {
                        promise.complete(new Versioned<byte[]>(data, new ZkVersion(stat.getVersion())));
                    }
                } else if (KeeperException.Code.NONODE.intValue() == rc) {
                    promise.complete(new Versioned<byte[]>(null, null));
                } else {
                    promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc)));
                }
            }
        }, null);
        return promise;
    }

    public static CompletableFuture<ZkVersion> zkSetData(ZooKeeperClient zkc,
                                                         String path, byte[] data, ZkVersion version) {
        ZooKeeper zk;
        try {
            zk = zkc.get();
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            return FutureUtils.exception(zkException(e, path));
        } catch (InterruptedException e) {
            return FutureUtils.exception(zkException(e, path));
        }
        return zkSetData(zk, path, data, version);
    }

    /**
     * Set <code>data</code> to zookeeper <code>path</code>.
     *
     * @param zk
     *          zookeeper client
     * @param path
     *          path to set data
     * @param data
     *          data to set
     * @param version
     *          version used to set data
     * @return future representing the version after this operation.
     */
    public static CompletableFuture<ZkVersion> zkSetData(ZooKeeper zk, String path, byte[] data, ZkVersion version) {
        final CompletableFuture<ZkVersion> promise = new CompletableFuture<ZkVersion>();
        zk.setData(path, data, version.getZnodeVersion(), new AsyncCallback.StatCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx, Stat stat) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    promise.complete(new ZkVersion(stat.getVersion()));
                    return;
                }
                promise.completeExceptionally(
                        KeeperException.create(KeeperException.Code.get(rc)));
                return;
            }
        }, null);
        return promise;
    }

    public static CompletableFuture<Void> zkDelete(ZooKeeperClient zkc, String path, ZkVersion version) {
        ZooKeeper zk;
        try {
            zk = zkc.get();
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            return FutureUtils.exception(zkException(e, path));
        } catch (InterruptedException e) {
            return FutureUtils.exception(zkException(e, path));
        }
        return zkDelete(zk, path, version);
    }

    /**
     * Delete the given <i>path</i> from zookeeper.
     *
     * @param zk
     *          zookeeper client
     * @param path
     *          path to delete
     * @param version
     *          version used to set data
     * @return future representing the version after this operation.
     */
    public static CompletableFuture<Void> zkDelete(ZooKeeper zk, String path, ZkVersion version) {
        final CompletableFuture<Void> promise = new CompletableFuture<Void>();
        zk.delete(path, version.getZnodeVersion(), new AsyncCallback.VoidCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    promise.complete(null);
                    return;
                }
                promise.completeExceptionally(
                        KeeperException.create(KeeperException.Code.get(rc)));
                return;
            }
        }, null);
        return promise;
    }

    /**
     * Delete the given <i>path</i> from zookeeper.
     *
     * @param zkc
     *          zookeeper client
     * @param path
     *          path to delete
     * @param version
     *          version used to set data
     * @return future representing if the delete is successful. Return true if the node is deleted,
     * false if the node doesn't exist, otherwise future will throw exception
     *
     */
    public static CompletableFuture<Boolean> zkDeleteIfNotExist(ZooKeeperClient zkc, String path, ZkVersion version) {
        ZooKeeper zk;
        try {
            zk = zkc.get();
        } catch (ZooKeeperClient.ZooKeeperConnectionException e) {
            return FutureUtils.exception(zkException(e, path));
        } catch (InterruptedException e) {
            return FutureUtils.exception(zkException(e, path));
        }
        final CompletableFuture<Boolean> promise = new CompletableFuture<Boolean>();
        zk.delete(path, version.getZnodeVersion(), new AsyncCallback.VoidCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx) {
                if (KeeperException.Code.OK.intValue() == rc) {
                    promise.complete(true);
                } else if (KeeperException.Code.NONODE.intValue() == rc) {
                    promise.complete(false);
                } else {
                    promise.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc)));
                }
            }
        }, null);
        return promise;
    }

    public static CompletableFuture<Void> asyncClose(@Nullable AsyncCloseable closeable,
                                          boolean swallowIOException) {
        if (null == closeable) {
            return FutureUtils.Void();
        } else if (swallowIOException) {
            return FutureUtils.ignore(closeable.asyncClose());
        } else {
            return closeable.asyncClose();
        }
    }

    /**
     * Sync zookeeper client on given <i>path</i>.
     *
     * @param zkc
     *          zookeeper client
     * @param path
     *          path to sync
     * @return zookeeper client after sync
     * @throws IOException
     */
    public static ZooKeeper sync(ZooKeeperClient zkc, String path) throws IOException {
        ZooKeeper zk;
        try {
            zk = zkc.get();
        } catch (InterruptedException e) {
            throw new DLInterruptedException("Interrupted on checking if log " + path + " exists", e);
        }
        final CountDownLatch syncLatch = new CountDownLatch(1);
        final AtomicInteger syncResult = new AtomicInteger(0);
        zk.sync(path, new AsyncCallback.VoidCallback() {
            @Override
            public void processResult(int rc, String path, Object ctx) {
                syncResult.set(rc);
                syncLatch.countDown();
            }
        }, null);
        try {
            syncLatch.await();
        } catch (InterruptedException e) {
            throw new DLInterruptedException("Interrupted on syncing zookeeper connection", e);
        }
        if (KeeperException.Code.OK.intValue() != syncResult.get()) {
            throw new ZKException("Error syncing zookeeper connection ",
                    KeeperException.Code.get(syncResult.get()));
        }
        return zk;
    }

    /**
     * Close a closeable.
     *
     * @param closeable
     *          closeable to close
     */
    public static void close(@Nullable Closeable closeable) {
        if (null == closeable) {
            return;
        }
        try {
            Closeables.close(closeable, true);
        } catch (IOException e) {
            // no-op. the exception is swallowed.
        }
    }

    /**
     * Close an async closeable.
     *
     * @param closeable
     *          closeable to close
     */
    public static void close(@Nullable AsyncCloseable closeable)
            throws IOException {
        if (null == closeable) {
            return;
        }
        Utils.ioResult(closeable.asyncClose());
    }

    /**
     * Close an async closeable.
     *
     * @param closeable
     *          closeable to close
     */
    public static void closeQuietly(@Nullable AsyncCloseable closeable) {
        if (null == closeable) {
            return;
        }
        try {
            Utils.ioResult(closeable.asyncClose());
        } catch (IOException e) {
            // no-op. the exception is swallowed.
        }
    }

    /**
     * Close the closeables in sequence.
     *
     * @param closeables
     *          closeables to close
     * @return future represents the close future
     */
    public static CompletableFuture<Void> closeSequence(ExecutorService executorService,
                                             AsyncCloseable... closeables) {
        return closeSequence(executorService, false, closeables);
    }

    /**
     * Close the closeables in sequence and ignore errors during closing.
     *
     * @param executorService executor to execute closeable
     * @param ignoreCloseError whether to ignore errors during closing
     * @param closeables list of closeables
     * @return future represents the close future.
     */
    public static CompletableFuture<Void> closeSequence(ExecutorService executorService,
                                             boolean ignoreCloseError,
                                             AsyncCloseable... closeables) {
        List<AsyncCloseable> closeableList = Lists.newArrayListWithExpectedSize(closeables.length);
        for (AsyncCloseable closeable : closeables) {
            if (null == closeable) {
                closeableList.add(AsyncCloseable.NULL);
            } else {
                closeableList.add(closeable);
            }
        }
        return FutureUtils.processList(
                closeableList,
                ignoreCloseError ? AsyncCloseable.CLOSE_FUNC_IGNORE_ERRORS : AsyncCloseable.CLOSE_FUNC,
                executorService
        ).thenApply(VoidFunctions.LIST_TO_VOID_FUNC);
    }

    /**
     * Gets the parent of a path.
     *
     * @param path
     *            path to get the parent of
     * @return parent of the path or null if no parent exists.
     */
    public static String getParent(final String path) {
        if (path == null) {
            return null;
        }
        if (path.length() < 2) {
            return null;
        }
        int firstIndex = path.indexOf("/");
        if (firstIndex == -1) {
            return null;
        }
        int lastIndex = path.lastIndexOf("/");
        if (lastIndex == path.length() - 1) {
            lastIndex = path.substring(0, path.length() - 1).lastIndexOf("/");
        }
        if (lastIndex == -1) {
            return null;
        }
        if (lastIndex == 0) {
            return "/";
        }
        return path.substring(0, lastIndex);
    }

    /**
     * Convert the <i>throwable</i> to zookeeper related exceptions.
     *
     * @param throwable cause
     * @param path zookeeper path
     * @return zookeeper related exceptions
     */
    public static Throwable zkException(Throwable throwable, String path) {
        if (throwable instanceof KeeperException) {
            return new ZKException("Encountered zookeeper exception on " + path, (KeeperException) throwable);
        } else if (throwable instanceof ZooKeeperClient.ZooKeeperConnectionException) {
            return new ZKException("Encountered zookeeper connection loss on " + path,
                    KeeperException.Code.CONNECTIONLOSS);
        } else if (throwable instanceof InterruptedException) {
            return new DLInterruptedException("Interrupted on operating " + path, throwable);
        } else {
            return new UnexpectedException("Encountered unexpected exception on operatiing " + path, throwable);
        }
    }

    /**
     * Create transmit exception from transmit result.
     *
     * @param transmitResult
     *          transmit result (basically bk exception code)
     * @return transmit exception
     */
    public static BKTransmitException transmitException(int transmitResult) {
        return new BKTransmitException("Failed to write to bookkeeper; Error is ("
            + transmitResult + ") "
            + BKException.getMessage(transmitResult), transmitResult);
    }

    /**
     * A specific version of {@link FutureUtils#result(CompletableFuture)} to handle known exception issues.
     */
    public static <T> T ioResult(CompletableFuture<T> result) throws IOException {
        return FutureUtils.result(
            result,
            (cause) -> {
                if (cause instanceof IOException) {
                    return (IOException) cause;
                } else if (cause instanceof KeeperException) {
                    return new ZKException("Encountered zookeeper exception on waiting result",
                        (KeeperException) cause);
                } else if (cause instanceof BKException) {
                    return new BKTransmitException("Encountered bookkeeper exception on waiting result",
                        ((BKException) cause).getCode());
                } else if (cause instanceof InterruptedException) {
                    return new DLInterruptedException("Interrupted on waiting result", cause);
                } else {
                    return new IOException("Encountered exception on waiting result", cause);
                }
            });
    }

    /**
     * A specific version of {@link FutureUtils#result(CompletableFuture, long, TimeUnit)}
     * to handle known exception issues.
     */
    public static <T> T ioResult(CompletableFuture<T> result, long timeout, TimeUnit timeUnit)
            throws IOException, TimeoutException {
        return FutureUtils.result(
            result,
            (cause) -> {
                if (cause instanceof IOException) {
                    return (IOException) cause;
                } else if (cause instanceof KeeperException) {
                    return new ZKException("Encountered zookeeper exception on waiting result",
                        (KeeperException) cause);
                } else if (cause instanceof BKException) {
                    return new BKTransmitException("Encountered bookkeeper exception on waiting result",
                        ((BKException) cause).getCode());
                } else if (cause instanceof InterruptedException) {
                    return new DLInterruptedException("Interrupted on waiting result", cause);
                } else {
                    return new IOException("Encountered exception on waiting result", cause);
                }
            },
            timeout,
            timeUnit);
    }

    /**
     * Abort async <i>abortable</i>.
     *
     * @param abortable the {@code AsyncAbortable} object to be aborted, or null, in which case this method
     *                  does nothing.
     * @param swallowIOException if true, don't propagate IO exceptions thrown by the {@code abort} methods
     * @throws IOException if {@code swallowIOException} is false and {@code abort} throws an {@code IOException}
     */
    public static void abort(@Nullable AsyncAbortable abortable,
                             boolean swallowIOException)
            throws IOException {
        if (null == abortable) {
            return;
        }
        try {
            ioResult(abortable.asyncAbort());
        } catch (Exception ioe) {
            if (swallowIOException) {
                log.warn("IOException thrown while aborting Abortable {} : ", abortable, ioe);
            } else {
                throw ioe;
            }
        }
    }

}