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

"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.client;

import rice.environment.logging.Logger;
import rice.p2p.commonapi.appsocket.*;
import rice.p2p.commonapi.rawserialization.MessageDeserializer;
import rice.pastry.*;
import rice.pastry.messaging.*;
import rice.pastry.standard.*;
import rice.pastry.routing.*;
import rice.pastry.leafset.*;

import java.io.IOException;
import java.util.*;

/**
 * A PastryAppl is an abstract class that every Pastry application extends. This
 * is the external Pastry API.
 *
 * @version $Id: PastryAppl.java 3309 2006-06-22 11:58:03Z jeffh $
 * @author Peter Druschel
 */
public abstract class PastryAppl implements Observer {
  /**
   * DESCRIBE THE FIELD
   */
  protected MessageDeserializer deserializer;

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

  /**
   * DESCRIBE THE FIELD
   */
  protected PastryNode thePastryNode;

  /**
   * DESCRIBE THE FIELD
   */
  protected int address;

  /**
   * DESCRIBE THE FIELD
   */
  protected Logger logger;

  /**
   * Buffered while node is not ready to prevent inconsistent routing.
   */
  LinkedList undeliveredMessages = new LinkedList();

  /**
   * holds the receiverSocket
   */
  protected AppSocketReceiver receiver;

  // constructor

  /**
   * Constructor.
   *
   * @param pn the pastry node that client will attach to.
   */
  public PastryAppl(PastryNode pn) {
    this(pn, null);
  }

  /**
   * Constructor. This constructor will perform the same tasks as the above
   * constructor, but will also create a Pastry address for this application,
   * which is dependent upon the given instance name and the class name.
   *
   * @param pn the pastry node that client will attach to.
   * @param instance The instance name of this appl.
   */
  public PastryAppl(PastryNode pn, String instance) {
    this(pn, instance, 0, null);
    register();
  }

  /**
   * Constructor for PastryAppl.
   *
   * @param pn DESCRIBE THE PARAMETER
   * @param instance DESCRIBE THE PARAMETER
   * @param address DESCRIBE THE PARAMETER
   * @param md DESCRIBE THE PARAMETER
   */
  public PastryAppl(PastryNode pn, String instance, int address, MessageDeserializer md) {
    this.address = address;
    if (instance != null) {
      this.instance = instance;
      if (address == 0) {
        this.address = StandardAddress.getAddress(this.getClass(), instance, pn.getEnvironment());
      }
    }

    thePastryNode = pn;
    logger = pn.getEnvironment().getLogManager().getLogger(getClass(), instance);
    deserializer = md;
    if (deserializer == null) {
      deserializer = new JavaSerializedDeserializer(pn);
    }
  }

  /**
   * Constructor. This constructor will perform the same tasks as the above
   * constructor, but will also create a Pastry address for this application,
   * using the specified port. Need to call register on this.
   *
   * @param pn the pastry node that client will attach to.
   * @param port DESCRIBE THE PARAMETER
   */
  public PastryAppl(PastryNode pn, int port) {
    this(pn, null, port, null);
  }

  /**
   * Returns the address of this application.
   *
   * @return the address.
   */
  public int getAddress() {
    return address;
  }

  // useful API methods

  /**
   * Gets the node id associated with this client.
   *
   * @return the node id.
   */
  public final Id getNodeId() {
    return thePastryNode.getNodeId();
  }

  /**
   * Gets the handle of the Pastry node associated with this client
   *
   * @return the node handle
   */
  public NodeHandle getNodeHandle() {
    return thePastryNode.getLocalHandle();
  }

  /**
   * Called by a layered Pastry application to obtain a copy of the leaf set.
   * The leaf set contains the nodeId to IP address binding of the l/2 nodes
   * with numerically closest counterclockwise and the l/2 nodes with
   * numerically closest clockwise nodeIds, relatively to the local node's id.
   *
   * @return the local node's leaf set
   */
  public LeafSet getLeafSet() {
    return thePastryNode.getLeafSet().copy();
  }

  /**
   * Called by a layered Pastry application to obtain a copy of the routing
   * table. The routing table contains the nodeId to IP address bindings of R
   * nodes that share the local node's id in the first n digits, and differ in
   * the n+1th digit, for 0 <= n <= ceiling(log_2^b N), where N is the total
   * number of currently live nodes in the Pastry network. The routing table may
   * be incomplete, may contain nodes that cannot be reached from the local node
   * or have failed, and the table may change at any time.
   *
   * @return The RoutingTable value
   */
  public RoutingTable getRoutingTable() {
    return thePastryNode.getRoutingTable();
  }

  /**
   * Called by the layered Pastry application to check if the local pastry node
   * is the one that is currently closest to the object key id.
   *
   * @param key the object key id
   * @return true if the local node is currently the closest to the key.
   */
  public boolean isClosest(Id key) {
    return thePastryNode.isClosest(key);
  }

  /**
   * Sets the Deserializer attribute of the PastryAppl object
   *
   * @param deserializer The new Deserializer value
   */
  protected void setDeserializer(MessageDeserializer deserializer) {
    this.deserializer = deserializer;
  }

  /**
   * DESCRIBE THE METHOD
   */
  public void register() {
    thePastryNode.registerReceiver(getAddress(), this);

    thePastryNode.addLeafSetListener(new LeafSetObserver());
    thePastryNode.addRouteSetListener(new RouteSetObserver());

    thePastryNode.registerApp(this);
    // just adds it to a list

    thePastryNode.addObserver(this);
  }

  // internal methods

  /**
   * DESCRIBE THE METHOD
   *
   * @param msg DESCRIBE THE PARAMETER
   */
  public void receiveMessageInternal(RawMessageDelivery msg) {
    Message m;
    try {
      m = msg.deserialize(deserializer);
    } catch (IOException ioe) {
      if (logger.level <= Logger.SEVERE) {
        logger.logException("Error deserializing " + msg + " in " + this + ".  Message will be dropped.", ioe);
      }
      return;
    } catch (RuntimeException re) {
      if (logger.level <= Logger.SEVERE) {
        logger.logException("Error deserializing " + msg + " in " + this + ".  Message will be dropped.", re);
      }
      throw re;
    }
    receiveMessage(m);
  }

  /**
   * 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);
    }
  }

}

