/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: DataStreamIoApi.java,v 1.2 2005/09/07 13:38:07 rampil Exp $
 */
package org.logical_paradox.vsan.node.api;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.util.StringUtils;
import org.logical_paradox.vsan.dac.DacProperty;
import org.logical_paradox.vsan.dac.ResponseCode;
import org.logical_paradox.vsan.node.AbstractApi;
import org.logical_paradox.vsan.node.Database;
import org.logical_paradox.vsan.node.DatabaseAccessException;
import org.logical_paradox.vsan.node.dbr.DbFile;
import org.logical_paradox.vsan.node.dbr.DbRecord;

/**
 * ストリーム入出力関連のAPI
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.2 $
 */
public class DataStreamIoApi extends AbstractApi {
	/** ロガー */
	private static final Log log = LogFactory.getLog(DataStreamIoApi.class);

	/**
	 * コンストラクタ
	 * @param db データベース
	 */
	public DataStreamIoApi(Database db) {
		super(db);
	}
	/**
	 * 指定されたレコード番号を先頭に、入力ストリームからデータを取得してデータファイルへ書き込む
	 * @param recordBegin 開始レコード番号
	 * @param in データソースたる入力ストリーム
	 * @param size データサイズ
	 * @return 書き込みを完了したレコード番号
	 * @throws DatabaseAccessException パラメータの不正などが原因で処理を続行できなかった
	 * @throws IOException クライアントとの通信に障害が発生した
	 */
	public long storeStream(long recordBegin, InputStream in, int size) throws DatabaseAccessException, IOException {
		List dbfiles = getSortedDatafiles();

		// レコード作成用のバッファ
		byte[] recordBuffer = new byte[getDatabase().getRecordSize()];

		// 指定されたレコード番号が含まれているデータファイルを探す
		int idx = getTargetDbFile(dbfiles, recordBegin);

		if(idx < 0) {
			// 指定されたレコード番号を含むデータファイルが存在しなかった
			log.error("レコード番号[" + recordBegin + "]は、このドメインデータベースに存在しません");
			throw new DatabaseAccessException(ResponseCode.NO_RECORD_EXISTS);
		}

		/*
		 * 全てのデータの出力が完了するまでループする
		 */
		long rno = recordBegin;
		int datasize = 0;
		String writeCacheStr = DacProperty.getProperty("writecache");
		int writeCache = StringUtils.isIntStr(writeCacheStr) ? Integer.parseInt(writeCacheStr) : 1;

		List cache = new ArrayList(writeCache);

		DbFile dbfile = (DbFile)dbfiles.get(idx);
		int writeBytes = 0;

		while(writeBytes < size) {
			// データを書き込む
			recordBuffer[datasize++] = (byte)in.read();
			writeBytes++;

			if(writeBytes % recordBuffer.length == 0) {
				// バッファがいっぱいになったのでレコードを作成する
				DbRecord record = new DbRecord(rno, recordBuffer.length, recordBuffer, datasize);
				cache.add(record);
				if(cache.size() == writeCache) {
					// キャッシュなしの設定か、あるいは書き込みキャッシュがいっぱいになった場合は出力
					try {
						dbfile.store(cache);
					} catch (IOException e) {
						log.error("データファイルへの書き込みに失敗:", e);
						throw new DatabaseAccessException(ResponseCode.DATAFILE_WRITE_ERROR);
					}
				}
				// バッファの初期化
				recordBuffer = new byte[getDatabase().getRecordSize()];
				datasize = 0;
				rno++;

				// レコード番号が変更になったので、データファイルが変更になったかもしれない
				DbFile dbf = null;
				for(int i = idx; i < dbfiles.size(); i++) {
					dbf = (DbFile)dbfiles.get(i);
					if(dbf.includes(rno) == true) {
						idx = i;
						// いままで扱っていたのと違うデータファイルの場合、
						// 現在の書き込みキャッシュの内容を一度現行のデータファイルに対して出力する
						dbfile.store(cache);
						cache.clear();

						// 次のデータファイルに切り替える
						dbfile = dbf;
						break;
					}
				}
				if(dbfile == null) {
					// 書き込み先レコードを扱っているデータファイルが、このドメインデータベースにない
					log.error("レコード番号[" + rno + "]は、このドメインデータベースに存在しません");
					throw new DatabaseAccessException(ResponseCode.NO_RECORD_EXISTS);
				}
			}
		}
		if(datasize > 0) {
			// 残りのレコードを出力する
			DbRecord record = new DbRecord(rno, recordBuffer.length, recordBuffer, datasize);
			// 現在選択されているデータファイルに対してレコードを出力
			cache.add(record);
			dbfile.store(cache);
		}

		// 書き込みを完了したレコード番号を返す
		return rno;
	}
	/**
	 * 指定されたレコードを復元して返す
	 * @param records 復元したいレコード番号の配列(順列にソートされていること)
	 * @return 復元されたレコード
	 */
	public DbRecord[] restoreStream(long[] records) throws DatabaseAccessException, IOException {
		DbRecord[] dbr = new DbRecord[records.length];

		if(records == null || records.length == 0) {
			return new DbRecord[0];
		}

		// 最初のレコード番号を含むデータファイルを探す
		List dbfiles = getSortedDatafiles();
		long recordBegin = records[0];

		int idx = getTargetDbFile(dbfiles, recordBegin);
		if(idx < 0) {
			// 指定されたレコード番号を含むデータファイルが存在しなかった
			log.error("レコード番号[" + recordBegin + "]は、このドメインデータベースに存在しません");
			throw new DatabaseAccessException(ResponseCode.NO_RECORD_EXISTS);
		}

		DbFile dbf = (DbFile)dbfiles.get(idx);

		// 指定された全レコードを取得する
		int di = 0;
		while(di < dbr.length) {
			long rno = records[di];
			if(dbf.includes(rno) == true) {
				// 現在選択されているデータファイルが、目的のレコードを含む場合
				dbr[di++] = dbf.restore(rno);
			} else {
				// 現在選択されているデータファイルが、目的のレコードを含まない場合
				// データファイルを再選択する
				idx++;
				if(idx >= dbfiles.size()) {
					// 指定されたレコード番号を含むデータファイルが存在しなかった
					log.error("レコード番号[" + recordBegin + "]は、このドメインデータベースに存在しません");
					throw new DatabaseAccessException(ResponseCode.NO_RECORD_EXISTS);
				}
				dbf = (DbFile)dbfiles.get(idx);
			}
		}

		return dbr;
	}
	/**
	 * 指定されたレコード番号を含むデータファイルを返す
	 * @param dbfiles データベースファイルのリスト(レコード番号順にソート済みのもの)
	 * @param recordno レコード番号
	 * @return dbfiles上のインデックス(-1:存在しない)
	 */
	protected int getTargetDbFile(List dbfiles, long recordBegin) {
		// 指定されたレコード番号が含まれているデータファイルを探す
		int idx = 0;
		for(Iterator it = dbfiles.iterator(); it.hasNext(); idx++) {
			DbFile dbf = (DbFile)it.next();
			if(dbf.getRecordBegin() <= recordBegin && dbf.getRecordEnd() >= recordBegin) {
				return idx;
			}
		}
		return -1;
	}
}
