package com.jscape.inet.sftp;

import com.jscape.inet.sftp.protocol.marshaling.MessageCodec;
import com.jscape.inet.sftp.protocol.messages.Message;
import com.jscape.inet.sftp.protocol.v3.marshaling.Messages3;
import com.jscape.inet.sftp.protocol.v3.marshaling.SshFxpExtendedReplyCheckFileNameCodec;
import com.jscape.inet.sftp.protocol.v3.messages.*;
import com.jscape.inet.ssh.protocol.v2.connection.SessionChannelConnection;
import com.jscape.inet.ssh.protocol.v2.connection.SessionConnection;
import com.jscape.inet.ssh.protocol.v2.connection.SubsystemType;
import com.jscape.util.Time;
import com.jscape.util.file.PosixFilePermissions;
import com.jscape.util.file.UnixFileType;
import com.jscape.util.io.ByteArrayInputStream;
import com.jscape.util.net.connection.Connector;
import com.jscape.util.net.connection.SyncRawConnection;
import com.jscape.util.service.AbstractService;

import java.io.*;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import static com.jscape.inet.sftp.FileService.OperationException.wrap;
import static com.jscape.util.CollectionTools.list;
import static com.jscape.util.Parameters.assertGreater;
import static com.jscape.util.Parameters.assertNotNull;

/**
 * @author Alex Usun
 */
public class SftpFileService3
        extends AbstractService<Void>
        implements FileService {

    public static final int DEFAULT_PIPELINE_WINDOW_SIZE = 8;

    private static final int VERSION = 3;

    private final Connector<SessionConnection> connector;
    private final int transferBufferSizeBytes;
    private final boolean subsystemResponseRequired;
    private final int pipelineWindowSize;
    private final MessageCodec messageCodec;
    private final AtomicInteger messageIds;
    private final Logger logger;
    private SessionConnection sessionConnection;
    private SessionChannelConnection channelConnection;
    private InputStream in;
    private OutputStream out;
    private volatile boolean transferCancelled;

    public SftpFileService3(Connector<SessionConnection> connector, int transferBufferSizeBytes, boolean subsystemResponseRequired, int pipelineWindowSize, Logger logger) {
        assertNotNull(connector);
        this.connector = connector;

        assertGreater(transferBufferSizeBytes, 0, "Bad buffer size value.");
        this.transferBufferSizeBytes = transferBufferSizeBytes;

        this.subsystemResponseRequired = subsystemResponseRequired;

        assertGreater(pipelineWindowSize, 0, "Bad pipeline window size.");
        this.pipelineWindowSize = pipelineWindowSize;

        this.messageCodec = Messages3.init(new MessageCodec());
        this.messageIds = new AtomicInteger(0);

        assertNotNull(logger);
        this.logger = logger;
    }

    @Override
    public boolean available() {
        if (this.sessionConnection == null || this.sessionConnection.closed()) {
            return false;
        }

        try {
            exchange(new SshFxpRealPath(nextMessageId(), "."));
            return true;
        } catch (ServerException e) {
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public PathInfo infoAbout(String path, boolean followLink)
            throws OperationException {
        try {
            SshFxpName response = exchange(followLink ? new SshFxpReadLink(nextMessageId(), path) : new SshFxpRealPath(nextMessageId(), path));
            return infoFor(response.entries[0]);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    public FileAttributes attributesOf(String path, boolean followLink)
            throws OperationException {
        try {
            SshFxpAttrs response = exchange(followLink ? new SshFxpStat(nextMessageId(), path) : new SshFxpLstat(nextMessageId(), path));
            return response.attributes;
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    public void attributes(String path, FileAttributes attributes)
            throws OperationException {
        try {
            exchange(new SshFxpSetStat(nextMessageId(), path, attributes));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public boolean exists(String path)
            throws OperationException {
        try {
            exchange(new SshFxpStat(nextMessageId(), path));
            return true;
        } catch (ServerException e) {
            if (e.code == StatusCode.SSH_FX_NO_SUCH_FILE.code) {
                return false;
            }

            throw wrap(e);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public UnixFileType typeOf(String path, boolean followLink)
            throws OperationException {
        FileAttributes attributes = attributesOf(path, followLink);
        return attributes.type();
    }

    @Override
    public long sizeOf(String path, boolean followLink)
            throws OperationException {
        FileAttributes attributes = attributesOf(path, followLink);
        return attributes.size();
    }

    @Override
    public PosixFilePermissions permissionsOf(String path, boolean followLink)
            throws OperationException {
        FileAttributes attributes = attributesOf(path, followLink);
        return attributes.permissions;
    }

    @Override
    public void permissions(String path, PosixFilePermissions permissions)
            throws OperationException {
        attributes(path, FileAttributes.attributesFor(permissions));
    }

    @Override
    public long createTimeOf(String path, boolean followLink)
            throws OperationException {
        return 0L;
    }

    @Override
    public void createTime(String path, long timeMillis)
            throws OperationException {
        throw new OperationException("Operation not implemented.");
    }

    @Override
    public long accessTimeOf(String path, boolean followLink)
            throws OperationException {
        FileAttributes attributes = attributesOf(path, followLink);
        return attributes.accessTime();
    }

    @Override
    public void accessTime(String path, long timeMillis)
            throws OperationException {
        attributes(path, new FileAttributes(null, null, null, null, null, (int) (timeMillis / 1000), null, Collections.<String, String>emptyMap()));
    }

    @Override
    public long modificationTimeOf(String path, boolean followLink)
            throws OperationException {
        FileAttributes attributes = attributesOf(path, followLink);
        return attributes.modificationTime();
    }

    @Override
    public void modificationTime(String path, long timeMillis)
            throws OperationException {
        attributes(path, FileAttributes.attributesFor(timeMillis));
    }

    @Override
    public HashInfo hashOf(String path, long offset, long length, int blockSize, String... algorithms)
            throws OperationException {
        try {
            List<String> algorithmList = list(algorithms);
            SshFxpExtendedReplyRaw response = exchange(new SshFxpExtendedCheckFileName(nextMessageId(), path, algorithmList, offset, length, blockSize));
            // todo: use more elegant method
            SshFxpExtendedReplyCheckFileName checkFileNameResponse = (SshFxpExtendedReplyCheckFileName) new SshFxpExtendedReplyCheckFileNameCodec().read(new ByteArrayInputStream(response.data), response.id);

            return new HashInfo(checkFileNameResponse.hashAlgorithmUsed, checkFileNameResponse.data);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void createSymbolicLink(String linkPath, String targetPath)
            throws OperationException {
        try {
            exchange(new SshFxpSymLink(nextMessageId(), linkPath, targetPath));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void createDirectory(String path, PosixFilePermissions permissions)
            throws OperationException {
        try {
            exchange(new SshFxpMkdir(nextMessageId(), path, new FileAttributes(null, UnixFileType.DIRECTORY, permissions, null)));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void removeDirectory(String path)
            throws OperationException {
        try {
            exchange(new SshFxpRmdir(nextMessageId(), path));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void removeFile(String path)
            throws OperationException {
        try {
            exchange(new SshFxpRemove(nextMessageId(), path));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void rename(String sourcePath, String destinationPath)
            throws OperationException {
        try {
            exchange(new SshFxpRename(nextMessageId(), sourcePath, destinationPath));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public List<PathInfo> listDirectory(String path)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(new SshFxpOpendir(nextMessageId(), path));

        try {
            List<PathInfo> entries = new LinkedList<>();
            
            try {
                Message response;
                while (!this.transferCancelled && (response = exchange(new SshFxpReaddir(nextMessageId(), handle))) instanceof SshFxpName) {
                    SshFxpName nameResponse = (SshFxpName) response;
                    for (NameEntry entry : nameResponse.entries) {
                        if (acceptable(entry)) {
                            PathInfo info = infoFor(entry);
                            entries.add(info);
                        }
                    }
                }
            } catch (Exception e) {
                closeHandleSafe(handle);
                throw e;
            }
    
            closeHandle(handle);

            return entries;
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void readFile(String sourcePath, long sourceOffset, OutputStream destination, boolean textModeRequired, TransferListener listener)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.reading(nextMessageId(), sourcePath));

        try {
            try {
                Message response;
                long transferredBytes = 0;
                long offset = sourceOffset;
        
                while (!this.transferCancelled && (response = exchange(new SshFxpRead(nextMessageId(), handle, offset, this.transferBufferSizeBytes))) instanceof SshFxpData) {
                    SshFxpData dataResponse = (SshFxpData) response;
                    destination.write(dataResponse.data);
                    offset += dataResponse.data.length;
                    transferredBytes += dataResponse.data.length;
                    listener.onTransferProgress(transferredBytes);
                }
            } catch (Exception e) {
                closeHandleSafe(handle);
                throw e;
            }
    
            closeHandle(handle);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void readFile(String sourcePath, long sourceOffset, OutputStream destination, boolean textModeRequired, TransferListener listener, ExecutorService executorService)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.reading(nextMessageId(), sourcePath));
        ReadOperation operation = new ReadOperation(handle, sourceOffset, destination, listener);

        try {
            executorService.submit(operation);
            operation.execute();
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void appendFile(InputStream source, String destinationPath, boolean textModeRequired, TransferListener listener)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.appending(nextMessageId(), destinationPath));

        try {
            try {
                byte[] buffer = new byte[this.transferBufferSizeBytes];
                int read;
                long transferredBytes = 0;
                long offset = 0;
        
                while (!this.transferCancelled && (read = source.read(buffer)) != -1) {
                    exchange(new SshFxpWrite(nextMessageId(), handle, offset, buffer, 0, read));
                    offset += read;
                    transferredBytes += read;
                    listener.onTransferProgress(transferredBytes);
                }
            } catch (Exception e) {
                closeHandleSafe(handle);
                throw e;
            }
    
            closeHandle(handle);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void appendFile(InputStream source, String destinationPath, boolean textModeRequired, TransferListener listener, ExecutorService executorService)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.appending(nextMessageId(), destinationPath));
        WriteOperation operation = new WriteOperation(source, handle, 0, listener);

        try {
            executorService.submit(operation);
            operation.execute();
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void writeFile(InputStream source, String destinationPath, long destinationOffset, boolean textModeRequired, TransferListener listener)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.writing(nextMessageId(), destinationPath, destinationOffset == 0));

        try {
            try {
                byte[] buffer = new byte[this.transferBufferSizeBytes];
                int read;
                long transferredBytes = 0;
                long offset = destinationOffset;
        
                while (!this.transferCancelled && (read = source.read(buffer)) != -1) {
                    exchange(new SshFxpWrite(nextMessageId(), handle, offset, buffer, 0, read));
                    offset += read;
                    transferredBytes += read;
                    listener.onTransferProgress(transferredBytes);
                }
            } catch (Exception e) {
                closeHandleSafe(handle);
                throw e;
            }
    
            closeHandle(handle);
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void writeFile(InputStream source, String destinationPath, long destinationOffset, boolean textModeRequired, TransferListener listener, ExecutorService executorService)
            throws OperationException {
        this.transferCancelled = false;
        byte[] handle = openHandle(SshFxpOpen.writing(nextMessageId(), destinationPath, destinationOffset == 0));
        WriteOperation operation = new WriteOperation(source, handle, destinationOffset, listener);

        try {
            executorService.submit(operation);
            operation.execute();
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    @Override
    public void cancelTransferOperation() {
        this.transferCancelled = true;
    }

    public SessionConnection sessionConnection() {
        return this.sessionConnection;
    }

    @Override
    protected void actualStart()
            throws Exception {
        this.sessionConnection = this.connector.connect();

        try {
            this.channelConnection = this.sessionConnection.openSessionChannel(true);
            this.channelConnection.subsystem(SubsystemType.SFTP.name, this.subsystemResponseRequired);
            this.in = new BufferedInputStream(new SyncRawConnection.ConnectionInputStream(this.channelConnection));
            this.out = new BufferedOutputStream(new SyncRawConnection.ConnectionOutputStream(this.channelConnection));
            this.messageIds.set(0);
            initProtocol();

            logServiceStarted();
        } catch (Exception e) {
            this.sessionConnection.close();
            throw e;
        }
    }

    @Override
    protected void actualStop() {
        this.channelConnection.close();
        this.sessionConnection.close();
        logServiceStopped();
    }

    private int nextMessageId() {
        return this.messageIds.getAndIncrement();
    }

    private void initProtocol()
            throws Exception {
        SshFxpVersion response = exchange(new SshFxpInit(VERSION, new HashMap<String, String>()));

        if (response.version != 0 && response.version != VERSION) {
            throw new UnsupportedVersionException(VERSION, response.version);
        }
    }

    private byte[] openHandle(Message request)
            throws OperationException {
        try {
            SshFxpHandle response = exchange(request);
            return response.handle;
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    private void closeHandle(byte[] handle)
            throws OperationException {
        try {
            exchange(new SshFxpClose(nextMessageId(), handle));
        } catch (Exception e) {
            throw wrap(e);
        }
    }

    private void closeHandleSafe(byte[] handle) {
        try {
            exchange(new SshFxpClose(nextMessageId(), handle));
        } catch (Exception e) {
            logError(e);
        }
    }

    private void send(Message message)
            throws Exception {
        logOutgoingMessage(message);
        this.messageCodec.write(message, this.out);
    }

    @SuppressWarnings("unchecked")
    private <M extends Message> M receive()
            throws Exception {
        Message message = this.messageCodec.read(this.in);
        logIncomingMessage(message);

        try {
            return (M) message;
        } catch (Exception e) {
            throw new Exception("Unexpected incoming message: " + message);
        }
    }

    @SuppressWarnings("unchecked")
    private <M> M exchange(Message request)
            throws Exception {
        send(request);
        Message response = receive();
        assertNoError(response);

        try {
            return (M) response;
        } catch (Exception e) {
            throw new Exception("Unexpected incoming message: " + response);
        }
    }

    private boolean acceptable(NameEntry entry) {
        return !".".equals(entry.filename) && !"..".equals(entry.filename);
    }

    private void assertNoError(Message response)
            throws Exception {
        if (response instanceof SshFxpStatus) {
            SshFxpStatus statusResponse = (SshFxpStatus) response;

            if (statusResponse.code != StatusCode.SSH_FX_OK.code && statusResponse.code != StatusCode.SSH_FX_EOF.code) {
                throw new ServerException(statusResponse.code, statusResponse.message, statusResponse.languageTag);
            }
        }
    }

    private PathInfo infoFor(NameEntry entry) {
        return new PathInfo(entry.filename,
                            entry.attributes.type(),
                            entry.attributes.size(),
                            entry.attributes.userId != null ? entry.attributes.userId.toString() : null,
                            entry.attributes.groupId != null ? entry.attributes.groupId.toString() : null,
                            entry.attributes.permissions,
                            entry.attributes.accessTime(),
                            entry.attributes.modificationTime());
    }

    private void logServiceStarted() {
        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.log(Level.FINE, String.format("SFTP connection created: [%s <-> %s].",
                                                      this.channelConnection.localAddress(),
                                                      this.channelConnection.remoteAddress()));
        }
    }

    private void logServiceStopped() {
        if (this.logger.isLoggable(Level.FINE)) {
            this.logger.log(Level.FINE, String.format("SFTP connection closed: [%s <-> %s].",
                                                      this.channelConnection.localAddress(),
                                                      this.channelConnection.remoteAddress()));
        }
    }

    private void logOutgoingMessage(Message message) {
        if (this.logger.isLoggable(Level.FINER)) {
            this.logger.log(Level.FINER, String.format("Outgoing SFTP message at [%s <-> %s]: %s.",
                                                       this.channelConnection.localAddress(),
                                                       this.channelConnection.remoteAddress(),
                                                       message));
        }
    }

    private void logIncomingMessage(Message message) {
        if (this.logger.isLoggable(Level.FINER)) {
            this.logger.log(Level.FINER,
                            String.format("Incoming SFTP message at [%s <-> %s]: %s.",
                                          this.channelConnection.localAddress(),
                                          this.channelConnection.remoteAddress(),
                                          message));
        }
    }

    private void logError(Throwable e) {
        if (this.logger.isLoggable(Level.WARNING)) {
            this.logger.log(Level.WARNING, "Operation error.", e);
        }
    }

    public class ReadOperation
            implements Runnable {

        private final byte[] handle;
        private long offset;
        private final OutputStream out;
        private final TransferListener listener;
        private final Lock lock;
        private final Condition completedCondition;
        private final AtomicInteger expectedResponses;
        private long transferredBytes;
        private volatile boolean eof;
        private volatile boolean completed;
        private volatile Exception error;
        private volatile boolean stopped;

        public ReadOperation(byte[] handle, long offset, OutputStream out, TransferListener listener) {
            this.handle = handle;
            this.offset = offset;
            this.out = out;
            this.listener = listener;
            this.lock = new ReentrantLock();
            this.completedCondition = this.lock.newCondition();
            this.expectedResponses = new AtomicInteger(0);
        }

        public void execute()
                throws Exception {
            while (!transferCancelled && !completed()) {
                sendRequest();
                awaitRequestWindow();
            }

            close();
        }

        private void awaitRequestWindow() {
            while (this.expectedResponses.get() > pipelineWindowSize) {
                Time.millis(1).sleepSafe();
            }
        }

        public void run() {
            try {
                while (exchangeInProgress()) {
                    Message response = receive();
                    this.expectedResponses.decrementAndGet();

                    if (response instanceof SshFxpData) {
                        SshFxpData dataResponse = (SshFxpData) response;
                        write(dataResponse.data);
                    } else {
                        SshFxpStatus statusResponse = (SshFxpStatus) response;

                        if (statusResponse.code == StatusCode.SSH_FX_EOF.code) {
                            this.eof = true;
                        } else if (statusResponse.code != StatusCode.SSH_FX_OK.code) {
                            processError(new ServerException(statusResponse.code, statusResponse.message, statusResponse.languageTag));
                        }
                    }
                }
            } catch (Exception e) {
                processError(e);
            } finally {
                onCompleted();
            }
        }

        public void sendRequest()
                throws Exception {
            send(new SshFxpRead(nextMessageId(), this.handle, this.offset, transferBufferSizeBytes));
            this.expectedResponses.incrementAndGet();
            this.offset += transferBufferSizeBytes;
        }

        private void close()
                throws Exception {
            this.lock.lock();

            try {
                send(new SshFxpClose(nextMessageId(), this.handle));
                this.expectedResponses.incrementAndGet();
                this.stopped = true;
                awaitCompleted();
                assertNoErrors();
            } finally {
                this.lock.unlock();
            }
        }

        private boolean completed() {
            return this.eof || this.error != null;
        }

        private boolean exchangeInProgress() {
            this.lock.lock();

            try {
                return !this.stopped || this.expectedResponses.get() != 0;
            } finally {
                this.lock.unlock();
            }
        }

        private void write(byte[] data) {
            try {
                this.out.write(data);
                this.transferredBytes += data.length;
                this.listener.onTransferProgress(this.transferredBytes);
            } catch (Exception e) {
                processError(e);
            }
        }

        private void processError(Exception error) {
            if (this.error != null) {
                this.error = error;
            }
        }

        private void assertNoErrors()
                throws Exception {
            if (this.error != null) {
                throw this.error;
            }
        }

        private void awaitCompleted() {
            while (!this.completed) {
                try {
                    this.completedCondition.await(1, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    // ignore
                }
            }
        }

        private void onCompleted() {
            this.lock.lock();

            try {
                this.completed = true;
                this.completedCondition.signalAll();
            } finally {
                this.lock.unlock();
            }
        }
    }

    public class WriteOperation
            implements Runnable {

        private final InputStream in;
        private final byte[] handle;
        private long offset;
        private final TransferListener listener;
        private final Lock lock;
        private final Condition completedCondition;
        private final AtomicInteger expectedResponses;
        private final AtomicReference<Exception> error;
        private final byte[] buffer;
        private long transferredBytes;
        private volatile boolean eof;
        private volatile boolean completed;
        private volatile boolean stopped;

        public WriteOperation(InputStream in, byte[] handle, long offset, TransferListener listener) {
            this.in = in;
            this.handle = handle;
            this.offset = offset;
            this.listener = listener;
            this.lock = new ReentrantLock();
            this.completedCondition = this.lock.newCondition();
            this.expectedResponses = new AtomicInteger(0);
            this.error = new AtomicReference<Exception>();
            this.buffer = new byte[transferBufferSizeBytes];
        }

        public void execute()
                throws Exception {
            while (!transferCancelled && !completed()) {
                sendRequest();
                awaitRequestWindow();
            }

            close();
        }

        @Override
        public void run() {
            try {
                while (exchangeInProgress()) {
                    Message response = receive();
                    this.expectedResponses.decrementAndGet();

                    SshFxpStatus statusResponse = (SshFxpStatus) response;

                    if (statusResponse.code != StatusCode.SSH_FX_OK.code) {
                        processError(new ServerException(statusResponse.code, statusResponse.message, statusResponse.languageTag));
                    }
                }
            } catch (Exception e) {
                processError(e);
            } finally {
                onCompleted();
            }
        }

        private void awaitRequestWindow() {
            while (this.expectedResponses.get() > pipelineWindowSize) {
                Time.millis(1).sleepSafe();
            }
        }

        private boolean completed() {
            return this.eof || this.error.get() != null;
        }

        private void sendRequest()
                throws Exception {
            try {
                int read = this.in.read(this.buffer);

                if (read != -1) {
                    send(new SshFxpWrite(nextMessageId(), this.handle, this.offset, this.buffer, 0, read));
                    this.expectedResponses.incrementAndGet();
                    this.offset += read;
                    this.transferredBytes += read;
                    this.listener.onTransferProgress(this.transferredBytes);
                } else {
                    this.eof = true;
                }
            } catch (IOException e) {
                processError(e);
            }
        }

        private void close()
                throws Exception {
            this.lock.lock();

            try {
                this.stopped = true;
                send(new SshFxpClose(nextMessageId(), this.handle));
                this.expectedResponses.incrementAndGet();
                awaitCompleted();
                assertNoErrors();
            } finally {
                this.lock.unlock();
            }
        }

        private boolean exchangeInProgress() {
            this.lock.lock();

            try {
                return !this.stopped || this.expectedResponses.get() != 0;
            } finally {
                this.lock.unlock();
            }
        }

        private void processError(Exception error) {
            this.error.compareAndSet(null, error);
        }

        private void assertNoErrors()
                throws Exception {
            if (this.error.get() != null) {
                throw this.error.get();
            }
        }

        private void awaitCompleted() {
            while (!this.completed) {
                try {
                    this.completedCondition.await(1, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    // ignore
                }
            }
        }

        private void onCompleted() {
            this.lock.lock();

            try {
                this.completed = true;
                this.completedCondition.signalAll();
            } finally {
                this.lock.unlock();
            }
        }
    }
}
