/* 
-*- coding: latin-1 -*-

This file is part of RefactorErl.

RefactorErl is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published
by the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

RefactorErl is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with RefactorErl.  If not, see <http://plc.inf.elte.hu/erlang/>.

The Original Code is RefactorErl.

The Initial Developer of the Original Code is Eötvös Loránd University.
Portions created  by Eötvös Loránd University and ELTE-Soft Ltd.
are Copyright 2007-2025 Eötvös Loránd University, ELTE-Soft Ltd.
and Ericsson Hungary. All Rights Reserved.
*/

package com.refactorerl.ui.communication;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.logging.Handler;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpErlangAtom;
import com.ericsson.otp.erlang.OtpErlangBoolean;
import com.ericsson.otp.erlang.OtpErlangDecodeException;
import com.ericsson.otp.erlang.OtpErlangExit;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangPid;
import com.ericsson.otp.erlang.OtpErlangTuple;
import com.ericsson.otp.erlang.OtpMbox;
import com.ericsson.otp.erlang.OtpNode;
import com.ericsson.otp.erlang.OtpPeer;
import com.ericsson.otp.erlang.OtpSelf;
import com.refactorerl.ui.common.OtpErlangHelper;
import com.refactorerl.ui.communication.event.IEvent;
import com.refactorerl.ui.communication.event.IEventDispatcher;
import com.refactorerl.ui.communication.event.IEventHandler;
import com.refactorerl.ui.communication.exceptions.CommunicationException;
import com.refactorerl.ui.communication.exceptions.ConnectionException;
import com.refactorerl.ui.communication.exceptions.DeniedRequestException;

// NOTE fix for possible JInterface non timeout sendRPC bug: 
// if it returns null,the message needs to be sent again

// NOTE sendRPC() is in race condition with requestReqId()

/**
 * ReferlProxy is the central class of the communication layer. It represents a
 * server instance of an Erlang node running RefactorErl. This class contains
 * methods to connect to, to disconnect from and to communicate with a
 * RefactorErl instance.
 * 
 * As a reminder, the address of an Erlang node can be constructed according to
 * the following rule: address ::= alive_name @ host_name
 * 
 * @author Daniel Lukacs, 2014 ELTE IK
 *
 */
public class ReferlProxy {

	// NOTE send and sendRPC of JInterface are blocking calls

	private volatile OtpConnection connection;
	private OtpMbox mbox;

	private String aliveName;
	private String serverAddress;
	private final String cookie;

	/**
	 * Upper bound for successive unsuccessful connection attempts.
	 */
	private final int reconnectAttempts = 5;
	/**
	 * The time to wait between between an unsuccessful connection attempt and a
	 * next attempt.
	 */
	private final int reconnectDelay = 1000;
	/**
	 * The time to wait for a closing connection. After expiration the
	 * connection will be left in its half-closed state.
	 */
	private final int terminationDelay = 1000;

	/**
	 * For internal technical reasons, the message loops starts a new message
	 * receiving loop if no message arrives after that time.
	 */
	private final int messageLoopTimeout = 500;
	private String mboxClientSuffix = "mboxc"; //$NON-NLS-1$
	private String mboxSuffix = "mbox"; //$NON-NLS-1$

//	private Handler[] logHandlers;

	private Logger logger;

	private boolean asyncMessengerOn;

	private IEventDispatcher edp;

	// one for message loop, one for posting
	private ExecutorService threadPool;

	/**
	 * 
	 * @param edp
	 *            An already started event dispatcher. Used internally and
	 *            change events will also fired on this event dispatcher.
	 * @param aliveName
	 *            An alive name for the Erlang node created when the start()
	 *            method is called
	 * @param serverAddress
	 *            The server address of the RefactorErl node to connect to when
	 *            this proxy is started
	 * @param logHandlers
	 */
	public ReferlProxy(IEventDispatcher edp, String aliveName, String cookie,
			String serverAddress, Handler[] logHandlers) {
		this.edp = edp;

		this.aliveName = aliveName;
		this.serverAddress = serverAddress;
		this.cookie = cookie;
	//	this.logHandlers = logHandlers;
		configureLogger(logHandlers);
	}

	/**
	 * Same as send(OtpErlangObject request, IReferlProgressMonitor m, Integer
	 * timeout), except the blocking method in this method won't timeout if no
	 * reply arrives.
	 * 
	 * @param request
	 * @param m
	 * @return
	 * @throws ConnectionException
	 * @throws CommunicationException
	 */
	public OtpErlangObject send(OtpErlangObject request,
			IReferlProgressMonitor m) throws ConnectionException,
			CommunicationException {
		return send(request, m, null);

	}

	/**
	 * Method to do synchronous communication with RefactorErl. As opposed to
	 * sendRPC(String, String, OtpErlangObject), this method can be called and
	 * executed concurrently.
	 * 
	 * This method will listen for progress events related to the request and
	 * refreshes the supplied IReferlProgressMonitor accordingly. If no reply
	 * arrives in the supplied timeout frame, the method will stop waiting and
	 * returns null.
	 * 
	 * @param request
	 *            An OtpErlangTuple containing the OtpErlangAtom name of the
	 *            requested function and the corresponding parameters.
	 * @param m
	 *            An IReferlProgressMonitor which will be refreshed upon the
	 *            arrival of progress events.
	 * @param timeout
	 *            The time frame after the method will stop waiting for reply.
	 * @return The full reply message or null if timed out.
	 * @throws ConnectionException
	 *             Thrown if there was an error in the connection.
	 * @throws CommunicationException
	 *             Thrown if null or ill-formed request was sent or the request
	 *             violated the communication protocol set by RefactorErl.
	 */
	public OtpErlangObject send(final OtpErlangObject request,
			IReferlProgressMonitor m, Integer timeout)
			throws ConnectionException, CommunicationException {

		// register listener for progress - update monitor
		// register listener for reply - release method lock
		// post o
		// lock until reply
		// return result

		final CountDownLatch latch;
		final IReferlProgressMonitor progressMonitor;

		class Wrapper {
			OtpErlangObject result = null;
			int percentCompleted = 0; // normalised to 100

			public void setResult(OtpErlangObject result) {
				this.result = result;
			}

			public void setPercentCompleted(int percentCompleted) {
				this.percentCompleted = percentCompleted;
			}

		}
		;
		final Wrapper wrapper = new Wrapper();

		progressMonitor = m;

		latch = new CountDownLatch(1);

		OtpErlangObject reqId = requestReqId();

		String topic = MessageTopicHelper.getReplyTopic(reqId);

		edp.subscribe(topic, new IEventHandler() {

			@Override
			public void handleEvent(IEvent event) {
				Object o = edp.getValue(event);
				if (!(o instanceof OtpErlangObject))
					return;
				OtpErlangObject msg = (OtpErlangObject) o;
				String msgType = MessageParseHelper.getType(msg);

				if (msgType.equals("reply")) { //$NON-NLS-1$
					edp.unsubscribe(this);

					wrapper.setResult(msg);

					latch.countDown(); // control goes back to send()

				} else if (msgType.equals("progress")) { //$NON-NLS-1$
					if (progressMonitor != null) {

						try {
							if (MessageParseHelper
									.parseIsProgressCompleted(msg)) {
								progressMonitor.done();
								progressMonitor.beginTask(OtpErlangHelper
										.toUnquotedString(request), 100);
							}
							int actualPercent = (int) Math.round(MessageParseHelper
									.parseProgressTaskPercent(msg) * 100.0);
							int lastPercent = actualPercent
									- wrapper.percentCompleted;

							progressMonitor.worked(lastPercent,
									MessageParseHelper
											.parseProgressTaskInfo(msg));

							wrapper.setPercentCompleted(actualPercent);

						} catch (IllegalArgumentException e) { // unfinished
							getLogger().warning(Messages.ReferlProxy_4 + msg);
						}
					}
				} else if (msgType.equals("question")) { //$NON-NLS-1$
					// unfinished
				}

			}
		});

		if (progressMonitor != null)
			progressMonitor.beginTask(
					OtpErlangHelper.toUnquotedString(request), 100);

		postWithReqId(request, reqId);
		try {

			if (timeout == null)
				latch.await();
			else
				latch.await(timeout, TimeUnit.MILLISECONDS);

			// result will be set in handleEvent(), before it calls
			// latch.countDown()

		} catch (InterruptedException e) {
			wrapper.setResult(null);
		} finally {
			latch.countDown(); // not really needed
		}

		if (progressMonitor != null) {
			progressMonitor.done();
		}

		return wrapper.result;
	}

	// NOTE untested
	public FutureTask<?> post(final OtpErlangObject request) { // async
		// (non-modifiers)

		FutureTask<?> f = new FutureTask<Void>(new Callable<Void>() {

			@Override
			public Void call() throws Exception {

				OtpErlangObject reqId;
				reqId = requestReqId();
				postWithReqId(request, reqId);
				return null;
			}
		});

		threadPool.execute(f);
		return f;

	}

	/**
	 * Internal method embodying the actual communication phase with
	 * RefactorErl.
	 * 
	 * @param request
	 * @param reqId
	 * @throws CommunicationException
	 * @throws ConnectionException
	 */
	private void postWithReqId(OtpErlangObject request, OtpErlangObject reqId)
			throws CommunicationException, ConnectionException {
		try {
			OtpErlangTuple nodeInfo = new OtpErlangTuple(new OtpErlangObject[] {
					mbox.self(), new OtpErlangAtom(mbox.getName()) });

			getLogger().info(
					Messages.ReferlProxy_0 + nodeInfo
							+ "|" + reqId + "|" + request); //$NON-NLS-2$ //$NON-NLS-3$

			// > rpc:call('my_server@my_host', reflib_ui_router, request,
			// [NodeInfo, ReqId, Request]).

			OtpErlangObject msg = sendRPC(
					"reflib_ui_router", "request", new OtpErlangObject[] { nodeInfo, reqId, request }); //$NON-NLS-1$ //$NON-NLS-2$

			if (!msg.equals(new OtpErlangAtom("ok"))) { //$NON-NLS-1$
				if (msg.equals(new OtpErlangAtom("deny"))) { //$NON-NLS-1$
					getLogger().info(Messages.ReferlProxy_13);
					throw new DeniedRequestException(Messages.ReferlProxy_14
							+ Messages.ReferlProxy_15);
				} else {

					getLogger().severe(
							Messages.ReferlProxy_16 + request
									+ Messages.ReferlProxy_17 + msg.toString());

					throw new CommunicationException(Messages.ReferlProxy_18
							+ request + Messages.ReferlProxy_19
							+ msg.toString());
				}
			}

		} catch (NullPointerException e) {
			throw new CommunicationException("RPC write error.");
		}

	}

	/**
	 * Directly calls the supplied RefactorErl function circumventing the
	 * RefactorErl communication protocol. As opposed to send(OtpErlangObject,
	 * IReferlProgressMonitor, Integer), this method is synchronized, so only
	 * one thread will be able to execute it at a time. All other calling threads
	 * will be temporarily blocked, waiting for their turn in the queue. 
	 * 
	 * @param module
	 *            Containing module of the RefactorErl function to call
	 * @param function
	 *            The name of the RefactorErl function to call
	 * @param args
	 *            The arguments of the RefactorErl function to call
	 * @return A return value of the called function.
	 * @throws ConnectionException
	 *             Thrown if there was an error in the connection.
	 * @throws CommunicationException
	 *             Thrown if the method was unable to call the supplied
	 *             function.
	 */
	public synchronized OtpErlangObject sendRPC(String module, String function,
			OtpErlangObject... args) throws CommunicationException,
			ConnectionException {
		// rpc:call('my_server@my_host', code, is_loaded, [Module]).

		// if previous call returned false
		// rpc:call('my_server@my_host', code, load_file, [Module]).

		// either way
		// rpc:call('my_server@my_host', Module, Function, ArgsList).

		try {
			List<String> argsStrings = new ArrayList<>();
			for (OtpErlangObject o : args) {
				argsStrings.add(o.toString());
			}

			getLogger().info(
					Messages.ReferlProxy_21 + module
							+ "," + function + "," + argsStrings); //$NON-NLS-2$ //$NON-NLS-3$

			connection
					.sendRPC(
							"code", "is_loaded", new OtpErlangObject[] { new OtpErlangAtom(module) }); //$NON-NLS-1$ //$NON-NLS-2$

			if (connection.receiveRPC().equals(new OtpErlangBoolean(false))) {
				getLogger().info(
						Messages.ReferlProxy_1 + module
								+ Messages.ReferlProxy_27);
				connection
						.sendRPC(
								"code", "load_file", new OtpErlangObject[] { new OtpErlangAtom(module) }); //$NON-NLS-1$ //$NON-NLS-2$
				OtpErlangTuple loadReply = (OtpErlangTuple) connection
						.receiveRPC();
				if (!loadReply.elementAt(0).equals(new OtpErlangAtom("module"))) { //$NON-NLS-1$
					getLogger().info(
							Messages.ReferlProxy_31 + module
									+ Messages.ReferlProxy_32
									+ loadReply.toString());
					throw new CommunicationException(loadReply.toString());
				}
				getLogger().info(
						Messages.ReferlProxy_33 + module
								+ Messages.ReferlProxy_34);
			}

			connection.sendRPC(module, function, args);
			OtpErlangObject reply = connection.receiveRPC();
			getLogger().info(reply.toString());
			return reply;

		} catch (IOException | OtpErlangExit | OtpAuthException e) {
			throw new ConnectionException(Messages.ReferlProxy_35
					+ e.getMessage());
		}

	}

	/**
	 * Initializes this ReferlProxy instance. Calling this method before any
	 * other non-static method of this class is mandatory. This method will try
	 * connect to RefactorErl and if it fails, retry for a fixed number of
	 * attempts. After successfully connecting, a message handling loop will be
	 * started in a new thread which will handle reply and progress messages
	 * arriving from RefactorErl.
	 * 
	 * @return True, if the message handler was registered to RefactorErl and
	 *         the message handling thread was started.
	 */

	public boolean start() {

		threadPool = Executors.newFixedThreadPool(2);

		OtpSelf clientProxy = null;
		try {
			if(cookie != null && !cookie.isEmpty()){
				clientProxy = new OtpSelf(aliveName, cookie);
			} else { 
				clientProxy = new OtpSelf(aliveName);
			}
		} catch (IOException e1) {
			getLogger().severe(Messages.ReferlProxy_36);
			return false;
		}
		
		OtpPeer serverProxy = new OtpPeer(serverAddress);
		connection = forceConnect(clientProxy, serverProxy);

		if (connection == null) {
			getLogger().severe(Messages.ReferlProxy_37);
			return false;
		}

		try {
			// NOTE Eclipse forbids blocking calls in Activator.start()
			// JInterface send methods are all blocking calls
			initMessageLoop();
		} catch (ConnectionException | CommunicationException e) {
			getLogger().severe(Messages.ReferlProxy_38);
			return false;
		}

		threadPool.execute(new Runnable() {
			@Override
			public void run() {
				runMessageLoop();
			}
		});

		try {
			// NOTE race condition is unlikely, just making sure
			Thread.sleep(250);
		} catch (InterruptedException e) {
		}

		return true;
	}

	/**
	 * Frees up the resources of this ReferlProxy instance and unregisters the
	 * registered message handler in the RefactorErl server instance.
	 */
	public void stop() {

		try {
			closeMessageLoop();
		} catch (ConnectionException e) {
			getLogger().severe(Messages.ReferlProxy_39);
			return;
		}

		threadPool.shutdown();
		try {
			threadPool
					.awaitTermination(terminationDelay, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e1) {

		}
		if (!threadPool.isTerminated())
			threadPool.shutdownNow();

		connection.close();
		getLogger().info(Messages.ReferlProxy_40);

		deconfigureLogger();
	}

	/**
	 * Creates a new Erlang message box to exclusively listen to the messages
	 * arriving from the RefactorErl server, and registers it to the RefactorErl
	 * server.
	 * 
	 * @throws ConnectionException
	 *             Thrown if there was an error in the connection when
	 *             registering.
	 * @throws CommunicationException
	 */
	private void initMessageLoop() throws ConnectionException,
			CommunicationException {
		try {
			// NOTE client names must be unique. any client started with the
			// same name must be closed before this, or mbox.receive() will hang
			if(cookie != null && !cookie.isEmpty()){
				client = new OtpNode(aliveName + mboxClientSuffix, cookie);
			} else {
				client = new OtpNode(aliveName + mboxClientSuffix);
			}
			mbox = client.createMbox(aliveName + mboxSuffix);
		} catch (IOException e) {
			throw new ConnectionException(Messages.ReferlProxy_41);
		}

		try {

			OtpErlangObject reply = sendRPC(
					"reflib_ui_router", "add_msg_handler", new OtpErlangObject[] { mbox.self() }); //$NON-NLS-1$ //$NON-NLS-2$

			if (!reply.equals(new OtpErlangAtom("ok"))) { //$NON-NLS-1$
				getLogger().severe(Messages.ReferlProxy_45 + reply.toString());
				throw new ConnectionException(Messages.ReferlProxy_46
						+ reply.toString());
			}

			reply = mbox.receive();

			if (!reply.equals(new OtpErlangAtom("installed"))) { //$NON-NLS-1$
				getLogger().severe(Messages.ReferlProxy_48 + reply.toString());
				throw new ConnectionException(Messages.ReferlProxy_49
						+ reply.toString());

			}
		} catch (OtpErlangExit | OtpErlangDecodeException e) {
			throw new ConnectionException(Messages.ReferlProxy_50);
		}

		asyncMessengerOn = true;
	}

	/**
	 * Unregisters the registered message handler in the RefactorErl server
	 * instance.
	 * 
	 * @throws ConnectionException
	 *             Thrown if there was an error in the connection when
	 *             unregistering.
	 */
	private void closeMessageLoop() throws ConnectionException {
		mbox.close();
		client.close();
		asyncMessengerOn = false;

		OtpErlangObject reply = null;
		try {

			reply = sendRPC(
					"reflib_ui_router", "del_msg_handler", new OtpErlangObject[] { mbox.self() }); //$NON-NLS-1$ //$NON-NLS-2$

			if (!reply.equals(new OtpErlangAtom("ok"))) //$NON-NLS-1$
				getLogger().warning(Messages.ReferlProxy_54);

		} catch (CommunicationException e) {
			getLogger().severe(Messages.ReferlProxy_55);
			throw new ConnectionException(Messages.ReferlProxy_56);
		}

		getLogger().info(Messages.ReferlProxy_57);
	}

	/**
	 * A message handling loop which will handle reply and progress messages
	 * arriving from RefactorErl by firing events on the appropriate topics on
	 * the EventDispatcher instance supplied in this classes constructor.
	 */
	private void runMessageLoop() {
		getLogger().info(Messages.ReferlProxy_58);

		while (asyncMessengerOn) {
			OtpErlangObject msg = null;

			try {

				// msg = mbox.receive(); //cant be interrupted by
				// closing (bug?)
				msg = mbox.receive(messageLoopTimeout);
				if (msg == null) // timeout
					continue;

			} catch (OtpErlangExit e) {
				getLogger().severe(Messages.ReferlProxy_59);
				throw new RuntimeException(Messages.ReferlProxy_60); // couldn't
																		// throw
				// ConnectionException
				// from inside
				// thread
			} catch (OtpErlangDecodeException e) {
				getLogger().warning(Messages.ReferlProxy_61);
				continue;
			}

			getLogger().info(Messages.ReferlProxy_62 + msg.toString());

			edp.fire(MessageTopicHelper.parseTopic(msg), msg);

		}

	}

	/**
	 * Acquires a request ID from the RefactorErl server, which is necessary to
	 * send requests according to the communication protocol set by RefactorErl.
	 * 
	 * @return A request ID tuple.
	 * @throws CommunicationException
	 * @throws ConnectionException
	 */
	private OtpErlangObject requestReqId() throws CommunicationException,
			ConnectionException {

		// > rpc:call('my_server@my_host', reflib_ui_router, getid, []).
		try {
			OtpErlangTuple reqId = (OtpErlangTuple) sendRPC(
					"reflib_ui_router", "getid", new OtpErlangObject[0]); //$NON-NLS-1$ //$NON-NLS-2$

			if (reqId == null) {
				getLogger().severe(Messages.ReferlProxy_65);

				throw new CommunicationException(Messages.ReferlProxy_66);
			}
			return reqId;
		} catch (IllegalArgumentException | ClassCastException e) {
			getLogger().severe(Messages.ReferlProxy_67);

			throw new CommunicationException(Messages.ReferlProxy_68);
		}

	}

	private OtpConnection forceConnect(OtpSelf clientProxy, OtpPeer serverProxy) {

		int attempts = 1;
		while (attempts < reconnectAttempts) {

			try {

				connection = clientProxy.connect(serverProxy);

				break;

			} catch (IOException | OtpAuthException e) {
				getLogger().warning(
						Messages.ReferlProxy_70 + (attempts++)
								+ Messages.ReferlProxy_71);

				try {
					Thread.sleep(reconnectDelay);
				} catch (InterruptedException ex) {
					break;
				}

			}
		}

		return connection;
	}

	private void configureLogger(Handler[] handlers) {
		if (handlers != null && logger == null) {
			logger = Logger.getLogger(getClass().getName());
			logger.setLevel(Level.ALL);
			logger.setUseParentHandlers(false);

			for (Handler h : handlers) {
				logger.addHandler(h);
			}

		}
	}

	private void deconfigureLogger() {
		if (logger != null) {
			for (Handler h : logger.getHandlers()) {
				logger.removeHandler(h);
			}
		}
	}
	
	public void resetLogHandlers(Handler[] handlers){
		deconfigureLogger();
		for (Handler h : handlers) {
			logger.addHandler(h);
		}
	}

	public String getNodeName() {
		return connection.self().node();
	}

	public Logger getLogger() {
		return logger;
	}

	private Random r = new Random();
	private OtpNode client;

	public OtpErlangPid generatePid() {

		// return connection.self().createPid(); // not unique between sessions
		return new OtpErlangPid(getNodeName(), r.nextInt(), r.nextInt(),
				r.nextInt());
	}

}
