socketendpoint.java

一个对等计算发生器

/***************************************************************************
 *                                                                         *
 *                            SocketEndpoint.java                          *
 *                            -------------------                          *
 *   date                 : 12.08.2004                                     *
 *   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        *
 *                                                                         *
 ***************************************************************************/

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

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;

import de.uniba.wiai.lspi.chord.com.Endpoint;
import de.uniba.wiai.lspi.chord.com.Node;
import de.uniba.wiai.lspi.chord.data.URL;
import de.uniba.wiai.lspi.util.logging.Logger;

/**
 * This class represents an {@link Endpoint} for communication over socket
 * protocol. It provides a <code>ServerSocket</code> to that clients can
 * connect and starts for each incoming connection a {@link RequestHandler} that
 * handles {@link Request requests} for method invocations from remote nodes.
 * These {@link Request requests} are sent by {@link SocketProxy one proxy}
 * representing the node, that this is the endpoint for, at another node.
 * 
 * @author sven
 * @version 1.0.1
 */
public final class SocketEndpoint extends Endpoint implements Runnable {

	/**
	 * Logger for this endpoint.
	 */
	private Logger logger;

	/**
	 * {@link Set} containing all {@link RequestHandler}s created by this
	 * endpoint.
	 */
	private Set<RequestHandler> handlers = new HashSet<RequestHandler>();

	/**
	 * The Socket this endpoint listens to for connections.
	 */
	private ServerSocket mySocket = null;

	/**
	 * The {@link java.util.concurrent.Executor} responsible for carrying out
	 * executions of methods with help of an instance of {@link InvocationThread}.
	 */
	private final ThreadPoolExecutor invocationExecutor = InvocationThread
			.createInvocationThreadPool();

	/**
	 * Creates a new <code>SocketEndpoint</code> for the given {@link Node}
	 * with {@link URL url}. <code>url</code> must have the protocol indexed
	 * by <code>{@link URL#SOCKET_PROTOCOL}</code> in the
	 * <code>{@link URL#KNOWN_PROTOCOLS}</code> array.
	 * 
	 * @param node1
	 *            The {@link Node} node this endpoint provides connections to.
	 * @param url1
	 *            The {@link URL} of this endpoint.
	 */
	public SocketEndpoint(Node node1, URL url1) {
		super(node1, url1);
		this.logger = Logger.getLogger(SocketEndpoint.class);
		this.logger.info("Initialisation finished.");
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see de.uniba.wiai.lspi.chord.com.Endpoint#openConnections()
	 */
	protected void openConnections() {
		/* Open server socket on port specified by url */
		try {
			this.logger.debug("Trying to open server socket on port "
					+ this.url.getPort());
			this.mySocket = new ServerSocket(this.url.getPort());
			this.setState(LISTENING);
			this.logger.debug("Server socket opened on port "
					+ this.url.getPort() + ". Starting listener thread.");
			/* and start thread to listen for incoming connections. */
			Thread listenerThread = new Thread(this, "SocketEndpoint_"
					+ this.url + "_Thread");
			listenerThread.start();
			this.logger
					.debug("Listener Thread " + listenerThread + "started. ");
		} catch (IOException e) {
			/* TODO: change type of exception */
			throw new RuntimeException(
					"SocketEndpoint could not listen on port "
							+ this.url.getPort() + " " + e.getMessage());
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see de.uniba.wiai.lspi.chord.com.Endpoint#entriesAcceptable()
	 */
	protected void entriesAcceptable() {
		this.logger.debug("entriesAcceptable() called");
		this.setState(ACCEPT_ENTRIES);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see de.uniba.wiai.lspi.chord.com.Endpoint#closeConnections()
	 */
	protected void closeConnections() {
		this.setState(STARTED);
		/* try to close socket */
		try {
			this.mySocket.close();
		} catch (IOException e) {
			/* should not occur */
			this.logger.warn("Could not close socket " + this.mySocket, e);
		}
	}

	/**
	 * Run method from {@link Runnable} to accept connections from clients. This
	 * method runs until {@link #closeConnections()} is called. It creates
	 * threads responsible for the handling of requests from other nodes.
	 */
	public void run() {

		// Cleaner cleaner = new Cleaner();
		while (this.getState() > STARTED) {
			this.logger.debug("Waiting for incoming connection.");
			try {
				Socket incomingConnection = this.mySocket.accept();
				this.logger.debug("Incoming connection " + incomingConnection);
				/*
				 * Create a handler for requests that come in over the newly
				 * created socket.
				 */
				this.logger
						.debug("Creating request handler for incoming connection.");
				RequestHandler handler = new RequestHandler(this.node,
						incomingConnection, this);
				/*
				 * Remember handler to be able to close its connection (shut it
				 * down.
				 */
				this.handlers.add(handler);
				/* Start handler thread */
				this.logger.debug("Request handler created. Starting thread.");
				handler.start();
				this.logger.debug("Request handler thread started.");
			} catch (IOException e) {
				/* Can this happen? */
				if ((this.getState() > STARTED)) {
					this.logger.debug(
							"Could not accept connection from other node!", e);
				} else {
					/* Socket has been closed */
				}
				/* TODO: go on or notify some one? */
			}
		}
		this.logger.info("Listener thread stopped.");
		/* Disconnect all */
		for (RequestHandler handler : this.handlers) {
			handler.disconnect();
		}
		this.handlers.clear();
	}

	/**
	 * Schedule an invocation of a local method to be executed.
	 * 
	 * @param invocationThread
	 */
	void scheduleInvocation(InvocationThread invocationThread) {
		this.invocationExecutor.execute(invocationThread); 
	}

}