path: root/dozentenmodul/src/main/java/org/openslx/dozmod/filetransfer/DownloadTask.java

package org.openslx.dozmod.filetransfer;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.RandomAccessFile;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import javax.net.ssl.SSLContext;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.openslx.bwlp.thrift.iface.TransferInformation;
import org.openslx.bwlp.thrift.iface.TransferState;
import org.openslx.dozmod.Config;
import org.openslx.filetransfer.DataReceivedCallback;
import org.openslx.filetransfer.Downloader;
import org.openslx.filetransfer.FileRange;
import org.openslx.filetransfer.Transfer;
import org.openslx.filetransfer.WantRangeCallback;
import org.openslx.filetransfer.util.ChunkList;
import org.openslx.filetransfer.util.FileChunk;
import org.openslx.thrifthelper.ThriftManager;
import org.openslx.util.Util;

/**
 * Execute file download in a background thread and update the progress.
 */
public class DownloadTask extends TransferTask {

	/**
	 * Logger instance for this class.
	 */
	private final static Logger LOGGER = LogManager.getLogger(DownloadTask.class);
	
	private static final AtomicInteger THREAD_ID = new AtomicInteger();
	
	private final String host;
	private final int portPlain;
	private final int portSsl;
	private final SSLContext sslCtx;
	private final String downloadToken;
	private final RandomAccessFile fileHandle;
	private final ChunkList chunks;
	private final long startTime;
	private boolean fileWritable = true;

	public DownloadTask(String host, TransferInformation ti, SSLContext ctx, File destinationFile, long fileSize,
			List<byte[]> sha1Sums) throws FileNotFoundException {
		super(destinationFile, fileSize);
		this.host = host;
		this.portPlain = ti.plainPort;
		this.portSsl = ti.sslPort;
		this.downloadToken = ti.token;
		this.sslCtx = ctx;
		this.fileHandle = new RandomAccessFile(destinationFile, "rw");
		this.chunks = new ChunkList(fileSize, sha1Sums);
		this.startTime = System.currentTimeMillis();
	}

	private class DownloadHandler implements WantRangeCallback, DataReceivedCallback {
		private FileChunk current = null;
		private byte[] buffer = null;
		// progress counter
		private long currentSpeed = 0;
		private long currentBytes = 0;
		private long lastUpdate = 0;
		private long lastBytes = 0;

		@Override
		public FileRange get() {
			handleCompletedChunk(current, buffer);
			consecutiveInitFails.lazySet(0);
			try {
				current = chunks.getMissing();
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				return null;
			}
			if (current == null)
				return null;
			buffer = new byte[current.range.getLength()];
			return current.range;
		}

		@Override
		public boolean dataReceived(final long fileOffset, final int dataLength, final byte[] data) {
			if (current == null)
				throw new IllegalStateException("dataReceived without current chunk");
			if (!current.range.contains(fileOffset, fileOffset + dataLength))
				throw new IllegalStateException("dataReceived with file data out of range");
			System.arraycopy(data, 0, buffer, (int) (fileOffset - current.range.startOffset), dataLength);
			currentBytes += dataLength;
			final long now = System.currentTimeMillis();
			if (lastUpdate + UPDATE_INTERVAL_MS < now) {
				synchronized (this) {
					// Calculate updated speed
					lastBytes = (lastBytes * 2 + currentBytes) / 3;
					currentSpeed = (1000 * lastBytes) / (now - lastUpdate);
					lastUpdate = now;
				}
				// Reset counters
				currentBytes = 0;
			}
			return fileWritable;
		}

		private long getCurrentSpeed() {
			synchronized (this) {
				return currentSpeed;
			}
		}

	}

	private void handleCompletedChunk(FileChunk chunk, byte[] buffer) {
		if (chunk == null)
			return;
		// TODO: Hash check, async
		try {
			synchronized (fileHandle) {
				fileHandle.seek(chunk.range.startOffset);
				fileHandle.write(buffer, 0, chunk.range.getLength());
			}
			chunks.markCompleted(chunk, true);
		} catch (Exception e) {
			LOGGER.error("Could not write to file at offset " + chunk.range.startOffset, e);
			fileWritable = false;
		}
	}

	private class DownloadThread extends TransferThread {
		private Downloader downloader = null;
		private DownloadHandler cb = new DownloadHandler();
		
		public DownloadThread() {
			super("UpConn#" + THREAD_ID.incrementAndGet());
		}
		
		private Exception initPlain(Exception ex) {
			if (portPlain <= 0 || portPlain > 65535)
				return ex;
			LOGGER.info("Establishing plain download connection to " + host + ":" + portPlain);
			try {
				downloader = new Downloader(host, portPlain, Config.TRANSFER_TIMEOUT, null, downloadToken);
			} catch (Exception e) {
				LOGGER.info("Connection failed");
				return e;
			}
			return null;
		}
		
		private Exception initSsl(Exception ex) {
			if (portSsl <= 0 || portSsl > 65535 || sslCtx == null)
				return ex;
			LOGGER.info("Establishing SSL download connection to " + host + ":" + portSsl);
			try {
				downloader = new Downloader(host, portSsl, Config.TRANSFER_TIMEOUT, sslCtx, downloadToken);
			} catch (Exception e) {
				LOGGER.info("Connection failed");
				return e;
			}
			return null;
		}

		@Override
		public void run() {
			Exception ex = null;
			switch (Config.getFileTransferMode()) {
			case SSL:
				ex = initSsl(ex);
				if (downloader == null) {
					ex = initPlain(ex);
				}
				break;
			case SSL_ONLY:
				ex = initSsl(ex);
				break;
			case PLAIN:
			default:
				ex = initPlain(ex);
				if (downloader == null) {
					ex = initSsl(ex);
				}
				break;
			}
			if (downloader == null) {
				if (ex == null) {
					LOGGER.warn("Could not initialize new downloader because neither plain"
							+ " nor SSL transfer data is given");
				} else {
					LOGGER.warn("Could not initialize new downloader, all connection methods failed", ex);
				}
				consecutiveInitFails.incrementAndGet();
				connectFailed(this);
				return;
			}
			connectSucceeded(this);

			boolean ret = downloader.download(cb, cb);
			if (!ret) {
				consecutiveInitFails.incrementAndGet();
			}
			if (cb.current != null) {
				chunks.markFailed(cb.current);
			}
			transferEnded(this, ret);
		}

		@Override
		protected Transfer getTransfer() {
			return downloader;
		}

		@Override
		public long getCurrentSpeed() {
			return cb.getCurrentSpeed();
		}

	}
	
	@Override
	public void cancel() {
		super.cancel();
		if (downloadToken != null) {
			try {
				ThriftManager.getSatClient().cancelDownload(downloadToken);
			} catch (Exception e) {
			}
		}
	}

	@Override
	protected void cleanup() {
		Util.safeClose(fileHandle);
	}

	@Override
	protected TransferEvent getTransferEvent() {
		final TransferState state;
		final byte[] progress = chunks.getStatusArray().array();
		final String error;
		if (consecutiveInitFails.get() > 20) {
			state = TransferState.ERROR;
			error = "Cannot talk to server after 20 tries...";
		} else if (chunks.isComplete() && getTransferCount() == 0) {
			Util.safeClose(fileHandle);
			state = TransferState.FINISHED;
			error = null;
		} else {
			state = TransferState.WORKING;
			error = null;
		}
		long speed = 0;
		long timeRemaining = 0;
		long virtualSpeed = 0;
		synchronized (transfers) {
			for (TransferThread thread : transfers) {
				speed += thread.getCurrentSpeed();
			}
		}
		// 0 = complete, 1 = missing, 2 = uploading, 3 = queued for copying, 4 = copying
		if (progress != null) {
			int missing = 0;
			for (byte b : progress) {
				if (b != 0) {
					missing++;
				}
			}
			final long bytesRemaining = CHUNK_SIZE * (long) missing;
			timeRemaining = (1000 * bytesRemaining) / (speed + 1);
			virtualSpeed = ((progress.length - missing) * CHUNK_SIZE * 1000) / (System.currentTimeMillis() - startTime + 1);
		}
		return new TransferEvent(state, progress, speed, virtualSpeed, timeRemaining, error);
	}

	@Override
	protected TransferThread createNewThread() {
		return new DownloadThread();
	}
}