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.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import com.nifty.cloud.ncs.NiftyCloudClientException;
import com.nifty.cloud.ncs.NiftyCloudServiceException;
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.transfer.Transfer.TransferState;

/**
 * ファイルアップロードを監視するクラスです。
 */
public class UploadMonitor extends AbstractTransferMonitor<String, PartETag>
{
	private final StorageClient client;
	private final NcsObject object;
	private final UploadCallable uploadCallable;
	private String uploadId;

	private static final int POLL_INTERVAL = 5000;

	/**
	 * UploadMonitorを構築します。
	 *
	 * @param client
	 *  ストレージクライアント
	 * @param configuration
	 *  Manager設定
	 * @param threadPool
	 *  スレッドプール
	 * @param object
	 *  アップロードするオブジェクト情報を含むNcsObject
	 * @param file
	 *  アップロードするファイル
	 */
	public UploadMonitor(StorageClient client, TransferManagerConfiguration configuration, ExecutorService threadPool,
			NcsObject object, File file)
	{
		super(object);

		this.client = client;
		super.threadPool = threadPool;
		this.futures = new ArrayList<Future<PartETag>>();
		this.object = object;
		this.uploadCallable = new UploadCallable(client, configuration, threadPool, object, file);

		nextFuture = getDelayExecutor().submit(this);
	}

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

		this.client = client;
		this.threadPool = threadPool;
		this.object = object;
		this.uploadCallable = new UploadCallable(client, configuration, threadPool, object, input, contentLength);

		nextFuture = getDelayExecutor().submit(this);
	}

//	/**
//	 * スケジューリングされたThreadPoolEcecutorを取得します。
//	 *
//	 * @return
//	 */
//	private static synchronized ScheduledExecutorService getDelayExecutor()
//	{
//		if ( timedThreadPool == null )
//			timedThreadPool = new ScheduledThreadPoolExecutor(1);
//		return (timedThreadPool);
//	}

//	/**
//	 * スレッドプールを終了します。
//	 */
//	public static synchronized void shutdownNow()
//	{
//		if (timedThreadPool != null) {
//			timedThreadPool.shutdownNow();
//		}
//	}

	/**
	 * アップロードの監視を行います。
	 */
    @Override
    public String call() throws Exception
    {
    	try {
    		if ( uploadId == null ) {
    			// 処理開始
    			transfer.setState(TransferState.InProgress);
    			return (upload());
    		} else {
    			return (poll());
    		}

    	} catch ( CancellationException e ) {
    		transfer.setState(TransferState.Canceled);
    		abortUpload();
    		throw (new NiftyCloudClientException("Upload canceled"));

    	} catch ( Exception e ) {
    		transfer.setState(TransferState.Failed);
    		abortUpload();
    		throw (e);
    	}
    }

    /**
     * パートアップロードの結果に対してポーリングします。
     * 全てのパートのアップロードが完了した場合はCompleteMultipartUploadを実行し、
     * マルチパートアップロードを完了します。
     *
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private String poll() throws InterruptedException, ExecutionException
    {
    	boolean isProcessing = false;
    	for ( Future<PartETag> f : futures ) {
    		if ( f.isDone() ) {
    			// 終了している場合は結果を取得し、エラーの場合は処理中断
    			f.get();
    		} else {
    			isProcessing = true;
    		}
    	}

    	if (isProcessing) {
    		reschedule();
			return (null);
    	}

//    	for ( Future<PartETag> f : futures ) {
//    		if ( !f.isDone() ) {
//    			reschedule();
//    			return (null);
//    		}
//    	}

//    	for ( Future<PartETag> f : futures ) {
//    		if ( f.isCancelled() ) {
//    			throw (new CancellationException());
//    		}
//    	}

    	return (completeMultipartUpload());
    }

    private void abortUpload()
    {
    	cancel();
    	uploadCallable.abortUpload();
    }

    /**
     * アップロードを開始します。
     *
     * @return
     * @throws Exception
     * @throws InterruptedException
     */
    private String upload() throws Exception, InterruptedException
    {
    	String result = uploadCallable.call();

    	if ( result != null ) {
    		uploadComplete();
    	} else {
    		// 並列アップロードの場合非同期処理で、結果が返らないため
    		// モニターで監視する
    		uploadId = uploadCallable.getMultipartUploadId();
    		futures.addAll(uploadCallable.getFutures());
    		reschedule();
    	}

    	return (result);
    }

    /**
     * アップロード完了フラグを立てます。
     */
    private void uploadComplete()
    {
    	transfer.setState(TransferState.Completed);
    }

    private void reschedule()  {
//    	nextFuture = getDelayExecutor().schedule(new Callable<String>() {
//    		public String call() throws Exception
//    		{
//    			nextFuture = threadPool.submit(UploadMonitor.this);
//    			return (null);
//    		}
//    	}, POLL_INTERVAL, TimeUnit.MILLISECONDS);
    	nextFuture = getDelayExecutor().schedule(this, POLL_INTERVAL, TimeUnit.MILLISECONDS);
    }

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

    	CompleteMultipartUploadResult completeMultipartUploadResult =
    			client.completeMultipartUpload(bucketName, objectName, uploadId, collectPartETags());

    	uploadComplete();

    	return (completeMultipartUploadResult.getETag());
    }

    /**
     * パートアップロードの結果からPartETagを収集します。
     *
     * @return
     */
    private List<PartETag> collectPartETags() {
    	final List<PartETag> partETags = new ArrayList<PartETag>(futures.size());
    	for (Future<PartETag> future : futures) {
    		try {
    			partETags.add(future.get());
    		} catch (Exception e) {
    			throw ( new NiftyCloudClientException("Unable to upload part: " + e.getCause().getMessage(), e.getCause()) );
    		}
    	}
    	return partETags;
    }

    /**
     * アップロードが完了するまで待機します。
     * 完了した場合は結果が返却されます。
     *
     * @return アップロードしたオブジェクトのMD5ハッシュ値
     * @throws NiftyCloudClientException
     * @throws NiftyCloudServiceException
     * @throws InterruptedException
     */
    public String waitForUploadResult()
    		throws NiftyCloudClientException, NiftyCloudServiceException, InterruptedException
    {
    	try {
    		String result = null;
    		while (!isDone() || result == null) {
    			Future<?> f = nextFuture;
    			result = (String)f.get();
    		}
    		return (result);

    	} catch (ExecutionException e) {
    		e.printStackTrace();
    		throw (getNiftyCloudException(e));
    	}
    }

    public String getBucketName()
    {
    	return (object.getBucketName());
    }

    public String getObjectName()
    {
    	return (object.getObjectName());
    }
}
