requesthandler.java

Chord package into p2psim

/***************************************************************************
 *                                                                         *
 *                            RequestHandler.java                          *
 *                            -------------------                          *
 *   date                 : 02.09.2004, 14:02                              *
 *   copyright            : (C) 2004-2008 Distributed and                  *
 *                              Mobile Systems Group                       *
 *                              Lehrstuhl fuer Praktische Informatik       *
 *                              Universitaet Bamberg                       *
 *                              http://www.uni-bamberg.de/pi/              *
 *   email                : [e-mail]                   *
 *                          [e-mail]                 *
 *                                                                         *
 *                                                                         *
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   A copy of the license can be found in the license.txt file supplied   *
 *   with this software or at: http://www.gnu.org/copyleft/gpl.html        *
 *                                                                         *
 ***************************************************************************/

/*
 * RequestHandler.java
 *
 * Created on 2. September 2004, 14:02
 */

package de.uniba.wiai.lspi.chord.com.socket;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.Vector;

import de.uniba.wiai.lspi.chord.com.CommunicationException;
import de.uniba.wiai.lspi.chord.com.Endpoint;
import de.uniba.wiai.lspi.chord.com.EndpointStateListener;
import de.uniba.wiai.lspi.chord.com.Entry;
import de.uniba.wiai.lspi.chord.com.Node;
import de.uniba.wiai.lspi.chord.com.RefsAndEntries;
import de.uniba.wiai.lspi.chord.data.ID;
import de.uniba.wiai.lspi.util.logging.Logger;
import static de.uniba.wiai.lspi.util.logging.Logger.LogLevel.*;

/**
 * This class handles {@link Request requests} for a single incoming connection
 * from another node sent through a {@link SocketProxy proxy} that represents
 * the local node at the remote node.
 * 
 * @author sven
 * @version 1.0.5
 */
final class RequestHandler extends Thread implements EndpointStateListener {

	/**
	 * Logger for this class.
	 */
	private static Logger logger = Logger.getLogger(RequestHandler.class);

	/**
	 * The node this RequestHandler invokes methods on.
	 */
	private Node node;

	/**
	 * The socket over that this RequestHandler receives requests.
	 */
	private Socket connection;

	/**
	 * {@link ObjectOutputStream}to write answers to.
	 */
	private ObjectOutputStream out;

	/**
	 * {@link ObjectInputStream}to read {@link Request requests}from.
	 */
	private ObjectInputStream in;

	/**
	 * Indicates if this RequestHandler is connected. Used in {@link #run()} to
	 * determine if this is still listening for requests.
	 */
	boolean connected = true;

	/**
	 * The state that the {@link SocketEndpoint endpoint}, that started this
	 * request handler, is currently in. See constants of class
	 * {@link de.uniba.wiai.lspi.chord.com.Endpoint}.
	 */
	private int state;

	/**
	 * The {@link SocketEndpoint endpoint}that started this handler.
	 */
	private SocketEndpoint endpoint;

	/**
	 * This {@link Vector}contains {@link Thread threads}waiting for a state
	 * of the {@link SocketEndpoint endpoint}that permits the execution of the
	 * methods the threads are about to execute. This is also used as
	 * synchronization variable for these threads.
	 */
	private Set<Thread> waitingThreads = new HashSet<Thread>();

	/**
	 * Creates a new instance of RequestHandler
	 * 
	 * @param node_
	 *            The {@link Node node}to delegate requested methods to.
	 * @param connection_
	 *            The {@link Socket}over which this receives requests.
	 * @param ep
	 * @throws IOException 
	 * 
	 * @throws IOException
	 *             Thrown if the establishment of a connection over the provided
	 *             socket fails.
	 */
	RequestHandler(Node node_, Socket connection_, SocketEndpoint ep) throws IOException {
		super("RequestHandler_" + ep.getURL());

		if (RequestHandler.logger.isEnabledFor(INFO)) {
			RequestHandler.logger.info("Initialising RequestHandler. Socket "
					+ connection_ + ", " + ", Endpoint " + ep);
		}
		// logger = Logger.getLogger(this.getClass().toString() +
		// connection.toString());
		this.node = node_;
		this.connection = connection_;
		this.out = new ObjectOutputStream(this.connection.getOutputStream());
		try {
			this.in = new ObjectInputStream(this.connection.getInputStream());
		} catch (IOException e1) {
			out.close(); 
			throw e1; 
		}
		try {
			Request r = (Request) this.in.readObject();
			if (r.getRequestType() != MethodConstants.CONNECT) {
				Response resp = new Response(Response.REQUEST_FAILED, r
						.getRequestType(), r.getReplyWith());
				try {
					out.writeObject(resp);
				} catch (IOException e) {
				}
				try {
					out.close();
				} catch (IOException e) {}
				try {
					in.close();
				} catch (IOException e) {} 
				throw new IOException("Unexpected Message received! " + r);
			} else {
				Response resp = new Response(Response.REQUEST_SUCCESSFUL, r
						.getRequestType(), r.getReplyWith());
				out.writeObject(resp); 
			}
		} catch (ClassNotFoundException e) {
			throw new IOException("Unexpected class type received! " + e.getMessage()); 
		}
		this.endpoint = ep;
		this.state = this.endpoint.getState();
		this.endpoint.register(this);
		logger.info("RequestHandler initialised.");
	}

	/**
	 * Returns a reference to the endpoint this {@link RequestHandler} belongs
	 * to.
	 * 
	 * @return Reference to the endpoint this {@link RequestHandler} belongs to.
	 */
	SocketEndpoint getEndpoint() {
		return this.endpoint;
	}

	/**
	 * The task of this Thread. Listens for incoming requests send over the
	 * {@link #connection}of this thread. The thread can be stopped by invoking
	 * {@link #disconnect()}.
	 */
	@Override
	public void run() {
		/*
		 * As long as this is connected
		 */
		while (this.connected) {
			Request request = null;
			try {
				/* wait for incoming requests */
				logger.debug("Waiting for request...");
				request = (Request) this.in.readObject();
				if (request.getRequestType() == MethodConstants.SHUTDOWN) {
					logger.debug("Received shutdown request");
					this.disconnect();
				} else {
					logger.debug("Received request " + request);
					new InvocationThread(this, request, /*
														 * "Request_" +
														 * MethodConstants.getMethodName(request
														 * .getRequestType()) +
														 * "_" +
														 * request.getReplyWith(),
														 */this.out);
				}
			} catch (IOException e) {
				/*
				 * This can also occur if disconnect() is called, as the socket
				 * is closed then
				 */
				logger.debug("Exception occured while receiving a request. "
						+ "Maybe socket has been closed.");
				/* cannot do anything here but disconnect */
				this.disconnect();
			} catch (ClassNotFoundException cnf) {
				/* Should not occur as all nodes should have the same classes */
				logger.error("Exception occured while receiving a request ",
						cnf);
				/* cannot do anything here */
				this.disconnect();
			} catch (Throwable t) {
				logger
						.fatal("Unexpected throwable while receiving message!",
								t);
				this.disconnect();
			}
		}
	}

	/**
	 * Method to create failure responses and send them to the requestor.
	 * 
	 * @param t
	 * @param failure
	 * @param request
	 */
	void sendFailureResponse(Throwable t, String failure, Request request) {
		if (!this.connected) {