/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: DacConnection.java,v 1.3 2005/09/17 17:35:49 rampil Exp $
 */
package org.logical_paradox.vsan.dac;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;

import org.logical_paradox.common.io.ExpandableBuffer;
import org.logical_paradox.common.util.ByteStreamConverter;
import org.logical_paradox.vsan.dac.command.Protocol;
import org.logical_paradox.vsan.domain.exception.DomainAccessException;

/**
 * DACに接続するための接続オブジェクト
 * Connectionに対する操作をDACのプロトコルに変換して通信を行う
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.3 $
 */
public class DacConnection {
	/** DACに接続しているソケット */
	private Socket socket;
	/** 接続先 */
	private String host;

	/**
	 * コンストラクタ
	 * @param host 接続先ノードのアドレス
	 * @throws IOException 接続に失敗した
	 */
	protected DacConnection(String host) throws IOException {
		if(host == null || host.length() == 0) {
			throw new IllegalArgumentException();
		}

		String[] tokens = host.split(":");
		if(tokens == null || tokens.length < 2) {
			socket = new Socket(host, DacConfig.DEFAULT_PORT_NUMBER);
		} else {
			socket = new Socket(tokens[0], Integer.parseInt(tokens[1]));
		}

		this.host = host;
	}
	/**
	 * DACに接続する
	 * @param host 接続先ノードのアドレス
	 * @return 接続オブジェクト
	 * @throws IOException 接続に失敗した
	 */
	public static final DacConnection open(String host) throws IOException {
		return new DacConnection(host);
	}
	/**
	 * 比較する
	 * 比較は、接続先文字列での比較とする
	 * @param o 比較対象
	 * @return true:同じ / false:違う
	 */
	public boolean equals(Object o) {
		DacConnection con = (DacConnection)o;
		return getHost().equals(con.getHost());
		
	}
	/**
	 * ハッシュコードを返す．
	 * @return ハッシュコード
	 */
	public int hashCode() {
		return getHost().hashCode();
	}
	/**
	 * 接続先ノードのアドレスを返す
	 * @return 接続先ノードのアドレス
	 */
	public String getHost() {
		return host;
	}
	/**
	 * 最適ストアレコード番号を検索して返す．
	 * 指定されたサイズ分の空き領域を検索する．
	 * @return 選択されたレコード番号(-1: 最適ストアレコード番号なし)
	 */
	public long getBestStoreRecordNo(int size) {
		return -1;
	}
	/**
	 * 所定位置からデータを書き込む
	 * @param recordBegin 書き込み開始位置
	 * @param bstream 書き込むデータ
	 * @throws DomainAccessException 書き込みに失敗した
	 */
	public void write(long recordBegin, byte[] bstream) throws DomainAccessException {
		/*
		 * 送信データの作成
		 */
		ByteBuffer bb = ByteBuffer.allocate(1 + 8 + 4 + bstream.length).order(ByteOrder.BIG_ENDIAN);
		bb.put((byte)Protocol.P_STORE);
		bb.putLong(recordBegin);
		bb.putInt(bstream.length);
		bb.put(bstream);

		// DACにコマンドを送信する
		Response response = null;
		try {
			response = send(bb.array(), true);
		} catch(IOException e) {
			throw new DomainAccessException(e.getMessage());
		}

		if(response.getResponseCode() == ResponseCode.OK) {
			return;
		} else {
			throw new DomainAccessException("fatal error is caused in executing STORE command: code=[" + response.getResponseCode() + "]");
		}
	}
	/**
	 * 所定位置からデータを読み込む
	 * @param recordBegin 読み込み開始位置
	 * @param bstream 読み込んだデータを格納するバッファ
	 * @return 読み込んだデータ数
	 */
	public int read(long recordBegin, int records, byte[] bstream) throws DomainAccessException {
		/*
		 * 送信データの作成
		 */
		ByteBuffer bb = ByteBuffer.allocate(1 + 4 + (records * 8)).order(ByteOrder.BIG_ENDIAN);
		bb.put((byte)Protocol.P_RESTORE);
		bb.putInt(records);

		// レコードリスト
		for(int i = 0; i < records; i++) {
			bb.putLong(recordBegin + i);
		}

		// DACにコマンドを送信する
		Response response = null;

		try {
			response = send(bb.array(), true);
		} catch(IOException e) {
			throw new DomainAccessException(e.getMessage());
		}

		if(response.getResponseCode() != ResponseCode.OK) {
			throw new DomainAccessException("fatal error is caused in executing RESTORE command: code=[" + response.getResponseCode() + "]");
		}

		ByteBuffer buffer = ByteBuffer.allocate(response.getData().length).order(ByteOrder.BIG_ENDIAN);
		buffer.put(response.getData());
		buffer.rewind();

		// レコード数
		int replyRecords = buffer.getInt();
		int offset = 0;
		boolean exitf = false;

		for(int i = 0; i < replyRecords && exitf == false; i++) {
			// レコードを読む
			// ブロックサイズ / レコード番号 / 実データサイズ
			buffer.getInt();
			buffer.getLong();
			int datasize = buffer.getInt();
			byte[] dumpTarget = new byte[datasize];

			buffer.get(dumpTarget);
			int length = 0;
			if(offset + dumpTarget.length >= bstream.length) {
				exitf = true;
				length = bstream.length - offset;
			} else {
				length = dumpTarget.length;
			}

			System.arraycopy(dumpTarget, 0, bstream, offset, length);

			offset += length;
		}

		return offset;
	}
	/**
	 * DACにコマンドを送信する．
	 * @param data データ(コマンドコードを含む)
	 * @param ignore true:I/Oエラーを無視する / false:無視しないで例外を送出する
	 * @return DACからの回答
	 * @throws IOException 通信障害
	 */
	protected Response send(byte[] data, boolean ignore) throws IOException {
		byte[] rc = new byte[0];
		int resultcode = -1;

		try {
			// 通信
			BufferedOutputStream bw = new BufferedOutputStream(socket.getOutputStream());
			bw.write(data);
			bw.flush();

			// データ受信
			ExpandableBuffer exb = new ExpandableBuffer();

			// パケット数受信
			BufferedInputStream bi = new BufferedInputStream(socket.getInputStream());

			// 実行ステータス受信
			byte[] status = new byte[4];
			bi.read(status);
			resultcode = ByteStreamConverter.getInt(status);

			byte[] packetnum = new byte[4];
			bi.read(packetnum);

			int packets = ByteStreamConverter.getInt(packetnum);

			// パケットの読み込み
			byte[] packetsz = new byte[4];
			for(int i = 0; i < packets; i++) {
				bi.read(packetsz);
				int packetlen = ByteStreamConverter.getInt(packetsz);
				byte[] body = new byte[packetlen];

				// データの受信
				bi.read(body);
				exb.write(body);
			}

			rc = exb.byteStream();
		} catch(IOException e) {
			if(ignore == false) {
				throw e;
			}
		}
		return new Response(resultcode, rc);
	}
	/**
	 * DACとの接続を閉じる
	 * DACにも接続数の問題があるので、必ずクローズすること
	 * @throws IOException クローズに失敗した
	 */
	public void close() throws IOException {
		if(socket != null && socket.isClosed() == false) {
			// DAC上のセッションを閉じる
			send(new byte[]{Protocol.P_CLOSE_SESSION}, false);

			socket.close();
		}
	}
}
