/*
 * LOGICAL-PARADOX.ORG
 * Copyright (C)2005 satoshi akabane(akabane@logical-paradox.org)
 * $Id: Listener.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.io.InputStream;
import java.net.Socket;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.logical_paradox.common.net.ReadWriteBuffer;
import org.logical_paradox.common.session.IllegalSessionStateException;
import org.logical_paradox.common.session.Session;
import org.logical_paradox.common.util.ByteStreamConverter;
import org.logical_paradox.vsan.dac.command.CommandFactory;
import org.logical_paradox.vsan.dac.command.exception.CommandExecutionException;
import org.logical_paradox.vsan.dac.command.exception.SessionTerminationException;
import org.logical_paradox.vsan.dac.command.exception.ShutdownException;
import org.logical_paradox.vsan.dac.command.handler.DacCommand;
import org.logical_paradox.vsan.node.DatabaseAccessException;

/**
 * リスナー
 * @author satoshi akabane@logical-paradox.org
 * @version $Revision: 1.3 $
 */
public class Listener extends Thread {
	/** ロガー */
	private static final Log log = LogFactory.getLog(Listener.class);

	/** ディスクアレイコントローラー本体 */
	private final DiskArrayController dac;
	/** DACコンフィグレーション */
	private final DacConfig config;
	/** クライアントソケット */
	private Socket clientSocket;
	/** セッション */
	private Session session;
	/** ロックオブジェクト */
	private Object lock = new Object();
	/** リスナー番号 */
	private final int listenerNo;
	/** 受信バッファ */
	private ReadWriteBuffer recv;
	/** ノードコンテキスト */
	private final NodeContext context;

	/**
	 * コンストラクタ
	 * @param parent 親スレッド
	 * @param config_ DACコンフィグレーション
	 * @param no リスナー番号
	 */
	public Listener(DiskArrayController parent, NodeContext context_, int no) {
		context = context_;
		config = context_.getConfig();
		listenerNo = no;
		recv = new ReadWriteBuffer(config.getRecvsize());
		dac = parent;
	}
	/**
	 * クライアントソケットをこのリスナーで扱うように設定する
	 * @param client_ クライアントソケット
	 */
	public void accept(Socket client_) {
		// クライアントソケットを設定してロックを解除する
		synchronized(lock) {
			clientSocket = client_;
			lock.notifyAll();
		}
	}
	/**
	 * リスナー番号を返す
	 * @return リスナー番号
	 */
	public int getListenerNo() {
		return listenerNo;
	}
	/**
	 * スレッドメイン
	 */
	public void run() {
		try {
			while(true) {
				synchronized(lock) {
					// クライアントソケット接続待ち
					while(clientSocket == null) {
						lock.wait();
					}

					// クライアントからのデータ受信処理
					try {
						while(true) {
							if(session == null) {
								// まだ存在しないなら，新しいセッションを開始する
								session = DiskArrayController.getSessionManager().getSession();
								// クライアントソケットを設定する
								session.setAttribute(DacConstant.SKEY_SOCKET_INPUT_STREAM, new BufferedInputStream(clientSocket.getInputStream()));
								// ノードコンテキストを設定する
								session.setAttribute(DacConstant.SKEY_NODE_CONTEXT, context);
							}
							// 1バイト取得する
							InputStream in = (InputStream)session.getAttribute(DacConstant.SKEY_SOCKET_INPUT_STREAM);
							int b = in.read();

							if(b == -1) {
								// 接続がクローズされた
								throw new IOException();
							}

							DacCommand command = CommandFactory.getCommand(b);
							if(command == null) {
								// 指定されたタイプのコマンドが存在しないのでエラーとして無視する
								log.warn("指定タイプ[" + b + "]に対応するコマンドが未定義です");
								continue;
							}

							// コマンドの実行
							boolean execSuccess = false;
							try {
								session = command.execute(session);
								// レスポンスがある場合は，内容を送信する
								Response response = (Response)session.getAttribute(DacConstant.SKEY_DAC_RESPONSE);
								if(response != null) {
									sendResponse(response);
								}
								execSuccess = true;
							} catch (CommandExecutionException e) {
								log.error("コマンド実行中に例外が発生しました", e);
								sendResponse(new Response(ResponseCode.COMMAND_EXECUTION_EXCEPTION, new byte[0]));
							} catch (DatabaseAccessException e) {
								log.error("データベースアクセス障害", e);
								sendResponse(new Response(e.getErrorCode(), new byte[0]));
							} finally {
								// ソケットから送信されてきて、未受信のデータを全て捨てる
								// ただし、エラーが発生してゴミが残ってしまった場合のみ
								if(execSuccess == false) {
									int remains = in.available();
									for(int i = 0; i < remains; i++) {
										in.read();
									}
								}
								// セッションから，回答データを削除する
								session.removeAttribute(DacConstant.SKEY_DAC_RESPONSE);
							}
						}
					} catch (IOException e) {
						closeSession();
						lock.notifyAll();
					} catch(SessionTerminationException e) {
						closeSession();
						lock.notifyAll();

						// シャットダウンの指令である場合，DACのステータスを「終了中」に設定する
						if(e instanceof ShutdownException) {
							dac.setServerStatus(DiskArrayController.E_CLOSING);
						}
					} catch (IllegalSessionStateException e) {
						log.error("新規セッションの取得に失敗しました");
						closeSession();
						lock.notifyAll();
					}
				}
			}
		} catch(InterruptedException e) {
			// tearDown()が外部からコールされると，ここに処理がくる
			log.info("リスナー" + listenerNo + "を停止しました");
		}
	}
	/**
	 * クライアントに対してDACからのレスポンスを返却する
	 * @param data 送信するデータ
	 */
	protected void sendResponse(Response response) throws IOException {
		BufferedOutputStream bo = new BufferedOutputStream(clientSocket.getOutputStream());

		// まず，レスポンスコードを送信する
		bo.write(ByteStreamConverter.convert(response.getResponseCode()));

		byte[] data = response.getData();

		int packetCount = (data.length / DacConstant.PACKET_SIZE) + 1;
		int begin = 0;

		// 先に予定パケット数を送信する
		byte[] packetnum = ByteStreamConverter.convert(packetCount);
		bo.write(packetnum);

		// パケットを送信する
		for(int i = 0; i < packetCount; i++) {
			int len = DacConstant.PACKET_SIZE;
			if(begin + len >= data.length) {
				len = data.length - begin;
			}
			byte[] datalen = ByteStreamConverter.convert(len);
			// パケットサイズの送信
			bo.write(datalen);
			// データブロックの送信
			bo.write(data, begin, len);
		}
		bo.flush();
	}
	/**
	 * クライアントとの接続を強制的に切断する
	 * 入出力例外などが発生しても，全て無視する
	 */
	protected void closeSession() {
		log.info("セッションをクローズします");
		try {
			if(clientSocket != null) {
				clientSocket.close();
			}
		} catch (IOException e) {
			// 強制切断中に何らかの入出力例外が発生したが，無視する(先にクライアントがcloseした場合など)
		} finally {
			clientSocket = null;
			// セッションを強制的にクローズする
			if(session != null) {
				DiskArrayController.getSessionManager().removeSession(session.getSessionId());
				session = null;
			}
			// ウェイティングリストに追加する
			dac.addWaitingThread(this);
		}
	}
	/**
	 * リスナーをシャットダウンする
	 */
	protected void tearDown() {
		log.info("リスナー" + listenerNo + "が、シャットダウン要求を受信しました");
		closeSession();

		interrupt();
	}
}
