package jp.hasc.hasctool.core.runtime.hierarchize.remote;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;

import jp.hasc.hasctool.core.data.EnumCommand;
import jp.hasc.hasctool.core.messaging.MessageProcessor;
import jp.hasc.hasctool.core.messaging.MessageQueue;
import jp.hasc.hasctool.core.runtime.AbstractTask;
import jp.hasc.hasctool.core.runtime.RuntimeContext;
import jp.hasc.hasctool.core.runtime.hierarchize.XbdExecutionRequest;
import jp.hasc.hasctool.ui.views.XERequestSenderView;

import com.thoughtworks.xstream.XStream;

/**
 * このブロックへ入力された{@link XbdExecutionRequest}を、TCP経由で{@link XERequestReceiver}へ送信します。
 * @author iwasaki
 */
public class XERequestSender extends AbstractTask {
	/** logger for this class */
	private final static org.apache.commons.logging.Log LOG = org.apache.commons.logging.LogFactory
			.getLog(XERequestSender.class);
	
	private MessageQueue inputQueue_;
	private String defaultXbdFilePath_ = null;
	
	private String receivers_;
	
	public MessageProcessor getInputPort() { return inputQueue_; }
	
	private boolean showView_ = false;
	private XERequestSenderView view_ = null;
	
	public static class JobStatus {
		public static final int JOB_STATUS_WAIT = 0;
		public static final int JOB_STATUS_DOING = 1;
		public static final int JOB_STATUS_DONE = 2;
		public int nodeIndex = -1;
		public int status = JOB_STATUS_WAIT;
	}
	private JobStatus[] jobStatuses_ = null;
	
	private boolean needJobStatuses() {
		return view_!=null;
	}
	
	@Override
	public void setup(RuntimeContext context) {
		super.setup(context);
		inputQueue_ = context.createDefaultMessageQueue();
		//
		if (showView_) view_=XERequestSenderView.openView(context, this);
	}
	
	private Object readyLock_ = new Object();
	
	private XStream xStreamForOutput_ = XERequestReceiver.createXStreamForTCPMessaging();
	
	private class ReceiverInfo {
		private String label_;
		private Socket socket_;
		private DataOutputStream dataOutputStream_;
		private DataInputStream dataInputStream_;
		private boolean ready_ = false;
		private int currentJobIndex_ = -1;
		
		public void start(String host, int port) throws IOException {
			InetAddress addr=InetAddress.getByName(host);
			socket_=new Socket(addr, port);
			dataOutputStream_=new DataOutputStream(socket_.getOutputStream());
			dataInputStream_=new DataInputStream(socket_.getInputStream());
			//
			new Thread() {
				@Override
				public void run() {
					try{
						XStream xs = XERequestReceiver.createXStreamForTCPMessaging();
						while(!isShutdown()) {
							Object msg = XERequestReceiver.receiveMessage(xs,dataInputStream_);
							if (XERequestEnumCommand.READY==msg) {
								if (needJobStatuses()) {
									if (jobStatuses_!=null) synchronized (jobStatuses_) {
										if (currentJobIndex_>=0)
											jobStatuses_[currentJobIndex_].status=JobStatus.JOB_STATUS_DONE;
									}
									if (view_!=null) {
										view_.onUpdate(currentJobIndex_);
									}
								}
								synchronized (readyLock_) {
									ready_=true;
									readyLock_.notifyAll();
								}
							}else if (XERequestEnumCommand.CLOSE_CONNECTION==msg) {
								close();
								return;
							}
						}
					}catch(IOException ex) {
						LOG.warn("IOException",ex);
					}
				}

			}.start();
		}

		private void close() throws IOException {
			LOG.debug("ReceiverInfo close");
			dataOutputStream_.close();
			dataInputStream_.close();
			socket_.close();
		}
		
		public String getLabel() {
			return label_;
		}

		public void setLabel(String label) {
			label_ = label;
		}

		public boolean isReady() {
			return ready_;
		}

		public void setReady(boolean ready) {
			ready_ = ready;
		}

		public void send(Object message) throws IOException {
			XERequestReceiver.sendMessage(xStreamForOutput_,message,dataOutputStream_);
		}

		public void setCurrentJobIndex(int currentJobIndex) {
			this.currentJobIndex_ = currentJobIndex;
		}

		public void requestClose() {
			try{
				send(XERequestEnumCommand.CLOSE_CONNECTION);
			}catch(IOException ex) {
				LOG.warn("IOException",ex);
			}
		}
	}
	
	private ArrayList<ReceiverInfo> receiverInfos_;
	private HashMap<String,ReceiverInfo> receiverInfosMap_;
	
	@Override
	protected void run() throws InterruptedException {
		// read all XbdExecutionRequests
		ArrayList<XbdExecutionRequest> requests=new ArrayList<XbdExecutionRequest>();
		while(true){
			Object msg=inputQueue_.readMessage();
			if (msg instanceof XbdExecutionRequest) {
				XbdExecutionRequest req = (XbdExecutionRequest) msg;
				if (req.getXbdFilePath()==null) req.setXbdFilePath(defaultXbdFilePath_);
				requests.add(req);
			}else if (EnumCommand.END==msg) {
				break;
			}
		}
		
		// setup receivers
		receiverInfos_=new ArrayList<XERequestSender.ReceiverInfo>();
		receiverInfosMap_=new HashMap<String, XERequestSender.ReceiverInfo>();
		updateReceivers();
		
		if (needJobStatuses()) {
			jobStatuses_=new JobStatus[requests.size()];
			for(int i=0;i<jobStatuses_.length;++i) jobStatuses_[i]=new JobStatus();
			if (view_!=null) {
				view_.onUpdate(XERequestSenderView.INDEX_DRAW_ALL);
			}
		}
		
		// send requests
		for(int i=0,inum=requests.size();i<inum;++i) {
			XbdExecutionRequest req = requests.get(i);
			
			// select a receiver
			ReceiverInfo targetReceiver;
			int nodeIndex=-1;
			synchronized (readyLock_) {
				loop: while(true) {
					synchronized (receiverInfos_) {
						for(int j=0,jnum=receiverInfos_.size();j<jnum;++j) {
							ReceiverInfo rinfo = receiverInfos_.get(j);
							if (rinfo.isReady()) {
								targetReceiver=rinfo;
								rinfo.setReady(false);
								nodeIndex=j;
								break loop;
							}
						}
					}
					readyLock_.wait();
				}
			}
			
			// send
			LOG.debug("send execution request to "+targetReceiver.getLabel());
			if (needJobStatuses()) {
				if (jobStatuses_!=null) synchronized (jobStatuses_) {
					jobStatuses_[i].nodeIndex=nodeIndex;
					jobStatuses_[i].status=JobStatus.JOB_STATUS_DOING;
				}
				if (view_!=null) {
					view_.onUpdate(i);
				}
			}
			try{
				targetReceiver.setCurrentJobIndex(i);
				targetReceiver.send(req);
			}catch(IOException ex) {
				LOG.warn("IOException",ex);
			}
		}
		
		// disconnect
		synchronized (receiverInfos_) {
			for(ReceiverInfo rinfo : receiverInfos_) {
				rinfo.requestClose();
			}
			receiverInfos_=null;
		}
		//
		LOG.info("all requests done");
	}

	public void updateReceivers() {
		// parse property
		String[] receiverArrays;		
		final String header="hostList=";
		if (receivers_.startsWith(header)) {
			receiverArrays = receivers_.substring(header.length()).split(",");
		}else{
			LOG.error("invalid format: "+receivers_);
			throw new RuntimeException("invalid format: "+receivers_);
		}
		// connect or disconnect
		if (receiverInfos_==null) return;
		synchronized (receiverInfos_) {
			HashMap<String,ReceiverInfo> oldMap=new HashMap<String, XERequestSender.ReceiverInfo>(receiverInfosMap_);
			receiverInfosMap_.clear();
			receiverInfos_.clear();
			for(String hostAndPort : receiverArrays) {
				try{
					ReceiverInfo rinfo = oldMap.get(hostAndPort);
					if (rinfo!=null) {
						// already connected
						oldMap.remove(hostAndPort);
					}else{
						// connect
						rinfo=new ReceiverInfo();
						rinfo.setLabel(hostAndPort);
						LOG.debug("connect to "+hostAndPort);
						//
						int idx = hostAndPort.indexOf(":");
						int port;
						String host;
						if (idx<0) {
							host=hostAndPort;
							port=XERequestReceiver.DEFAULT_TCP_PORT;
						}else{
							port=Integer.parseInt(hostAndPort.substring(idx+1));
							host=hostAndPort.substring(0,idx);
						}
						rinfo.start(host,port);
					}
					receiverInfos_.add(rinfo);
					receiverInfosMap_.put(hostAndPort, rinfo);
				}catch(IOException ex) {
					LOG.warn("IOException",ex);
				}
			}
			// disconnect from removed hosts
			for(ReceiverInfo rinfo : oldMap.values()) {
				rinfo.requestClose();
			}
		}
	}
	
	public void setReceivers(String receivers) {
		receivers_ = receivers;
	}

	public void setDefaultXbdFilePath(String defaultXbdFilePath) {
		defaultXbdFilePath_ = defaultXbdFilePath;
	}


	public void setShowView(boolean showView) {
		showView_ = showView;
	}

	public JobStatus[] getJobStatuses() {
		return jobStatuses_;
	}

	public String getReceivers() {
		return receivers_;
	}

}
