package com.jscape.filetransfer;

import com.jscape.util.IOTools;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.jscape.util.Parameters.assertNotNull;

/**
* @author Alex Usun
*/
public class FileTransferPipedOutputStream
        extends PipedOutputStream
        implements Runnable {

    private final FileTransfer transfer;
    private final String remoteFile;
    private final long offset;
    private final long length;
    private final PipedInputStream in;
    private final Lock lock;
    private final Condition completedCondition;
    private boolean completed;
    private volatile Exception error;

    public static FileTransferPipedOutputStream streamFor(FileTransfer transfer, String remoteFile, long offset, long length)
            throws Exception {
        PipedInputStream in = new PipedInputStream(64 * 1024);
        FileTransferPipedOutputStream out = new FileTransferPipedOutputStream(transfer, remoteFile, offset, length, in);
        out.connect(in);

        return out;
    }

    private FileTransferPipedOutputStream(FileTransfer transfer, String remoteFile, long offset, long length, PipedInputStream in)  {
        assertNotNull(transfer);
        this.transfer = transfer;

        assertNotNull(remoteFile);
        this.remoteFile = remoteFile;

        this.offset = offset;
        this.length = length;

        assertNotNull(in);
        this.in = in;

        this.lock = new ReentrantLock();
        this.completedCondition = this.lock.newCondition();
    }

    @Override
    public void run() {
        this.lock.lock();

        try {
            this.transfer.resumeUpload(this.in, this.length, this.remoteFile, this.offset);
        } catch (Exception e) {
            this.error = e;
        } finally {
            this.completed = true;
            this.completedCondition.signalAll();
            this.lock.unlock();
        }
    }

    @Override
    public void write(int data)
            throws IOException {
        super.write(data);
        assertNoErrors();
    }

    @Override
    public void write(byte[] data, int offset, int length)
            throws IOException {
        super.write(data, offset, length);
        assertNoErrors();
    }

    @Override
    public synchronized void flush()
            throws IOException {
        super.close();
        awaitDataUploaded();
        assertNoErrors();
    }

    @Override
    public void close()
            throws IOException {
        super.close();
        awaitDataUploaded();
        assertNoErrors();
    }

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

        try {
            while (!this.completed) {
                try {
                    this.completedCondition.await(1, TimeUnit.MILLISECONDS);
                } catch (Exception e) {
                    // ignore
                }
            }
        } finally {
            this.lock.unlock();
        }
    }

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

}
