/**
Copyright (c) 2007-2013 Alysson Bessani, Eduardo Alchieri, Paulo Sousa, and the authors indicated in the @author tags

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */
package bftsmart.tom;

import java.util.Arrays;
import java.util.Comparator;
import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import bftsmart.reconfiguration.ReconfigureReply;
import bftsmart.reconfiguration.views.View;
import bftsmart.tom.core.messages.TOMMessage;
import bftsmart.tom.core.messages.TOMMessageType;
import bftsmart.tom.util.Extractor;
//import bftsmart.tom.util.Logger;
import bftsmart.tom.util.TOMUtil;
import org.apache.log4j.Logger;

/**
 * This class implements a TOMSender and represents a proxy to be used on the
 * client side of the replicated system.
 * It sends a request to the replicas, receives the reply, and delivers it to
 * the application.
 */
public class ServiceProxy extends TOMSender {
	private final static Logger logger = Logger.getLogger(ServiceProxy.class);
	// Locks for send requests and receive replies
	protected ReentrantLock canReceiveLock = new ReentrantLock();
	protected ReentrantLock canSendLock = new ReentrantLock();
	private Semaphore sm = new Semaphore(0);
	private int reqId = -1; // request id
	private int operationId = -1; // request id
	private TOMMessageType requestType;
	private int replyQuorum = 0; // size of the reply quorum
	private TOMMessage replies[] = null; // Replies from replicas are stored here
	private int receivedReplies = 0; // Number of received replies
	private TOMMessage response = null; // Reply delivered to the application
	private int invokeTimeout = 40;
	private Comparator<byte[]> comparator;
	private Extractor extractor;
	private Random rand = new Random(System.currentTimeMillis());
	private int replyServer;
	private HashResponseController hashResponseController;
	private int invokeUnorderedHashedTimeout = 10;

	/**
	 * Constructor
	 *
	 * @see @bellow
	 */
	public ServiceProxy(int processId) {
		this(processId, "config", null, null);
	}

	/**
	 * Constructor
	 *
	 * @see @bellow
	 */
	public ServiceProxy(int processId, String configHome) {
		this(processId, configHome, null, null);
	}

	/**
	 * Constructor
	 *
	 * @param processId Process id for this client (should be different from replicas)
	 * @param configHome Configuration directory for BFT-SMART
	 * @param replyComparator used for comparing replies from different servers
	 *                        to extract one returned by f+1
	 * @param replyExtractor used for extracting the response from the matching
	 *                       quorum of replies
	 */
	public ServiceProxy(int processId, String configHome,
			Comparator<byte[]> replyComparator, Extractor replyExtractor) {
		if (configHome == null) {
			init(processId);
		} else {
			init(processId, configHome);
		}

		replies = new TOMMessage[getViewManager().getCurrentViewN()];

		comparator = (replyComparator != null) ? replyComparator : new Comparator<byte[]>() {
			@Override
			public int compare(byte[] o1, byte[] o2) {
				return Arrays.equals(o1, o2) ? 0 : -1;
			}
		};

		extractor = (replyExtractor != null) ? replyExtractor : new Extractor() {

			@Override
			public TOMMessage extractResponse(TOMMessage[] replies, int sameContent, int lastReceived) {
				return replies[lastReceived];
			}
		};
	}

	/**
	 * Get the amount of time (in seconds) that this proxy will wait for
	 * servers replies before returning null.
	 *
	 * @return the invokeTimeout
	 */
	public int getInvokeTimeout() {
		return invokeTimeout;
	}

	public int getInvokeUnorderedHashedTimeout() {
		return invokeUnorderedHashedTimeout;
	}

	/**
	 * Set the amount of time (in seconds) that this proxy will wait for
	 * servers replies before returning null.
	 *
	 * @param invokeTimeout the invokeTimeout to set
	 */
	public void setInvokeTimeout(int invokeTimeout) {
		this.invokeTimeout = invokeTimeout;
	}

	public void setInvokeUnorderedHashedTimeout(int timeout) {
		this.invokeUnorderedHashedTimeout = timeout;
	}

	public byte[] invokeOrdered(byte[] request) {
		return invoke(request, TOMMessageType.ORDERED_REQUEST);
	}

	public byte[] invokeUnordered(byte[] request) {
		return invoke(request, TOMMessageType.UNORDERED_REQUEST);
	}

	public byte[] invokeUnorderedHashed(byte[] request) {
		return invoke(request, TOMMessageType.UNORDERED_HASHED_REQUEST);
	}

	/**
	 * This method sends a request to the replicas, and returns the related reply.
	 * If the servers take more than invokeTimeout seconds the method returns null.
	 * This method is thread-safe.
	 *
	 * @param request Request to be sent
	 * @param reqType TOM_NORMAL_REQUESTS for service requests, and other for
	 *        reconfig requests.
	 * @return The reply from the replicas related to request
	 */
	public byte[] invoke(byte[] request, TOMMessageType reqType) {
		canSendLock.lock();

		// Clean all statefull data to prepare for receiving next replies
		Arrays.fill(replies, null);
		receivedReplies = 0;
		response = null;
		replyQuorum = getReplyQuorum();

		// Send the request to the replicas, and get its ID
		reqId = generateRequestId(reqType);
		operationId = generateOperationId();
		requestType = reqType;

		replyServer = -1;
		hashResponseController = null;

		if(requestType == TOMMessageType.UNORDERED_HASHED_REQUEST){

			replyServer = getRandomlyServerId();
			logger.info("["+this.getClass().getName()+"] replyServerId("+replyServer+") "
					+ "pos("+getViewManager().getCurrentViewPos(replyServer)+")");

			hashResponseController = new HashResponseController(getViewManager().getCurrentViewPos(replyServer),
					getViewManager().getCurrentViewProcesses().length);

			TOMMessage sm = new TOMMessage(getProcessId(),getSession(), reqId, operationId, request,
					getViewManager().getCurrentViewId(), requestType);
			sm.setReplyServer(replyServer);

			TOMulticast(sm);
		}else{
			TOMulticast(request, reqId, operationId, reqType);
		}

		logger.info("Sending request (" + reqType + ") with reqId=" + reqId);
		logger.info("Expected number of matching replies: " + replyQuorum);

		// This instruction blocks the thread, until a response is obtained.
		// The thread will be unblocked when the method replyReceived is invoked
		// by the client side communication system
		try {
			if(reqType == TOMMessageType.UNORDERED_HASHED_REQUEST){
				if (!this.sm.tryAcquire(invokeUnorderedHashedTimeout, TimeUnit.SECONDS)) {
					logger.info("######## UNORDERED HASHED REQUEST TIMOUT ########");
					canSendLock.unlock();
					return invoke(request,TOMMessageType.ORDERED_REQUEST);
				}
			}else{ 
				if (!this.sm.tryAcquire(invokeTimeout, TimeUnit.SECONDS)) {
					logger.info("###################TIMEOUT#######################");
					logger.info("Reply timeout for reqId=" + reqId);
					logger.info(getProcessId() + " // " + reqId + " // TIMEOUT // ");
					logger.info("Replies received: " + receivedReplies);
					canSendLock.unlock();

					return null;
				}
			}
		} catch (InterruptedException ex) {
			logger.error("", ex);
		}

		logger.info("Response extracted = " + response);

		byte[] ret = null;

		if (response == null) {
			//the response can be null if n-f replies are received but there isn't
			//a replyQuorum of matching replies
			logger.info("Received n-f replies and no response could be extracted.");

			canSendLock.unlock();
			if (reqType == TOMMessageType.UNORDERED_REQUEST || reqType == TOMMessageType.UNORDERED_HASHED_REQUEST) {
				//invoke the operation again, whitout the read-only flag
				logger.info("###################RETRY#######################");
				return invokeOrdered(request);
			} else {
				throw new RuntimeException("Received n-f replies without f+1 of them matching.");
			}
		} else {
			//normal operation
			//******* EDUARDO BEGIN **************//
			if (reqType == TOMMessageType.ORDERED_REQUEST) {
				//Reply to a normal request!
				if (response.getViewID() == getViewManager().getCurrentViewId()) {
					ret = response.getContent(); // return the response
				} else {//if(response.getViewID() > getViewManager().getCurrentViewId())
					//updated view received
					reconfigureTo((View) TOMUtil.getObject(response.getContent()));

					canSendLock.unlock();
					return invoke(request, reqType);
				}
			} else if (reqType == TOMMessageType.UNORDERED_REQUEST || reqType == TOMMessageType.UNORDERED_HASHED_REQUEST){
				if (response.getViewID() == getViewManager().getCurrentViewId()) {
					ret = response.getContent(); // return the response
				}else{
					canSendLock.unlock();
					return invoke(request,TOMMessageType.ORDERED_REQUEST);
				}
			} else {
				if (response.getViewID() > getViewManager().getCurrentViewId()) {
					//Reply to a reconfigure request!
					logger.info("Reconfiguration request' reply received!");
					Object r = TOMUtil.getObject(response.getContent());
					if (r instanceof View) { //did not executed the request because it is using an outdated view
						reconfigureTo((View) r);

						canSendLock.unlock();
						return invoke(request, reqType);
					}  else if (r instanceof ReconfigureReply) { //reconfiguration executed!
						reconfigureTo(((ReconfigureReply) r).getView());
						ret = response.getContent();
					} else{
						logger.info("Unknown response type");
					}
				} else {
					logger.info("Unexpected execution flow");
				}
			}
		}
		//******* EDUARDO END **************//

		canSendLock.unlock();
		return ret;
	}

	//******* EDUARDO BEGIN **************//
	protected void reconfigureTo(View v) {
		logger.info("Installing a most up-to-date view with id=" + v.getId());
		getViewManager().reconfigureTo(v);
		getViewManager().getViewStore().storeView(v);
		replies = new TOMMessage[getViewManager().getCurrentViewN()];
		getCommunicationSystem().updateConnections();
	}
	//******* EDUARDO END **************//

	/**
	 * This is the method invoked by the client side communication system.
	 *
	 * @param reply The reply delivered by the client side communication system
	 */
	@Override
	public void replyReceived(TOMMessage reply) {
            logger.info("Synchronously received reply from " + reply.getSender() + " with sequence number " + reply.getSequence());

                try {
			canReceiveLock.lock();
			if (reqId == -1) {//no message being expected
				logger.info("throwing out request: sender=" + reply.getSender() + " reqId=" + reply.getSequence());
				canReceiveLock.unlock();
				return;
			}

			int pos = getViewManager().getCurrentViewPos(reply.getSender());

			if (pos < 0) { //ignore messages that don't come from replicas
				canReceiveLock.unlock();
				return;
			}

			int sameContent = 1;
			if (reply.getSequence() == reqId && reply.getReqType() == requestType) {

				logger.info("Receiving reply from " + reply.getSender()
						+ " with reqId:" + reply.getSequence() + ". Putting on pos=" + pos);

				if(requestType == TOMMessageType.UNORDERED_HASHED_REQUEST)
				{
					response = hashResponseController.getResponse(pos,reply);
					if(response !=null){
						reqId = -1;
						this.sm.release(); // resumes the thread that is executing the "invoke" method
						canReceiveLock.unlock();
						return;
					}

				}else{
					if (replies[pos] == null) {
						receivedReplies++;
					}
					replies[pos] = reply;

					// Compare the reply just received, to the others
					
					for (int i = 0; i < replies.length; i++) {

						if ((i != pos || getViewManager().getCurrentViewN() == 1) && replies[i] != null
								&& (comparator.compare(replies[i].getContent(), reply.getContent()) == 0)) {
							sameContent++;
							if (sameContent >= replyQuorum) {
								response = extractor.extractResponse(replies, sameContent, pos);
								reqId = -1;
								this.sm.release(); // resumes the thread that is executing the "invoke" method
								canReceiveLock.unlock();
								return;
							}
						}
					}
				}
				
				if (response == null) {
					if (requestType.equals(TOMMessageType.ORDERED_REQUEST)) {
						if (receivedReplies == getViewManager().getCurrentViewN()) {
							reqId = -1;
							this.sm.release(); // resumes the thread that is executing the "invoke" method
						}
					}else if (requestType.equals(TOMMessageType.UNORDERED_HASHED_REQUEST)) {
						if (hashResponseController.getNumberReplies() == getViewManager().getCurrentViewN()) {
							reqId = -1;
							this.sm.release(); // resumes the thread that is executing the "invoke" method
						}
					} else {  // UNORDERED
						if (receivedReplies != sameContent) {
							reqId = -1;
							this.sm.release(); // resumes the thread that is executing the "invoke" method
						}
					}
				}
			} else {
				logger.info("Ignoring reply from " + reply.getSender()
						+ " with reqId:" + reply.getSequence() + ". Currently wait reqId= " + reqId);
                            
                        }

			// Critical section ends here. The semaphore can be released
			canReceiveLock.unlock();
		} catch (Exception ex) {
			logger.info("Problem at ServiceProxy.ReplyReceived()");
			logger.error("", ex);
			canReceiveLock.unlock();
		}
	}

	protected int getReplyQuorum() {
		if (getViewManager().getStaticConf().isBFT()) {
			return (int) Math.ceil((getViewManager().getCurrentViewN()
					+ getViewManager().getCurrentViewF()) / 2) + 1;
		} else {
			return (int) Math.ceil((getViewManager().getCurrentViewN()) / 2) + 1;
		}
	}

	private int getRandomlyServerId(){
		int numServers = super.getViewManager().getCurrentViewProcesses().length;
		int pos = rand.nextInt(numServers);

		return super.getViewManager().getCurrentViewProcesses()[pos];
	}

	private class HashResponseController{
		private TOMMessage reply;
		private byte [][] hashReplies;
		private int replyServerPos;
		private int countHashReplies;

		public HashResponseController(int replyServerPos, int length) {
			this.replyServerPos = replyServerPos;
			this.hashReplies = new byte[length][];
			this.reply = null;
			this.countHashReplies = 0;
		}


		public TOMMessage getResponse(int pos, TOMMessage tomMessage){

			if(hashReplies[pos]==null){
				countHashReplies++;
			}

			if(replyServerPos == pos){
				reply = tomMessage;
				hashReplies[pos] = TOMUtil.computeHash(tomMessage.getContent());
			}else{
				hashReplies[pos] = tomMessage.getContent();
			}
			logger.info("["+this.getClass().getName()+"] hashReplies["+pos+"]="+Arrays.toString(hashReplies[pos]));

			if(hashReplies[replyServerPos]!=null){
				int sameContent = 1;
				for (int i = 0; i < replies.length; i++) {
					if ((i != replyServerPos || getViewManager().getCurrentViewN() == 1) && hashReplies[i] != null
							&& (Arrays.equals(hashReplies[i], hashReplies[replyServerPos]))) {
						sameContent++;
						if (sameContent >= replyQuorum) {
							return reply;
						}
					}
				}	
			}
			return null;
		}


		public int getNumberReplies(){
			return countHashReplies;
		}
	}
}
