//------------------------------------------------------------------------------
// <copyright file="SessionBase.java" company="Microsoft">
//      Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright>
// <summary>
//      The implementation of the SessionBase Class
// </summary>
//------------------------------------------------------------------------------

package com.microsoft.hpc.scheduler.session;

import com.microsoft.hpc.brokerlauncher.*;
import com.microsoft.hpc.exceptions.*;
import com.microsoft.hpc.sessionlauncher.*;

import com.microsoft.schemas._2003._10.serialization.arrays.ArrayOfstring;

import java.io.IOException;
import java.net.URL;
import java.util.*;

import javax.security.auth.callback.Callback;
import javax.security.auth.callback.CallbackHandler;
import javax.security.auth.callback.UnsupportedCallbackException;
import javax.xml.ws.Dispatch;
import javax.xml.ws.Holder;

import org.apache.commons.lang.NotImplementedException;
import org.apache.cxf.endpoint.Client;
import org.apache.cxf.frontend.ClientProxy;
import org.apache.cxf.transport.http.HTTPConduit;
import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
import org.apache.cxf.ws.security.wss4j.WSS4JOutInterceptor;
import org.apache.ws.security.WSConstants;
import org.apache.ws.security.WSPasswordCallback;
import org.apache.ws.security.handler.WSHandlerConstants;
import org.datacontract.schemas._2004._07.system.Version;

import com.microsoft.hpc.SOATrace.*;

/**
 * The common part for the session
 * 
 */
public class SessionBase {
	/**
	 * the trace source name
	 */
	private static final String TraceSourceName = "com.microsoft.hpc.scheduler.session";

	static String getTraceSourceName() {
		return TraceSourceName;
	}

	/**
	 * The event source
	 */
	private static final String EventSource = "SOA Session API";

	static String getEventSource() {
		return EventSource;
	}

	/**
	 * the client version
	 */
	private static Version clientVersion = new Version();
	static {// initializer
		clientVersion.setBuild(3);
		clientVersion.setMajor(0);
		clientVersion.setMinor(0);
		clientVersion.setRevision(0);
	}

	/**
	 * the endpoint address
	 */
	private String _endpointReference;

	/**
	 * The schema for broker
	 */
	@SuppressWarnings("unused")
	private final TransportScheme _schema;

	private SOATraceSource _traceSource;

	/**
	 * the scheduler headnode name
	 */
	@SuppressWarnings("unused")
	private final String _headnode;

	private final SessionInfo _info;

	private final int serviceJobId;

	// static private IntPtr hwnd = IntPtr.Zero;
	// Not used for prompt for credentials

	static private Boolean bConsole;

	/**
	 * Username needed for SetHttps
	 */
	private static String username = "";
	/**
	 * Password needed for SetHttps
	 */
	private static String password = "";

	// It can't be constructed outside
	SessionBase(SessionInfo info, String headnode, SOATraceSource traceSource) {
		serviceJobId = info.getId();
		_schema = TransportScheme.valueOf(info.getTransportScheme().get(0));

		for (String uri : info.getBrokerEpr().getValue().getString()) {
			if (uri != null && uri != "") {
				_endpointReference = uri;
				// this will automatically get the priority order
				break;
			}
		}

		this._info = info;
		this._headnode = headnode;
		this._traceSource = traceSource;
	}

	SessionInfo getInfo() {
		return _info;
	}

	// public ISchedulerJob ServiceJob
	// Java clients are unable to connect to the scheduler

	public int getId() {
		return serviceJobId;
	}

	/**
	 * Get the default endpoint reference for the load balancer of this session
	 * 
	 * @return endpoint address
	 */
	public String getEndpointReference() {
		return _endpointReference;
	}

	/**
	 * the client version
	 * 
	 * @return client version number
	 */
	public Version getClientVersion() {
		return clientVersion;
	}

	public Version getServerVersion() {
		return this._info.getServerVersion().getValue();
	}

	SOATraceSource getTraceSource() {
		return this._traceSource;
	}

	/**
	 * Set the interface mode.
	 * 
	 * @param console
	 *            Use console or not
	 */
	public static void setInterfaceMode(Boolean console) {
		bConsole = console;
	}

	/**
	 * the help function to check the sanity of SessionStartInfo before creating
	 * session
	 * 
	 * @param startInfo
	 */
	protected static void checkSessionStartInfo(SessionStartInfo startInfo) {
		if (startInfo == null) {
			throw new ArgumentNullException("startInfo");
		}
		if (startInfo.headnode() == null) {
			throw new ArgumentException(SR.HeadnodeCantBeNull);
		}
		if (startInfo.getServiceName() == null) {
			throw new ArgumentException(SR.ServiceNameCantBeNull);
		}
		if (startInfo.getClientIdleTimeout() != null
				&& startInfo.getClientIdleTimeout().getValue() <= 0) {
			throw new ArgumentException(SR.ClientIdleTimeoutNotNegative);
		}
		if (startInfo.getSessionIdleTimeout() != null
				&& startInfo.getSessionIdleTimeout().getValue() < 0) {
			throw new ArgumentException(SR.SessionIdleTimeoutNotNegative);
		}
		if (startInfo.getMessagesThrottleStopThreshold() != null
				&& startInfo.getMessagesThrottleStopThreshold().getValue() < 0) {
			throw new ArgumentException(SR.MessageThrottleStopThresholdPositive);
		}
		if (startInfo.getMessagesThrottleStopThreshold() != null
				&& (startInfo.getMessagesThrottleStartThreshold() == null || startInfo
						.getMessagesThrottleStartThreshold().getValue() <= startInfo
						.getMessagesThrottleStopThreshold().getValue())) {
			throw new ArgumentException(SR.MessageThrottleStartGreaterStop);
		}
		if (startInfo.getMessagesThrottleStartThreshold() != null
				&& startInfo.getMessagesThrottleStartThreshold().getValue() < 0) {
			throw new ArgumentException(SR.MessageThrottleStartGreaterStop);
		}
		if (startInfo.getAllocationShrinkLoadRatioThreshold() != null
				&& startInfo.getAllocationShrinkLoadRatioThreshold().getValue() < 0) {
			throw new ArgumentException(
					SR.AllocationShrinkLoadRatioThresholdNonNegative);
		}
		if (startInfo.getAllocationShrinkLoadRatioThreshold() != null
				&& (startInfo.getAllocationGrowLoadRatioThreshold() == null || startInfo
						.getAllocationGrowLoadRatioThreshold().getValue() <= startInfo
						.getAllocationShrinkLoadRatioThreshold().getValue())) {
			throw new ArgumentException(SR.LoadRatioGrowGreaterThanShrink);
		}
	}

	static <T extends SessionBase> SessionBase createSessionInternal(
			SessionStartInfo startInfo, int timeoutMilliseconds, Boolean durable)
			throws TimeoutException, SessionException {
		Date targetTimeout;

		username = startInfo.getUsername().getValue();
		password = startInfo.getPassword().getValue();

		SessionBase.checkCredential(startInfo);

		SessionBase.checkSessionStartInfo(startInfo);
		if (timeoutMilliseconds < 0) {
			throw new ArgumentException(SR.InvalidTimeout,
					"timeoutMilliseconds");
		} else {
			targetTimeout = new Date(System.currentTimeMillis()
					+ timeoutMilliseconds);
		}

		if (!startInfo.getTransportScheme().contains(
				TransportScheme.Http.getName())) {
			throw new NotImplementedException(
					"Transport scheme not implemented: "
							+ TransportScheme.Http.getName());
		}

		ISessionLauncher client = getSessionLauncher(startInfo.headnode());

		IBrokerLauncher brokerLauncher = null;
		Exception innerException = null;
		SOATraceSource traceSource = null;

		try {
			setTimeout(client, getTimeout(targetTimeout));

			int sessionid = 0;
			List<String> eprs = null;

			for (int retry = 0; retry < 3; retry++) {
				try {
					Holder<ArrayOfstring> allocateResultHolder = new Holder<ArrayOfstring>();
					Holder<Integer> sessionIdHolder = new Holder<Integer>();

					SessionBase.checkCredential(startInfo);
					if (durable) {
						client.allocateDurable(startInfo, "https://",
								allocateResultHolder, sessionIdHolder);
					} else {
						client.allocate(startInfo, "https://",
								allocateResultHolder, sessionIdHolder);
					}

					traceSource = new SOATraceSource(Integer
							.toString(sessionid), SessionBase
							.getTraceSourceName(), SessionBase.getEventSource());

					sessionid = sessionIdHolder.value;
					ArrayOfstring allocateResult = allocateResultHolder.value;
					eprs = allocateResult.getString();

					if (eprs == null || eprs.size() == 0) {
						throw new SessionException(SR.NoBrokerNodeFound);
					}

					TraceHelper
							.TraceInfo(
									traceSource,
									"Get the EPR list from headnode. number of eprs=%d",
									eprs.size());
					break;

				} catch (ISessionLauncherAllocateSessionFaultFaultFaultMessage ex) {
					if (retry < 2) {
						startInfo.setUsername((String) null);
						startInfo.setPassword((String) null);
						continue;
					} else
						throw Utility
								.translateFaultException(ex.getFaultInfo());
				} catch (ISessionLauncherAllocateDurableSessionFaultFaultFaultMessage ex) {
					if (retry < 2) {
						startInfo.setUsername((String) null);
						startInfo.setPassword((String) null);
						continue;
					} else
						throw Utility
								.translateFaultException(ex.getFaultInfo());
				} catch (Exception e) {
					throw new SessionException(e);
				}
			}

			TraceHelper.TraceInfo(traceSource, "Get Eprs from SessionLauncher");

			for (String epr : eprs) {
				try {
					TraceHelper.TraceInfo(traceSource, "Try the broker at: %s",
							epr);
					brokerLauncher = getBrokerLauncher(epr);

					setTimeout(brokerLauncher, getTimeout(targetTimeout));

					SessionInfo info;
					if (!durable) {
						BrokerInitializationResult result = brokerLauncher
								.create(startInfo, sessionid);
						info = BuildSessionInfo(result, durable, startInfo
								.isSecure(), sessionid, epr, TransportScheme
								.valueOf(startInfo.getTransportScheme().get(0)));
					} else {
						BrokerInitializationResult result = brokerLauncher
								.createDurable(startInfo, sessionid);
						info = BuildSessionInfo(result, durable, startInfo
								.isSecure(), sessionid, epr, TransportScheme
								.valueOf(startInfo.getTransportScheme().get(0)));
					}

					if (!durable) {
						return new Session(info, startInfo.headnode(),
								startInfo.isShareSession(), traceSource);
					} else {
						return new DurableSession(info, startInfo.headnode(),
								traceSource);
					}
				} catch (TimeoutException e) {
					throw new TimeoutException(SR.ConectBrokerLauncherTimeout,
							timeoutMilliseconds);
				} catch (IBrokerLauncherCreateSessionFaultFaultFaultMessage e) {
					throw Utility.translateFaultException(e.getFaultInfo());
				} catch (IBrokerLauncherCreateDurableSessionFaultFaultFaultMessage e) {
					throw Utility.translateFaultException(e.getFaultInfo());
				} catch (Exception e) {
					innerException = e;
					TraceHelper.TraceException(traceSource, e);
				}

			}

			// if the innerException is not null, then the related job should be
			// submited, then need terminate the job here.
			if (innerException != null) {
				try {
					client.terminate(startInfo.headnode(), sessionid);
				} catch (Exception e) {
					// if terminate the session failed, do nothing
				}
			}
		} catch (TimeoutException e) {
			throw new TimeoutException(SR.ConnectSessionLauncherTimeout,
					timeoutMilliseconds);
		} finally {
			;// SafeCloseCommunicateObject...
		}

		throw new SessionException(SR.NoBrokerNodeAnswered, innerException);

	}

	// protected virtual void Dispose(Boolean disposing)
	// no scheduler to dispose
	public void dispose() {
		;
	}

	static <T extends SessionBase> SessionBase createSessionInternal(
			SessionStartInfo startInfo, Timeout timeout, Boolean durable)
			throws TimeoutException, SessionException {
		switch (timeout) {
		case Infinite:
			return createSessionInternal(startInfo, Integer.MAX_VALUE, durable);
		default:
			throw new ArgumentException("Invalid timeout: "
					+ timeout.toString());
		}
	}

	/**
	 * Close the session Java Changes: Resource will be released by GC
	 * 
	 * @throws Exception
	 */
	public void close() throws SessionException, TimeoutException {
		close(true);
	}

	/**
	 * Close the session
	 * 
	 * @param purge
	 *            if true, will remove the data in the server
	 * @throws Exception
	 */
	public void close(Boolean purge) throws SessionException, TimeoutException {
		close(purge, Integer.MAX_VALUE);
	}

	/**
	 * Close the session and release the resource
	 * 
	 * @param purge
	 *            if true, we will remove the data from the server
	 * @param timeoutMilliseconds
	 *            timeout for the purge operation
	 * @throws SessionException
	 *             , TimeoutException
	 */
	public void close(boolean purge, int timeoutMilliseconds)
			throws SessionException, TimeoutException {
		IBrokerLauncher broker = null;
		try {
			if (purge) {
				broker = getBrokerLauncher(_info.getBrokerLauncherEpr()
						.getValue());
				setTimeout(broker, getTimeout(new Date(System
						.currentTimeMillis()
						+ timeoutMilliseconds)));
				broker.close(getId());
			}
		} catch (TimeoutException e) {
			throw e;
		} catch (Exception ex) {
			TraceHelper.TraceException(getTraceSource(), ex);
			throw new SessionException(ex);
		}

	}

	// static internal void SaveCredential(SOATraceSource traceSource,
	// SessionStartInfo info)
	// don't really need to save credentials in the cache

	static SessionInfo attachInternal(final SessionAttachInfo attachInfo,
			int timeoutMilliseconds, final SOATraceSource traceSource)
			throws SessionException, TimeoutException {
		username = attachInfo.getUsername();
		password = attachInfo.getPassword();

		Date targetTimeout;

		Utility.throwIfNull(attachInfo, "attachInfo");
		if (attachInfo.getHeadnode() == null || attachInfo.getHeadnode() == "") {
			throw new ArgumentException(SR.HeadnodeCantBeNull);
		}

		if (timeoutMilliseconds < 0) {
			targetTimeout = new Date(System.currentTimeMillis()
					+ Integer.MAX_VALUE);
		} else {
			targetTimeout = new Date(System.currentTimeMillis()
					+ timeoutMilliseconds);
		}

		ISessionLauncher client = getSessionLauncher(attachInfo.getHeadnode());
		SessionInfo info = null;

		try {
			setTimeout(client, getTimeout(targetTimeout));
			info = client.getInfo(attachInfo.getHeadnode(), "https://",
					attachInfo.getId());
		} catch (ISessionLauncherGetInfoSessionFaultFaultFaultMessage e) {
			throw Utility.translateFaultException(e.getFaultInfo(), String
					.format(SR.CannotGetSessionInfo, attachInfo.getId()));
		} catch (TimeoutException e) {
			throw new TimeoutException(SR.ConnectSessionLauncherTimeout,
					timeoutMilliseconds);
		} catch (Exception e) {
			throw new SessionException(e);
		}

		if (info.getBrokerLauncherEpr() == null
				|| info.getBrokerLauncherEpr().getValue() == "") {
			for (String state : info.getJobState()) {
				if (state.equalsIgnoreCase("Configuring")
						|| state.equalsIgnoreCase("ExternalValidation")
						|| state.equalsIgnoreCase("Queued")
						|| state.equalsIgnoreCase("Running")
						|| state.equalsIgnoreCase("Submitted")
						|| state.equalsIgnoreCase("Validating")) {
					throw new SessionException(String.format(
							SR.AttachConfiguringSession, attachInfo.getId()));
				} else {
					throw new SessionException(String.format(
							SR.AttachNoBrokerSession, attachInfo.getId()));
				}
			}
		}

		IBrokerLauncher broker = getBrokerLauncher(info.getBrokerLauncherEpr()
				.getValue());
		try {
			try {
				setTimeout(broker, getTimeout(targetTimeout));
				BrokerInitializationResult brokerLauncherInfo = broker
						.attach(info.getId());
				info.setBrokerEpr(brokerLauncherInfo.getBrokerEpr());
				info.setControllerEpr(brokerLauncherInfo.getControllerEpr());
				info.setResponseEpr(brokerLauncherInfo.getResponseEpr());
			} catch (IBrokerLauncherAttachSessionFaultFaultFaultMessage e) {
				TraceHelper.TraceException(traceSource, e);
				throw Utility.translateFaultException(e.getFaultInfo());
			} catch (TimeoutException e) {
				throw new TimeoutException(SR.ConectBrokerLauncherTimeout,
						timeoutMilliseconds);
			} catch (Exception e) {
				throw new SessionException(e);
			}
		} finally {
			;// safeclose...
		}

		return info;
	}

	static SessionInfo attachInternal(SessionAttachInfo attachInfo,
			Timeout timeout, SOATraceSource traceSource)
			throws SessionException, TimeoutException {
		switch (timeout) {
		case Infinite:
			return attachInternal(attachInfo, Integer.MAX_VALUE, traceSource);
		default:
			throw new ArgumentException(SR.InvalidServiceOperationTimeout);
		}
	}

	/**
	 * Get the time span from the targeted time
	 * 
	 * @param targetTimeout
	 * @return
	 * @throws TimeoutException
	 */
	static long getTimeout(Date targetTimeout) throws TimeoutException {
		Date currentTime = new Date();
		if (targetTimeout.after(currentTime)) {
			return targetTimeout.getTime() - currentTime.getTime();
		} else {
			throw new TimeoutException();
		}
	}

	/**
	 * Verifies if a username is specified. (This function acts differently than
	 * .Net version because there is no way to verify credentials in java)
	 * 
	 * @param info
	 */
	static void checkCredential(SessionStartInfo info) {
		if (info.getUsername() == null || info.getPassword() == null
				|| info.getUsername().getValue() == ""
				|| info.getPassword().getValue() == "") {
			if (HpcJava.user == "") {
				if (!Credentials.promptForCredentials(bConsole, info)) {
					throw new ArgumentException("Must specify a username");
				}
			}
			com.microsoft.hpc.ObjectFactory factory = new com.microsoft.hpc.ObjectFactory();
			info.setUsername(factory
					.createSessionStartInfoContractUsername(HpcJava.user));
		}
	}

	// Java Implementation Functions
	/**
	 * Gets the port for the session Launcher
	 * 
	 * @param headnode
	 *            headnode name
	 * @return SessionLauncher service
	 * @throws SessionException
	 *             if a invalid sessionLauncher addr was given
	 */
	protected static ISessionLauncher getSessionLauncher(String headnode)
			throws SessionException {
		try {
			String sessionLauncherAddr = "https://" + headnode
					+ "/SessionLauncher?wsdl";
			SessionLauncher soaService = new SessionLauncher(new URL(
					sessionLauncherAddr));
			ISessionLauncher sessionLauncher = soaService
					.getBasicHttpEndpointForJava();
			java.util.Map<String, Object> requestContext = ((javax.xml.ws.BindingProvider) sessionLauncher)
					.getRequestContext();
			requestContext.put(Dispatch.ENDPOINT_ADDRESS_PROPERTY,
					sessionLauncherAddr);
			requestContext.put(Dispatch.USERNAME_PROPERTY, HpcJava.user);
			setHttps(sessionLauncher);
			return sessionLauncher;
		} catch (Exception e) {
			throw new SessionException(SR.EndpointAddressNotValid);
		}
	}

	/**
	 * Gets the port for the broker Launcher
	 * 
	 * @param brokerLauncherAddr
	 * @return BrokerLauncher service
	 * @throws SessionException
	 *             if a invalid brokerLaunchera addr was given
	 */
	protected static IBrokerLauncher getBrokerLauncher(String brokerLauncherAddr)
			throws SessionException {
		try {

			brokerLauncherAddr += "?wsdl";

			BrokerLauncher brokerService = new BrokerLauncher(new URL(
					brokerLauncherAddr));
			IBrokerLauncher brokerLauncher = brokerService
					.getBasicHttpEndpointForJava();

			java.util.Map<String, Object> requestContext = ((javax.xml.ws.BindingProvider) brokerLauncher)
					.getRequestContext();

			requestContext.put(Dispatch.ENDPOINT_ADDRESS_PROPERTY,
					brokerLauncherAddr);

			setHttps(brokerLauncher);
			return brokerLauncher;
		} catch (Exception e) {
			throw new SessionException(SR.EndpointAddressNotValid);
		}

	}

	/**
	 * Sets the authentication protocol
	 * 
	 * @param serviceObject
	 *            object to set https on
	 * @throws SessionException
	 *             if an error occurred trying to get session
	 */
	protected static void setHttps(Object serviceObject)
			throws SessionException {
		try {
			org.apache.cxf.endpoint.Client client = org.apache.cxf.frontend.ClientProxy
					.getClient(serviceObject);
			org.apache.cxf.endpoint.Endpoint cxfEndpoint = client.getEndpoint();

			Map<String, Object> outProps = new HashMap<String, Object>();
			outProps.put(WSHandlerConstants.ACTION,
					WSHandlerConstants.USERNAME_TOKEN);
			outProps.put(WSHandlerConstants.USER, username);
			outProps.put(WSHandlerConstants.PASSWORD_TYPE, WSConstants.PW_TEXT);
			outProps.put(WSHandlerConstants.PW_CALLBACK_CLASS,
					PWHandlerClass.class.getName());

			WSS4JOutInterceptor wssOut = new WSS4JOutInterceptor(outProps);
			cxfEndpoint.getOutInterceptors().add(wssOut);

		} catch (Exception e) {
			throw new SessionException(SR.EndpointAddressNotValid);
		}
	}

	/**
	 * Sets the HTTP client proxy timeout
	 * 
	 * @param serviceObject
	 * @param timeout
	 */
	private static void setTimeout(Object serviceObject, long timeout) {
		Client client = ClientProxy.getClient(serviceObject);

		HTTPConduit http = (HTTPConduit) client.getConduit();
		HTTPClientPolicy httpClientPolicy = http.getClient();

		httpClientPolicy.setConnectionTimeout(timeout);
		httpClientPolicy.setReceiveTimeout(timeout);
		http.setClient(httpClientPolicy);
	}

	/**
	 * Class to handle password for certificate
	 * 
	 */
	public static class PWHandlerClass implements CallbackHandler {

		public void handle(Callback[] callbacks) throws IOException,
				UnsupportedCallbackException {

			for (int i = 0; i < callbacks.length; i++) {
				WSPasswordCallback pwcb = (WSPasswordCallback) callbacks[i];
				// String id = pwcb.getIdentifier();

				// if (JavaClientTest.WSSUsername.equals(id)) {
				pwcb.setPassword(password);
				// } else {
				// pwcb.setPassword("User not recognized within PWHandlerClass.");
				// }
			}
		}
	}

	/**
	 * Class to build SessionInfo
	 * 
	 */
	private static SessionInfo BuildSessionInfo(
			BrokerInitializationResult result, Boolean durable, Boolean secure,
			int id, String brokerLauncherEpr, TransportScheme scheme) {
		com.microsoft.hpc.sessionlauncher.ObjectFactory of = new com.microsoft.hpc.sessionlauncher.ObjectFactory();
		SessionInfo info = new SessionInfo();
		info.setId(id);
		info.setDurable(durable);
		info.setBrokerEpr(result.getBrokerEpr());
		info.setBrokerLauncherEpr(of
				.createSessionInfoBrokerLauncherEpr(brokerLauncherEpr));
		info.setControllerEpr(result.getControllerEpr());
		info.setResponseEpr(result.getResponseEpr());
		info.setSecure(secure);
		Version serverVersion = new Version();
		serverVersion.setMajor(3);
		serverVersion.setMinor(0);
		info.setServerVersion(of.createSessionInfoServerVersion(serverVersion));
		info.getTransportScheme().add(scheme.getName());
		return info;
	}

}
