pastryappl.java

pastry的java实现的2.0b版

  /**
   * Called by pastry to deliver a message to this client.
   *
   * @param msg the message that is arriving.
   */
  public void receiveMessage(Message msg) {
    // NOTE: the idea is to synchronize on undeliveredMessages while making the check as to whether or not to add to the queue
    // the other part of the synchronization is just below, in update()
    // we don't want to hold the lock to undeliveredMessages when calling receiveMessage()
    synchronized (undeliveredMessages) {
      if (deliverWhenNotReady() || thePastryNode.isReady()) {
        // continue to receiveMessage()
      } else {
        // enable this if you want to forward RouteMessages when not ready, without calling the "forward()" method on the PastryAppl that sent the message
//      if (msg instanceof RouteMessage) {
//        RouteMessage rm = (RouteMessage)msg;
//        rm.routeMessage(this.localNode.getLocalHandle());
//        return;
//      }
        undeliveredMessages.add(msg);
        return;
      }
    }

    if (logger.level <= Logger.FINER) {
      logger.log(
        "[" + thePastryNode + "] recv " + msg);
    }
    if (msg instanceof RouteMessage) {
      RouteMessage rm = (RouteMessage) msg;

      if (enrouteMessage(rm.unwrap(), rm.getTarget(), rm.nextHop, rm.getOptions())) {
        rm.routeMessage(thePastryNode.getLocalHandle());
      }
    } else {
      messageForAppl(msg);
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param arg0 DESCRIBE THE PARAMETER
   * @param arg1 DESCRIBE THE PARAMETER
   */
  public void update(Observable arg0, Object arg1) {
    if (arg0 == thePastryNode) {
      Boolean b = (Boolean) arg1;
      if (b.booleanValue()) {
        Collection copy;
        synchronized (undeliveredMessages) {
          copy = new ArrayList(undeliveredMessages);
          undeliveredMessages.clear();
        }
        Iterator i = copy.iterator();
        while (i.hasNext()) {
          Message m = (Message) i.next();
          receiveMessage(m);
        }
      }
    }
  }

  /**
   * Sends a message to the Pastry node identified by dest. If that node has
   * failed or no point-to-point connection can be established to the node from
   * the local node in the Internet, the operation fails. Note that in this
   * case, it may still be possible to send the message to that node using
   * routeMsg.
   *
   * @param dest the destination node
   * @param msg the message to deliver.
   * @param opt send options that describe how the message is to be routed.
   * @return DESCRIBE THE RETURN VALUE
   */
  public boolean routeMsgDirect(NodeHandle dest, Message msg, SendOptions opt) {
    if (logger.level <= Logger.FINER) {
      logger.log(
        "[" + thePastryNode + "] routemsgdirect " + msg + " to " + dest);
    }
    if (!dest.isAlive()) {
      return false;
    }
    //RouteMessage rm = new RouteMessage(dest, msg, cred, opt, getAddress());
    //thePastryNode.receiveMessage(rm);

    // XXX Does routeMsgDirect need credentials?
    // Arguably, leafset messages don't need credentials because
    // individual nodeids may be signed. (not entirely true..)
    // But routeMsgDirect messages *do* need credentials. So do we
    // go back to using options to differentiate from routeMsg?

    dest.receiveMessage(msg);
    return dest.isAlive();
  }

  /**
   * Routes a message to the live node D with nodeId numerically closest to key
   * (at the time of delivery). The message is delivered to the application with
   * address addr at D, and at each Pastry node encountered along the route to
   * D.
   *
   * @param key the key
   * @param msg the message to deliver.
   * @param opt send options that describe how the message is to be routed.
   */
  public void routeMsg(Id key, Message msg, SendOptions opt) {
    if (logger.level <= Logger.FINER) {
      logger.log(
        "[" + thePastryNode + "] routemsg " + msg + " to " + key);
    }
    RouteMessage rm = new RouteMessage(key, msg, opt);
    thePastryNode.receiveMessage(rm);
  }

  // abstract methods, to be overridden by the derived application object

  /**
   * Called by pastry when a message arrives for this application.
   *
   * @param msg the message that is arriving.
   */
  public abstract void messageForAppl(Message msg);

  /**
   * Called by pastry when a message is enroute and is passing through this
   * node. If this method is not overridden, the default behaviour is to let the
   * message pass through.
   *
   * @param msg the message that is passing through.
   * @param key the key
   * @param nextHop the default next hop for the message.
   * @param opt the send options the message was sent with.
   * @return true if the message should be routed, false if the message should
   *      be cancelled.
   */
  public boolean enrouteMessage(Message msg, Id key, NodeHandle nextHop, SendOptions opt) {
    return true;
  }

  /**
   * Called by pastry when the leaf set changes.
   *
   * @param nh the handle of the node that was added or removed.
   * @param wasAdded true if the node was added, false if the node was removed.
   */
  public void leafSetChange(NodeHandle nh, boolean wasAdded) {
  }

  /**
   * Called by pastry when the route set changes.
   *
   * @param nh the handle of the node that was added or removed.
   * @param wasAdded true if the node was added, false if the node was removed.
   */
  public void routeSetChange(NodeHandle nh, boolean wasAdded) {
  }

  /**
   * Invoked when the Pastry node has joined the overlay network and is ready to
   * send and receive messages As of FreePastry 1.4.1, replaced by PastryNode
   * Observer pattern.
   */
  public void notifyReady() {
  }

  /**
   * Instructs the MessageDispatch how to behave when the PastryNode is not
   * ready. An application can override this method to return true if it wishes
   * to receive messages before Pastry is ready(). Most applications should
   * leave this as false, so that their application does not have inconsistent
   * routing. However Pastry's protocols (such as the join protocol) need to
   * receive messages before pastry is ready(). This is because they are
   * attempting to make pastry ready().
   *
   * @return false unless the node is a service
   */
  public boolean deliverWhenNotReady() {
    return false;
  }

  /**
   * Called when PastryNode is destroyed. Can be overloaded by applications.
   */
  public void destroy() {
  }


  // *******************  Applicatoin Socket Interface *****************
  /**
   * Called to open an ApplicationLevelSocket
   *
   * @param handle DESCRIBE THE PARAMETER
   * @param receiver DESCRIBE THE PARAMETER
   * @param timeout DESCRIBE THE PARAMETER
   */
  public void connect(NodeHandle handle, AppSocketReceiver receiver, int timeout) {
    thePastryNode.connect(handle, receiver, this, timeout);
  }

  /**
   * Sets an AppSocketReceiver to be called when the next socket arrives.
   *
   * @param receiver
   */
  public void accept(AppSocketReceiver receiver) {
    this.receiver = receiver;
  }

  /**
   * Calls receiver.receiveSocket(), then sets receiver to null. It sets it to
   * null to allow the application to provide SocketInitiationBackpressure
   *
   * @param socket the new socket from the network
   * @return false if receiver was null, true receiveSocket() was called
   */
  public boolean receiveSocket(AppSocket socket) {
    AppSocketReceiver theReceiver = receiver;
    receiver = null;
    if (theReceiver == null) {
      return false;
    } else {
      theReceiver.receiveSocket(socket);
      return true;
    }
  }

  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  private class LeafSetObserver implements NodeSetListener {
    /**
     * DESCRIBE THE METHOD
     *
     * @param nodeSetEventSource DESCRIBE THE PARAMETER
     * @param handle DESCRIBE THE PARAMETER
     * @param added DESCRIBE THE PARAMETER
     */
    public void nodeSetUpdate(NodeSetEventSource nodeSetEventSource, NodeHandle handle, boolean added) {
      leafSetChange(handle, added);
    }
  }

  /**
   * DESCRIBE THE CLASS
   *
   * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
   * @author jeffh
   */
  private class RouteSetObserver implements NodeSetListener {
    /**
     * DESCRIBE THE METHOD
     *
     * @param nodeSetEventSource DESCRIBE THE PARAMETER
     * @param handle DESCRIBE THE PARAMETER
     * @param added DESCRIBE THE PARAMETER
     */
    public void nodeSetUpdate(NodeSetEventSource nodeSetEventSource, NodeHandle handle, boolean added) {
      routeSetChange(handle, added);
    }
  }

}