pastryendpoint.java

pastry的java实现的2.0b版

/*************************************************************************

"FreePastry" Peer-to-Peer Application Development Substrate 

Copyright 2002, Rice University. All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:

- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.

- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.

- Neither  the name  of Rice  University (RICE) nor  the names  of its
contributors may be  used to endorse or promote  products derived from
this software without specific prior written permission.

This software is provided by RICE and the contributors on an "as is"
basis, without any representations or warranties of any kind, express
or implied including, but not limited to, representations or
warranties of non-infringement, merchantability or fitness for a
particular purpose. In no event shall RICE or contributors be liable
for any direct, indirect, incidental, special, exemplary, or
consequential damages (including, but not limited to, procurement of
substitute goods or services; loss of use, data, or profits; or
business interruption) however caused and on any theory of liability,
whether in contract, strict liability, or tort (including negligence
or otherwise) arising in any way out of the use of this software, even
if advised of the possibility of such damage.

********************************************************************************/

package rice.pastry.commonapi;

import java.io.IOException;
import java.security.InvalidParameterException;

import rice.*;
import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.p2p.commonapi.*;
import rice.p2p.commonapi.appsocket.AppSocketReceiver;
import rice.p2p.commonapi.rawserialization.*;
import rice.pastry.NodeSet;
import rice.pastry.PastryNode;
import rice.pastry.NodeHandleFactory;
import rice.pastry.client.PastryAppl;
import rice.pastry.dist.*;
import rice.pastry.leafset.LeafSet;
import rice.pastry.routing.RouteSet;
import rice.pastry.routing.SendOptions;
import rice.selector.TimerTask;

/**
 * This class serves as gluecode, which allows applications written for the
 * common API to work with pastry.
 *
 * @version $Id: PastryEndpoint.java 3274 2006-05-15 16:17:47Z jeffh $
 * @author Alan Mislove
 * @author Peter Druschel
 */
public class PastryEndpoint extends PastryAppl implements Endpoint {

  /**
   * DESCRIBE THE FIELD
   */
  protected Application application;

  /**
   * DESCRIBE THE FIELD
   */
  protected String instance;

  /**
   * The commonapi's deserializer. Java Deserializer by default.
   */
  MessageDeserializer appDeserializer;

  /**
   * Constructor.
   *
   * @param pn the pastry node that the application attaches to.
   * @param application DESCRIBE THE PARAMETER
   * @param instance DESCRIBE THE PARAMETER
   * @param register DESCRIBE THE PARAMETER
   */
  public PastryEndpoint(PastryNode pn, Application application, String instance, boolean register) {
    this(pn, application, instance, 0, register);
  }

  /**
   * Constructor.
   *
   * @param pn the pastry node that the application attaches to.
   * @param application DESCRIBE THE PARAMETER
   * @param instance DESCRIBE THE PARAMETER
   * @param address DESCRIBE THE PARAMETER
   * @param register DESCRIBE THE PARAMETER
   */
//  public PastryEndpoint(PastryNode pn, Application application, int address) {
//    this(pn, application, "[PORT " + address + "]", address, true);
//  }

  public PastryEndpoint(PastryNode pn, Application application, String instance, int address, boolean register) {
    super(pn, application.getClass().getName() + instance, address, null);
    appDeserializer = deserializer;
    // use this as the apps default deserializer
    deserializer = new PEDeserializer();
    this.application = application;
    if (register) {
      register();
    }
  }

  // API methods to be invoked by applications

  /**
   * Returns this node's id, which is its identifier in the namespace.
   *
   * @return The local node's id
   */
  public Id getId() {
    return thePastryNode.getNodeId();
  }

  /**
   * Returns a handle to the local node below this endpoint.
   *
   * @return A NodeHandle referring to the local node.
   */
  public NodeHandle getLocalNodeHandle() {
    return thePastryNode.getLocalHandle();
  }

  /**
   * Returns a unique instance name of this endpoint, sort of a mailbox name for
   * this application.
   *
   * @return The unique instance name of this application
   */
  public String getInstance() {
    return instance;
  }

  /*
   *  (non-Javadoc)
   *  @see rice.p2p.commonapi.Endpoint#getEnvironment()
   */
  /**
   * Gets the Environment attribute of the PastryEndpoint object
   *
   * @return The Environment value
   */
  public Environment getEnvironment() {
    return thePastryNode.getEnvironment();
  }

  /**
   * Gets the Deserializer attribute of the PastryEndpoint object
   *
   * @return The Deserializer value
   */
  public MessageDeserializer getDeserializer() {
    return appDeserializer;
  }

  /**
   * Sets the Deserializer attribute of the PastryEndpoint object
   *
   * @param md The new Deserializer value
   */
  public void setDeserializer(MessageDeserializer md) {
    appDeserializer = md;
  }

  /**
   * This operation forwards a message towards the root of key. The optional
   * hint argument specifies a node that should be used as a first hop in
   * routing the message. A good hint, e.g. one that refers to the key's current
   * root, can result in the message being delivered in one hop; a bad hint adds
   * at most one extra hop to the route. Either K or hint may be NULL, but not
   * both. The operation provides a best-effort service: the message may be
   * lost, duplicated, corrupted, or delayed indefinitely.
   *
   * @param key the key
   * @param msg the message to deliver.
   * @param hint the hint
   */
  public void route(Id key, Message msg, NodeHandle hint) {
    if (logger.level <= Logger.FINER) {
      logger.log(
        "[" + thePastryNode + "] route " + msg + " to " + key);
    }

    PastryEndpointMessage pm = new PastryEndpointMessage(this.getAddress(), msg, thePastryNode.getLocalHandle());
    if ((key == null) && (hint == null)) {
      throw new InvalidParameterException("key and hint are null!");
    }
    boolean noKey = false;
    if (key == null) {
      noKey = true;
      key = hint.getId();
    }
    rice.pastry.routing.RouteMessage rm = new rice.pastry.routing.RouteMessage((rice.pastry.Id) key,
      pm,
      (rice.pastry.NodeHandle) hint);

    if (noKey) {
      rm.getOptions().setMultipleHopsAllowed(false);
    }
    thePastryNode.receiveMessage(rm);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param key DESCRIBE THE PARAMETER
   * @param msg DESCRIBE THE PARAMETER
   * @param hint DESCRIBE THE PARAMETER
   */
  public void route(Id key, RawMessage msg, NodeHandle hint) {
    if (logger.level <= Logger.FINER) {
      logger.log(
        "[" + thePastryNode + "] route " + msg + " to " + key);
    }

    PastryEndpointMessage pm = new PastryEndpointMessage(this.getAddress(), msg, thePastryNode.getLocalHandle());
    if ((key == null) && (hint == null)) {
      throw new InvalidParameterException("key and hint are null!");
    }
    boolean noKey = false;
    if (key == null) {
      noKey = true;
      key = hint.getId();
    }
    rice.pastry.routing.RouteMessage rm = new rice.pastry.routing.RouteMessage((rice.pastry.Id) key,
      pm,
      (rice.pastry.NodeHandle) hint);

    if (noKey) {
      rm.getOptions().setMultipleHopsAllowed(false);
    }
    thePastryNode.receiveMessage(rm);
  }


  /**
   * Schedules a message to be delivered to this application after the provided
   * number of milliseconds.
   *
   * @param message The message to be delivered
   * @param delay The number of milliseconds to wait before delivering the
   *      message
   * @return DESCRIBE THE RETURN VALUE
   */
  public CancellableTask scheduleMessage(Message message, long delay) {
    PastryEndpointMessage pm = new PastryEndpointMessage(this.getAddress(), message, thePastryNode.getLocalHandle());
    return thePastryNode.scheduleMsg(pm, delay);
  }

  /**
   * Schedules a message to be delivered to this application every period number
   * of milliseconds, after delay number of miliseconds have passed.
   *
   * @param message The message to be delivered
   * @param delay The number of milliseconds to wait before delivering the fist
   *      message
   * @param period DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   */
  public CancellableTask scheduleMessage(Message message, long delay, long period) {
    PastryEndpointMessage pm = new PastryEndpointMessage(this.getAddress(), message, thePastryNode.getLocalHandle());
    return thePastryNode.scheduleMsg(pm, delay, period);
  }

  /**
   * Schedule the specified message for repeated fixed-rate delivery to the
   * local node, beginning after the specified delay. Subsequent executions take
   * place at approximately regular intervals, separated by the specified
   * period.
   *
   * @param msg a message that will be delivered to the local node after the
   *      specified delay
   * @param delay time in milliseconds before message is to be delivered
   * @param period time in milliseconds between successive message deliveries
   * @return the scheduled event object; can be used to cancel the message
   */
  public CancellableTask scheduleMessageAtFixedRate(Message msg,
                                                    long delay, long period) {
    PastryEndpointMessage pm = new PastryEndpointMessage(this.getAddress(), msg, thePastryNode.getLocalHandle());
    return thePastryNode.scheduleMsgAtFixedRate(pm, delay, period);
  }

  /**
   * This method produces a list of nodes that can be used as next hops on a
   * route towards key, such that the resulting route satisfies the overlay
   * protocol's bounds on the number of hops taken. If safe is true, the
   * expected fraction of faulty nodes in the list is guaranteed to be no higher
   * than the fraction of faulty nodes in the overlay; if false, the set may be
   * chosen to optimize performance at the expense of a potentially higher
   * fraction of faulty nodes. This option allows applications to implement
   * routing in overlays with byzantine node failures. Implementations that
   * assume fail-stop behavior may ignore the safe argument. The fraction of
   * faulty nodes in the returned list may be higher if the safe parameter is
   * not true because, for instance, malicious nodes have caused the local node
   * to build a routing table that is biased towards malicious
   * nodes~\cite{Castro02osdi}.
   *
   * @param key the message's key
   * @param num the maximal number of next hops nodes requested
   * @param safe
   * @return the nodehandle set
   */
  public NodeHandleSet localLookup(Id key, int num, boolean safe) {
    // safe ignored until we have the secure routing support

    // get the nodes from the routing table
    return getRoutingTable().alternateRoutes((rice.pastry.Id) key, num);
  }

  /**
   * This method produces an unordered list of nodehandles that are neighbors of
   * the local node in the ID space. Up to num node handles are returned.
   *
   * @param num the maximal number of nodehandles requested
   * @return the nodehandle set
   */
  public NodeHandleSet neighborSet(int num) {
    return getLeafSet().neighborSet(num);
  }

  /**
   * This method returns an ordered set of nodehandles on which replicas of the
   * object with key can be stored. The call returns nodes with a rank up to and
   * including max_rank. If max_rank exceeds the implementation's maximum