pastrynode.java

pastry的java实现的2.0b版

      // signal any apps that might be waiting for the node to get ready
      synchronized (this) {
        notifyAll();
      }
    } else {
      nodeIsReady(false);
      setChanged();
      notifyObservers(new Boolean(false));

      //        Vector tmpApps = new Vector(apps);
      //        Iterator it = tmpApps.iterator();
      //        while (it.hasNext())
      //           ((PastryAppl) (it.next())).notifyFaulty();
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param newHandle DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   */
  public abstract NodeHandle coalesce(NodeHandle newHandle);

  /**
   * Overridden by derived classes, and invoked when the node has joined
   * successfully. This one is for backwards compatability. It will soon be
   * deprecated.
   */
  public abstract void nodeIsReady();

  /**
   * Overridden by derived classes, and invoked when the node has joined
   * successfully. This should probably be abstract, but maybe in a later
   * version.
   *
   * @param state true when the node is ready, false when not
   */
  public void nodeIsReady(boolean state) {

  }

  /**
   * Overridden by derived classes to initiate the join process
   *
   * @param bootstrap Node handle to bootstrap with.
   */
  public abstract void initiateJoin(NodeHandle bootstrap);

  /**
   * Add a leaf set observer to the Pastry node.
   *
   * @param o the observer.
   * @deprecated use addLeafSetListener
   */
  public void addLeafSetObserver(Observer o) {
    leafSet.addObserver(o);
  }

  /**
   * Delete a leaf set observer from the Pastry node.
   *
   * @param o the observer.
   * @deprecated use deleteLeafSetListener
   */
  public void deleteLeafSetObserver(Observer o) {
    leafSet.deleteObserver(o);
  }

  /**
   * Adds a feature to the LeafSetListener attribute of the PastryNode object
   *
   * @param listener The feature to be added to the LeafSetListener attribute
   */
  public void addLeafSetListener(NodeSetListener listener) {
    leafSet.addNodeSetListener(listener);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param listener DESCRIBE THE PARAMETER
   */
  public void deleteLeafSetListener(NodeSetListener listener) {
    leafSet.deleteNodeSetListener(listener);
  }

  /**
   * Add a route set observer to the Pastry node.
   *
   * @param o the observer.
   * @deprecated use addRouteSetListener
   */
  public void addRouteSetObserver(Observer o) {
    routeSet.addObserver(o);
  }

  /**
   * Delete a route set observer from the Pastry node.
   *
   * @param o the observer.
   * @deprecated use deleteRouteSetListener
   */
  public void deleteRouteSetObserver(Observer o) {
    routeSet.deleteObserver(o);
  }

  /**
   * Adds a feature to the RouteSetListener attribute of the PastryNode object
   *
   * @param listener The feature to be added to the RouteSetListener attribute
   */
  public void addRouteSetListener(NodeSetListener listener) {
    routeSet.addNodeSetListener(listener);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param listener DESCRIBE THE PARAMETER
   */
  public void removeRouteSetListener(NodeSetListener listener) {
    routeSet.removeNodeSetListener(listener);
  }

  /**
   * message receiver interface. synchronized so that the external message
   * processing thread and the leafset/route maintenance thread won't interfere
   * with application messages.
   *
   * @param msg DESCRIBE THE PARAMETER
   */
  public synchronized void receiveMessage(Message msg) {
    myMessageDispatch.dispatchMessage(msg);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param delivery DESCRIBE THE PARAMETER
   */
  public synchronized void receiveMessage(RawMessageDelivery delivery) {
    myMessageDispatch.dispatchMessage(delivery);
  }

  /**
   * Registers a message receiver with this Pastry node.
   *
   * @param address the address that the receiver will be at.
   * @param receiver the message receiver.
   */

  public void registerReceiver(int address,
                               PastryAppl receiver) {
    myMessageDispatch.registerReceiver(address, receiver);
  }

  /**
   * Registers an application with this pastry node.
   *
   * @param app the application
   */

  public void registerApp(PastryAppl app) {
    if (isReady()) {
      app.notifyReady();
    }
    apps.add(app);
  }

  /**
   * Schedule the specified message to be sent to the local node after a
   * specified delay. Useful to provide timeouts.
   *
   * @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
   * @return the scheduled event object; can be used to cancel the message
   */
  public abstract ScheduledMessage scheduleMsg(Message msg, long delay);

  /**
   * Schedule the specified message for repeated fixed-delay delivery to the
   * local node, beginning after the specified delay. Subsequent executions take
   * place at approximately regular intervals separated by the specified period.
   * Useful to initiate periodic tasks.
   *
   * @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 abstract ScheduledMessage scheduleMsg(Message msg, long delay,
                                               long 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 abstract ScheduledMessage scheduleMsgAtFixedRate(Message msg,
                                                          long delay, long period);

  /**
   * DESCRIBE THE METHOD
   *
   * @return DESCRIBE THE RETURN VALUE
   */
  public String toString() {
    return "Pastry node " + myNodeId.toString();
  }

  // Common API Support

  /**
   * This returns a VirtualizedNode specific to the given application and
   * instance name to the application, which the application can then use in
   * order to send an receive messages.
   *
   * @param application The Application
   * @param instance An identifier for a given instance
   * @return The endpoint specific to this applicationk, which can be used for
   *      message sending/receiving.
   */
  public rice.p2p.commonapi.Endpoint registerApplication(
                                                         rice.p2p.commonapi.Application application, String instance) {
    return new rice.pastry.commonapi.PastryEndpoint(this, application, instance, true);
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param application DESCRIBE THE PARAMETER
   * @param instance DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   */
  public rice.p2p.commonapi.Endpoint buildEndpoint(
                                                   rice.p2p.commonapi.Application application, String instance) {
    return new rice.pastry.commonapi.PastryEndpoint(this, application, instance, false);
  }

  /**
   * Schedules a job for processing on the dedicated processing thread, should
   * one exist. CPU intensive jobs, such as encryption, erasure encoding, or
   * bloom filter creation should never be done in the context of the underlying
   * node's thread, and should only be done via this method.
   *
   * @param task The task to run on the processing thread
   * @param command The command to return the result to once it's done
   */
  public void process(Executable task, Continuation command) {
    try {
      myEnvironment.getProcessor().process(task,
        command,
        myEnvironment.getSelectorManager(),
        myEnvironment.getTimeSource(),
        myEnvironment.getLogManager());

//      command.receiveResult(task.execute());
    } catch (final Exception e) {
      command.receiveException(e);
    }
  }

  /**
   * Method which kills a PastryNode. Note, this doesn't implicitly kill the
   * environment. Make sure to call super.destroy() !!!
   */
  public void destroy() {
    if (logger.level <= Logger.INFO) {
      logger.log("Destroying " + this);
    }
    myMessageDispatch.destroy();
  }


  /**
   * DESCRIBE THE METHOD
   *
   * @param handle DESCRIBE THE PARAMETER
   * @param message DESCRIBE THE PARAMETER
   */
  public abstract void send(NodeHandle handle, Message message);

  /**
   * Called by PastryAppl to ask the transport layer to open a Socket to its
   * counterpart on another node.
   *
   * @param handle
   * @param receiver
   * @param appl
   * @param timeout DESCRIBE THE PARAMETER
   */
  public abstract void connect(NodeHandle handle, AppSocketReceiver receiver, PastryAppl appl, int timeout);

}