package com.nifty.cloud.ncs.transfer;

import java.io.File;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nifty.cloud.ncs.NiftyCloudClientException;
import com.nifty.cloud.ncs.service.CompleteMultipartUploadResult;
import com.nifty.cloud.ncs.service.NcsObject;
import com.nifty.cloud.ncs.service.PartETag;
import com.nifty.cloud.ncs.service.StorageClient;
import com.nifty.cloud.ncs.service.UploadPartRequest;

/**
 * アップロードのためのCallableクラスです。
 */
public class UploadCallable implements Callable<String>
{
	/** マルチパートアップロードの最大パート数 */
    public static final int MAXIMUM_UPLOAD_PARTS = 10000;

	private final StorageClient client;
	private final ExecutorService threadPool;
	private final NcsObject object;
	private File file;
	private InputStream inputStream;
	private long contentLength;
	private String multipartUploadId;

	private static final Log log = LogFactory.getLog(UploadCallable.class);
	private final TransferManagerConfiguration configuration;
	private final List<Future<PartETag>> futures = new ArrayList<Future<PartETag>>();

    private static final int DELAY_SUBMIT = 400;

	/**
	 * ファイルを指定してUploadCallableを構築します。
	 *
	 * @param client
	 *  ストレージクライアント
	 * @param configuration
	 *  Manager設定
	 * @param threadPool
	 *  スレッドプール
	 * @param object
	 *  アップロードするオブジェクト情報を含むNcsObject
	 * @param file
	 *  ファイル
	 */
	public UploadCallable(StorageClient client, TransferManagerConfiguration configuration, ExecutorService threadPool, NcsObject object, File file)
	throws NiftyCloudClientException
	{
		this(client, configuration, threadPool, object);
		if (file == null) {
			throw (new NiftyCloudClientException("The argument of file must be specified."));
		}
		this.file = file;
		this.contentLength = file.length();
	}

	/**
	 * 入力ストリームを指定してUploadCallableを構築します。
	 *
	 * @param client
	 *  ストレージクライアント
	 * @param configuration
	 *  Manager設定
	 * @param threadPool
	 *  スレッドプール
	 * @param object
	 *  アップロードするオブジェクト情報を含むNcsObject
	 * @param inputStream
	 *  アップロードするデータストリーム
	 * @param contentLength
	 *  データサイズ
	 */
	public UploadCallable(StorageClient client, TransferManagerConfiguration configuration, ExecutorService threadPool, NcsObject object, InputStream inputStream, long contentLength)
	{
		this(client, configuration, threadPool, object);
		this.inputStream = inputStream;
		this.contentLength = contentLength;
	}

	private UploadCallable(StorageClient client, TransferManagerConfiguration configuration, ExecutorService threadPool, NcsObject object)
	{
		this.client = client;
		this.configuration = configuration;

		this.threadPool = threadPool;
		this.object = object;
	}

	/**
	 * アップロードを行います。
	 */
	@Override
	public String call() throws Exception
	{
		if (isMultipartUpload()) {
			// マルチパートアップロード
			return (uploadInMultipart());
		} else {
			// 通常アップロード
			return (client.putObject(object, file, null));
		}
	}

	/**
	 * マルチパートアップロードを使用するかどうかの判定<br>
	 * {@link TransferManagerConfiguration#multipartUploadThreshold}のサイズより大きい場合はマルチパートアップロードを行います。
	 */
	public boolean isMultipartUpload()
	{
		return (contentLength > configuration.getMultipartUploadThreshold());
	}

	/**
	 * 並列処理可能かどうかの判定<br>
	 * アップロード対象がファイルの場合のみ可能
	 */
	private boolean isUploadParallelizable()
	{
		return (file != null);
	}

	/**
	 * Futureを取得します。
	 *
	 * @return Future
	 */
	List<Future<PartETag>> getFutures()
	{
		return (futures);
	}

	/**
	 * マルチパートアップロードIDを取得します。
	 *
	 * @return マルチパートアップロードID
	 */
	String getMultipartUploadId()
	{
		return (multipartUploadId);
	}

	/**
	 * マルチパートアップロードします。
	 *
	 * @return
	 * @throws Exception
	 */
	private String uploadInMultipart() throws Exception
	{
		final String bucketName = object.getBucketName();
		final String objectName = object.getObjectName();

		long optimalPartSize = getOptimalPartSize();

		// アップロードの初期化
		multipartUploadId = client.initiateMultipartUpload(object, null);
		log.debug("Initiated new multipart upload: " + multipartUploadId);

		try {
			UploadPartRequestFactory requestFactory = null;

			if (isUploadParallelizable()) {
				// 並列アップロード
				requestFactory = new UploadPartRequestFactory(bucketName, objectName,
						file, multipartUploadId, optimalPartSize);
				requestFactory.setParallel(true);
				uploadPartsInParallel(requestFactory);

				// 非同期処理のため終了を待たずnullを返却
				return (null);
			} else {
				// 直列アップロード
				requestFactory = new UploadPartRequestFactory(bucketName, objectName,
						inputStream, contentLength, multipartUploadId, optimalPartSize);

				return (uploadPartsInSeries(requestFactory));
			}

		} catch (Exception e) {
			//e.printStackTrace();
			abortUpload();
			throw (e);

		} finally {
			if (inputStream != null) {
				try {
					inputStream.close();
				} catch (Exception e) {
					log.warn("Unable to cleanly close input stream: " + e.getMessage(), e);
				}
			}
		}
	}

	/**
	 * アップロードを中断します。
	 *
	 */
	public void abortUpload()
	{
		try {
			// マルチアップロード中断
			client.abortMultipartUpload(object.getBucketName(), object.getBucketName(), multipartUploadId);
		} catch (Exception e2) {
			log.warn("Unable to abort multipart upload, you may need to manually remove uploaded parts: " + e2.getMessage(), e2);
		}
	}

	/**
	 * 最適なパートサイズを計算します。
	 *
	 * @param isUsingEncryption
	 * @return
	 */
	private long getOptimalPartSize()
	{
		double partSize = (double)contentLength / (double)MAXIMUM_UPLOAD_PARTS;
		partSize = Math.ceil(partSize);
		long optimalPartSize = (long)Math.max(partSize, configuration.getMinimumUploadPartSize());
		log.debug("Calculated optimal part size: " + optimalPartSize);

		return (optimalPartSize);
	}

	/**
	 * パートを直列でアップロードします。
	 *
	 * @param requestFactory
	 * @return
	 */
	private String uploadPartsInSeries(UploadPartRequestFactory requestFactory)
	{
		final List<PartETag> partETags = new ArrayList<PartETag>();
		String bucketName = object.getBucketName();
		String objectName = object.getObjectName();

		while (requestFactory.hasMoreRequests()) {
			if (threadPool.isShutdown()) {
				throw (new CancellationException("TransferManager has been shutdown"));
			}
			PartETag partETag = client.uploadPart(requestFactory.getNextUploadPartRequest());
			partETags.add(partETag);
		}

		CompleteMultipartUploadResult completeMultipartUploadResult =
			client.completeMultipartUpload(bucketName, objectName, multipartUploadId, partETags);

		return (completeMultipartUploadResult.getETag());
	}

	/**
	 * パートを並列でアップロードします。
	 *
	 * @param requestFactory
	 */
	private void uploadPartsInParallel(UploadPartRequestFactory requestFactory)
	{
		while (requestFactory.hasMoreRequests()) {
			if (threadPool.isShutdown()) {
				throw (new CancellationException("TransferManager has been shutdown."));
			}
			UploadPartRequest request = requestFactory.getNextUploadPartRequest();
			futures.add(threadPool.submit(new UploadPartCallable(client, request)));

			try {
				Thread.sleep(DELAY_SUBMIT);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
