socketpastrynodefactory.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.socket;

import java.io.IOException;
import java.net.*;
import java.nio.channels.*;

import rice.Continuation;
import rice.environment.Environment;
import rice.environment.logging.*;
import rice.environment.params.Parameters;
import rice.environment.params.simple.SimpleParameters;
import rice.environment.processing.Processor;
import rice.environment.processing.simple.SimpleProcessor;
import rice.environment.random.RandomSource;
import rice.environment.random.simple.SimpleRandomSource;
import rice.p2p.commonapi.*;
import rice.p2p.commonapi.rawserialization.*;
import rice.pastry.*;
import rice.pastry.Id;
import rice.pastry.dist.DistPastryNodeFactory;
import rice.pastry.leafset.LeafSet;
import rice.pastry.messaging.*;
import rice.pastry.routing.*;
import rice.pastry.socket.messaging.*;
import rice.pastry.standard.*;
import rice.selector.*;
import rice.pastry.NodeHandle;
import rice.pastry.messaging.Message;

/**
 * Pastry node factory for Socket-linked nodes.
 *
 * @version $Id: SocketPastryNodeFactory.java,v 1.6 2004/03/08 19:53:57 amislove
 *      Exp $
 * @author Alan Mislove
 */
public class SocketPastryNodeFactory extends DistPastryNodeFactory {

  private Environment environment;

  private NodeIdFactory nidFactory;

  private int port;

  /**
   * Large period (in seconds) means infrequent, 0 means never.
   */
  private int leafSetMaintFreq;

  private int routeSetMaintFreq;

  private RandomSource random;

  private InetAddress localAddress;

  MessageDeserializer deserializer = new SPNFDeserializer();

  // this one doesn't properly coalsece NodeHandles, but when this is used, there is no PastryNode yet!
  NodeHandleFactory nhf =
    new NodeHandleFactory() {
      public NodeHandle readNodeHandle(InputBuffer buf) throws IOException {
        return SocketNodeHandle.build(buf);
      }
    };

  /**
   * DESCRIBE THE FIELD
   */
  public static byte[] TOTAL_HEADER;

  /**
   * Constructor. Here is order for bind address 1) bindAddress parameter 2) if
   * bindAddress is null, then parameter: socket_bindAddress (if it exists) 3)
   * if socket_bindAddress doesn't exist, then InetAddress.getLocalHost()
   *
   * @param nf The factory for building node ids
   * @param bindAddress which address to bind to
   * @param startPort The port to start creating nodes on
   * @param env The environment.
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public SocketPastryNodeFactory(NodeIdFactory nf, InetAddress bindAddress,
                                 int startPort, Environment env) throws IOException {
    super(env);
    localAddress = bindAddress;
    if (localAddress == null) {
      if (env.getParameters().contains("socket_bindAddress")) {
        localAddress = env.getParameters().getInetAddress("socket_bindAddress");
      }
    }
    if (localAddress == null) {
      localAddress = InetAddress.getLocalHost();
      try {
        ServerSocket test = new ServerSocket();
        test.bind(new InetSocketAddress(localAddress, port));
      } catch (SocketException e) {
        Socket temp = new Socket("yahoo.com", 80);
        localAddress = temp.getLocalAddress();
        temp.close();

        if (logger.level <= Logger.WARNING) {
          logger.log("Error binding to default IP, using " + localAddress);
        }
      }
    }

    environment = env;
    nidFactory = nf;
    port = startPort;
    Parameters params = env.getParameters();
    leafSetMaintFreq = params.getInt("pastry_leafSetMaintFreq");
    routeSetMaintFreq = params.getInt("pastry_routeSetMaintFreq");

    if (params.contains("pastry_socket_use_own_random")
      && params.getBoolean("pastry_socket_use_own_random")) {
      if (params.contains("pastry_socket_random_seed")
        && !params.getString("pastry_socket_random_seed").equalsIgnoreCase(
        "clock")) {
        this.random = new SimpleRandomSource(params.getLong("pastry_socket_random_seed"), env.getLogManager(),
          "socket");
      } else {
        this.random = new SimpleRandomSource(env.getLogManager(), "socket");
      }
    } else {
      this.random = env.getRandomSource();
    }
  }

  /**
   * Constructor for SocketPastryNodeFactory.
   *
   * @param nf DESCRIBE THE PARAMETER
   * @param startPort DESCRIBE THE PARAMETER
   * @param env DESCRIBE THE PARAMETER
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public SocketPastryNodeFactory(NodeIdFactory nf, int startPort,
                                 Environment env) throws IOException {
    this(nf, null, startPort, env);
  }

  /**
   * This method returns the routes a remote node is using
   *
   * @param handle The node to connect to
   * @param local DESCRIBE THE PARAMETER
   * @return The leafset of the remote node
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public SourceRoute[] getRoutes(NodeHandle handle, NodeHandle local)
     throws IOException {
    SocketNodeHandle wHandle = (SocketNodeHandle) handle;

    RoutesResponseMessage lm = (RoutesResponseMessage) getResponse(wHandle.getAddress(), new RoutesRequestMessage());

    return lm.getRoutes();
  }

  /**
   * This method returns the remote leafset of the provided handle to the
   * caller, in a protocol-dependent fashion. Note that this method may block
   * while sending the message across the wire.
   *
   * @param handle The node to connect to
   * @return The leafset of the remote node
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public LeafSet getLeafSet(NodeHandle handle) throws IOException {
    SocketNodeHandle wHandle = (SocketNodeHandle) handle;

    LeafSetResponseMessage lm = (LeafSetResponseMessage) getResponse(wHandle.getAddress(), new LeafSetRequestMessage());

    return lm.getLeafSet();
  }

  /**
   * Gets the LeafSet attribute of the SocketPastryNodeFactory object
   *
   * @param handle DESCRIBE THE PARAMETER
   * @param c DESCRIBE THE PARAMETER
   * @return The LeafSet value
   */
  public CancellableTask getLeafSet(NodeHandle handle, final Continuation c) {
    SocketNodeHandle wHandle = (SocketNodeHandle) handle;

    return getResponse(wHandle.getAddress(), new LeafSetRequestMessage(),
      new Continuation() {
        public void receiveResult(Object result) {
          LeafSetResponseMessage lm = (LeafSetResponseMessage) result;
          c.receiveResult(lm.getLeafSet());
        }

        public void receiveException(Exception result) {
          c.receiveException(result);
        }
      });
  }

  /**
   * This method returns the remote route row of the provided handle to the
   * caller, in a protocol-dependent fashion. Note that this method may block
   * while sending the message across the wire.
   *
   * @param handle The node to connect to
   * @param row The row number to retrieve
   * @return The route row of the remote node
   * @exception IOException DESCRIBE THE EXCEPTION
   */
  public RouteSet[] getRouteRow(NodeHandle handle, int row) throws IOException {
    SocketNodeHandle wHandle = (SocketNodeHandle) handle;

    RouteRowResponseMessage rm = (RouteRowResponseMessage) getResponse(wHandle.getAddress(), new RouteRowRequestMessage(row));

    return rm.getRouteRow();
  }

  /**
   * Gets the RouteRow attribute of the SocketPastryNodeFactory object
   *
   * @param handle DESCRIBE THE PARAMETER
   * @param row DESCRIBE THE PARAMETER
   * @param c DESCRIBE THE PARAMETER
   * @return The RouteRow value
   */
  public CancellableTask getRouteRow(NodeHandle handle, int row,
                                     final Continuation c) {
    SocketNodeHandle wHandle = (SocketNodeHandle) handle;

    return getResponse(wHandle.getAddress(), new RouteRowRequestMessage(row),
      new Continuation() {
        public void receiveResult(Object result) {
          RouteRowResponseMessage rm = (RouteRowResponseMessage) result;
          c.receiveResult(rm.getRouteRow());
        }

        public void receiveException(Exception result) {
          c.receiveException(result);
        }
      });
  }

  /**
   * This method determines and returns the proximity of the current local node
   * to the provided NodeHandle. This will need to be done in a protocol-
   * dependent fashion and may need to be done in a special way.
   *
   * @param handle The handle to determine the proximity of
   * @param local DESCRIBE THE PARAMETER
   * @return The proximity of the provided handle
   */
  public int getProximity(NodeHandle local, NodeHandle handle) {
    EpochInetSocketAddress lAddress = ((SocketNodeHandle) local)
      .getEpochAddress();
    EpochInetSocketAddress rAddress = ((SocketNodeHandle) handle)
      .getEpochAddress();

//    lAddress = new EpochInetSocketAddress(new InetSocketAddress(lAddress
//        .getAddress().getAddress(), lAddress.getAddress().getPort() + 1));

    // if this is a request for an old version of us, then we return
    // infinity as an answer
    if (lAddress.getAddress().equals(rAddress.getAddress())) {
      return Integer.MAX_VALUE;
    }

    DatagramSocket socket = null;
    SourceRoute route = SourceRoute.build(new EpochInetSocketAddress[]{rAddress});

    try {
      socket = new DatagramSocket(lAddress.getAddress().getPort());
      socket.setSoTimeout(5000);

//      byte[] data = PingManager.addHeader(route, new PingMessage(route, route
//          .reverse(lAddress), environment.getTimeSource().currentTimeMillis()),
//          lAddress, environment, logger);

      SocketBuffer sb = new SocketBuffer(lAddress, route, new PingMessage(
      /*
       *  route, route
       *  .reverse(lAddress),
       */
        environment.getTimeSource().currentTimeMillis()));

      if (logger.level <= Logger.FINE) {
        logger.log("Sending Ping to " + rAddress + " from " + lAddress);
      }
      socket.send(new DatagramPacket(sb.getBuffer().array(), sb.getBuffer().limit(), rAddress.getAddress()));

      long start = environment.getTimeSource().currentTimeMillis();
      socket.receive(new DatagramPacket(new byte[10000], 10000));
      return (int) (environment.getTimeSource().currentTimeMillis() - start);
    } catch (IOException e) {
      return Integer.MAX_VALUE - 1;
    } finally {
      if (socket != null) {
        socket.close();
      }
    }
  }

  /**
   * This method anonymously sends the given message to the remote address,