requesthandler.java

一个对等计算发生器

/***************************************************************************
 *                                                                         *
 *                            RequestHandler.java                          *
 *                            -------------------                          *
 *   date                 : 02.09.2004, 14:02                              *
 *   copyright            : (C) 2004/2005 Distributed and                  *
 *								Mobile Systems Group                       *
 *                              Lehrstuhl fuer Praktische Informatik       *
 *                              Universitaet Bamberg                       *
 *                              http://www.lspi.wiai.uni-bamberg.de/       *
 *   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;

/**
 * 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.1
 */
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.
	 * 
	 * @uml.property name="connected"
	 */
	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
	 *             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());
		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());
		this.in = new ObjectInputStream(this.connection.getInputStream());
		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.info("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
						.warn("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) {
			return;
		}
		logger.debug("Trying to send failure response. Failure reason "
				+ failure);
		Response failureResponse = new Response(Response.REQUEST_FAILED,
				request.getRequestType(), request.getReplyWith());
		failureResponse.setFailureReason(failure);
		failureResponse.setThrowable(t);
		try {
			synchronized (this.out) {
				this.out.writeObject(failureResponse);
				this.out.flush();
				this.out.reset();