/* ************************************************************************
#
#  DivConq
#
#  http://divconq.com/
#
#  Copyright:
#    Copyright 2012 eTimeline, LLC. All rights reserved.
#
#  License:
#    See the license.txt file in the project's top-level directory for details.
#
#  Authors:
#    * Andy White
#
************************************************************************ */
package divconq.db.mumps;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.locks.ReentrantLock;

import com.jcraft.jsch.Channel;

import divconq.lang.Hub;
import divconq.lang.OperationResult;
import divconq.lang.TaskContext;
import divconq.lang.chars.Special;
import divconq.lang.chars.Utf8Decoder;
import divconq.lang.chars.Utf8Encoder;
import divconq.lang.StringBuilder32;
import divconq.log.Logger;
import divconq.work.IRunnable;
import divconq.work.TaskInfo;
import divconq.xml.XElement;

/**
 * Internal class.  Both a thread and an SSH channel.  All reading and writing takes place on this local thread,
 * when no data is available for reading or writing the thread blocks on the database queue and awaits a new
 * task to process.
 * 
 * @author Andy
 *
 */
public class Connection implements IRunnable {
	static public final byte[] pollstate = Utf8Encoder.encode("POLLSTATE\n");
	static public final byte[] halt = Utf8Encoder.encode("HALT\n");
	
	protected boolean busy = false;
	protected boolean stopping = false;
	protected long startStamp = 0;
	protected long lastStamp = 0;
	protected ReentrantLock lock = new ReentrantLock();
	protected DatabaseManager man = null;
	protected XElement config = null;
	
	protected Channel channel = null;
	protected InputStream mis = null;
	protected OutputStream mos = null;
	
	protected DatabaseTask dbtask = null;	
	
	/**
	 * Load the settings for this connection.
	 * 
	 * @param config xml holding the settings
	 * @param man the database pool this connection belongs to 
	 * @return messages about the success of this call
	 */
	public OperationResult init(XElement config, DatabaseManager man) {
		OperationResult or = new OperationResult(true);
		
		this.config = config;
		this.man = man;
		
		return or;
	}
	
	/**
	 * Make the connection to the database.  If successful then startup the thread to process requests.
	 * 
	 * @param slot the slot allocated in the database pool for this connection
	 * @return messages about the success of this call
	 */
	public OperationResult start() {
		OperationResult or = new OperationResult(true);
		
		try {
			this.channel = this.man.session.openChannel("shell");

			this.mis = this.channel.getInputStream();
			this.mos = this.channel.getOutputStream();
			
			this.channel.setXForwarding(false);

			// TODO configure
			this.channel.connect(3 * 1000);

			Utf8Decoder decoder = new Utf8Decoder();
			byte[] buffer = new byte[8 * 1024];		// tested 64 bytes - TODO try others, config, smaller?

			int readcnt = this.mis.read(buffer);
			ByteBuffer bb = ByteBuffer.wrap(buffer, 0, readcnt);
			
			// keep looping until we get HELLO.  before HELLO we may get Linux or GTM console messages
			while (true) {
				StringBuilder32 str = decoder.processBytesUntil(bb, Special.End);
				
				if (str != null) { 
					if (!str.toString().endsWith("HELLO")) {
						or.errorTr(318);
						return or;
					}
					
					break;
				}
				
				readcnt = this.mis.read(buffer);
				bb = ByteBuffer.wrap(buffer, 0, readcnt);
			}

			// make the connection available to manager
			Hub.instance.getWorkPool().submit(this);
		} 
		catch (Exception x) {
			or.errorTr(319, x);		
			this.stop();
		}		
		
		return or;
	}
	
	/**
	 * Block and await database tasks from database task queue.  When task is available
	 * setup the task context and write/read the request/response data. 
	 */
	@Override
	public void run(TaskInfo ctx) {
		if (this.stopping)
			return;
		
		try {
			// waiting 100ms
			this.dbtask = this.man.take();
			
			if (this.dbtask == null) { 
				this.completeWork();
				return;
			}
			
			if (!this.channel.isConnected() || this.stopping) {
				this.man.queue.add(this.dbtask);	// if we cannot handle, maybe some other connection can, return to queue
				this.dbtask = null;   // don't call complete on this
				this.completeWork();
				return;
			}
			
			this.lock.lockInterruptibly();
			
			this.busy = true;
			this.startStamp = System.currentTimeMillis();
			this.lastStamp = this.startStamp;
			
			this.lock.unlock();
			
			if (!this.channel.isConnected() || this.stopping) {
				this.man.queue.add(this.dbtask);	// if we cannot handle, maybe some other connection can, return to queue
				this.dbtask = null;   // don't call complete on this
				this.completeWork();
				return;
			}
		
			this.man.incTasksStarted();
			
			TaskContext.set(this.dbtask.context);
			
			Logger.trace("Database task starting");
			
			this.dbtask.mem.setPosition(0);
			this.dbtask.mem.copyToStream(this.mos);
			
			// toss memory so that we don't hold on to it longer than needed, gc can reclaim as needed
			this.dbtask.mem = null;			
			
			ResultParser p = new ResultParser();						
			p.load(this.mis, this.dbtask.result, this);
			
			// if parse errors then we should drop this connection, it may be in a bad state
			if (this.dbtask.result.hasErrors()) {
				this.stopping = true;
				this.completeWork();
				return;
			}
			
			this.mos.write(Connection.pollstate);		
			this.mos.flush();
			p.loadMessages(this.mis, this.dbtask.result, this);
			
			// if parse errors then we should drop this connection, it may be in a bad state
			if ((this.dbtask.result.getCode() == 321) || (this.dbtask.result.getCode() == 322)) 
				this.stopping = true;
		}
		catch (InterruptedException x) {
		}
		catch (Exception x) {
			Logger.warn("Database connection caught exception: " + x);
		}
		catch (ThreadDeath x) {
			Logger.warn("Database connection thread died: " + x);
		}
		
		this.completeWork();
	}

	public void completeWork() {
		Logger.trace("Database task completing");

		DatabaseTask task = this.dbtask;
		
		// free up resources so someone else can use this thread
		this.dbtask = null;
		this.busy = false;
		this.startStamp = 0;
		this.lastStamp = 0;
		
		// make the connection available to others
		if (!this.stopping)
			Hub.instance.getWorkPool().submit(this);
		
		// continue with the work of this task (the callback)
		try {
			if (task != null)
				task.complete();
		}
		catch (Exception x) {
			Logger.warn("Database task encountered error: " + x);
		}
		
		if (task != null)
			this.man.incTasksComlpeted();
	}
	
	/**
	 * If a command takes too long to complete then mark it as hung and give up on it.  Allow 
	 * the pool to reclaim this slot.
	 * 
	 * @param timestamp is the time when this thread will be considered hung
	 */
	public void stopTimeout(long timestamp) {
		if (this.busy && (this.lastStamp < timestamp)) {
			Logger.warn("Database connection thread hung");
			this.stop();
		}
	}
	
	/**
	 * stop right away if the thread is alive.
	 */
	public void stop() {
		this.stopping = true;
		
		try {
			this.disconnect();
		}
		catch (Exception x) {				
		}
	}
	
	/**
	 * Stops the thread only if the thread is not flagged as busy.
	 */
	public void stopNice() {
		this.stopping = true;
		
		try {
			this.lock.lockInterruptibly();
			
			// only if not busy
			try {
				if (!this.busy)
					this.disconnect();
			}
			catch (Exception x) {				
			}
			finally {
				this.lock.unlock();
			}
		} 
		catch (InterruptedException x) {
		}
	}
	
	/**
	 * Closes the SSH channel.
	 */
	public void disconnect() {		
		// try our hardest to clear the task, faster than having callers timeout
		//if (!this.busy && (this.dbtask != null)) {
		//	this.dbtask.result.errorTr(320);		TODO use this someplace
		//	this.dbtask.complete();
		//	this.dbtask = null;
		//}
		
		if ((this.channel == null) || !this.channel.isConnected())
			return;

		if (this.mos != null)
			try {
				this.mos.write(Connection.halt);
				this.mos.flush();
			} 
			catch (IOException e) {
			}
		
		this.channel.disconnect();
	}
}
