/*******************************************************************************
 * Bunshin : DHT Replication & Caching
 * Copyright (C) 2004-2005 Ruben Mondejar
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 ******************************************************************************/

package bunshin;

import bunshin.util.*;
import bunshin.storage.*;
import bunshin.listeners.*;
import bunshin.messaging.*;

import rice.p2p.commonapi.*;

import java.util.*;
import java.net.*;
import java.io.*;



/**
 * 
 * The provided implementation of Bunshin DHT.
 * 
 * @author <a href="mailto: ruben.mondejar@urv.net">Ruben Mondejar </a>
 * 
 */

public class BunshinImpl implements Bunshin, Application {

    public static ClassLoader extraClassLoader;
	private boolean first = true;

	// ini params
	private int time = 20000;

	private final int MAX_REQUEST_TIME = 30;

	private final int TTL_CACHE = 50;
	private final int MAX_CACHE = 16;

	private int replicationFactor = 1;
	
	// My data
	private StorageManager storage;

	// Information about the nodes that replicate my data
	private ReplicaTable replicaNHs = new ReplicaTable();  // (context, key)
															// --> Collection NH
															// node/s with
															// replica/s

	// replicas from other nodes
	private Hashtable<NodeHandle,StorageManager> replicaBuckets = new Hashtable<NodeHandle,StorageManager>(); // Node
																												// NH
																												// -->
																												// StorageManager
																												// (context,key,value)
	private Hashtable<Id,NodeHandle> replicaKeyOwner = new Hashtable<Id,NodeHandle>();  // Key
																						// -->
																						// Node
																						// NH

	// Cache support
	// private Hashtable cache = new Hashtable(); //ini_version, later --> Cache
	// struct (fifo & temporal)
	private Cache cache = new Cache(TTL_CACHE,MAX_CACHE); // (fifo & temporal)

	private Hashtable control = new Hashtable();  // context ->
													// (key,PriorityList)
	
	private Hashtable urlContext = new Hashtable();
	private Hashtable pathContext = new Hashtable();

	private Hashtable listeners = new Hashtable();

	private Endpoint endPoint = null;

	private PeriodicTask refreshTask = new PeriodicTask();

	private Timer timer = new Timer();

	private boolean debug = false;

	private boolean caching = false;

	/**
	 * Identification of the application.
	 */
	public static String applicationId= "Bunshin";
	/**
	 * Identification of the application instance.
	 */
	private String appId = applicationId;

	/**
	 * Bunshin Clients
	 */
	Hashtable put_clients = new Hashtable();
	Hashtable get_clients = new Hashtable();
	Hashtable remove_clients = new Hashtable();
	Hashtable url_clients = new Hashtable();

	/**
	 * Constructor
	 */
	public BunshinImpl(){}

	public BunshinImpl(String name)	{
		this.appId = name;
	}

	public void setId(String name)	{
		this.appId = name;
	}

	/**
	 * Gets the identification of the Application
	 */
	public String getId()	{
		return appId;
	}


    public void setRefreshTime(int time) {
    	this.time = time;    	
    }
    
	public void setEndPoint (Endpoint ep)	{
		endPoint = ep;	
		if (first) timer.schedule(refreshTask,0,time);
	  else first=false;
		log("I'm node "+endPoint.getLocalNodeHandle().getId());
	}
	
	public void setStorageManager(StorageManager storage) {
		this.storage = storage;
	}
	
	   /**
		 * Sets the information of a specific context (optional method)
		 * 
		 * @param context
		 *            Id of the context
		 * @param path
		 *            String path of the storage directory (DiskStorage and
		 *            XMLStorage)
		 * @param url
		 *            String url base (ex : http://www.infourl.com)
		 * @param extra
		 *            paths of class/jar for classLoader
		 */
    public void setInfoContext(String context, String path, String url, URL[] URLsList){
    	if (path!=null)pathContext.put(context,path);
    	if (url!=null) urlContext.put(context,url);
    	if (storage instanceof DiskStorage) ((DiskStorage)storage).setPath(context,path);
    	if (storage instanceof XMLStorage) ((XMLStorage)storage).setPath(context,path);
    	// System.out.println("context : "+context+" \npath "+path+" \nurl
		// "+url);
    	// context,URLsList
    	if (URLsList!=null) {
    	 // extraClassLoader = new URLClassLoader
			// (URLsList,Thread.currentThread().getContextClassLoader());
    	 // Thread.currentThread().setContextClassLoader(extraClassLoader);
    	 for(int i=0;i<URLsList.length;i++) System.out.println(URLsList[i]);
    	 extraClassLoader =  new URLClassLoader (URLsList); 
    	}
    }


	public void activateDebug()	{
		this.debug=true;
		System.out.println("DEBUG ON");
	}

	public void activateCache()	{
		this.caching = true;
		System.out.println("CACHE ON");
	}

	public StorageManager getStorageManager() {
		return storage;
	}

	private void log(String text)	{
		if (debug) System.out.println(text);
	}

	/**
	 * Returns whether or not is the owner for the given key
	 * 
	 * @param key
	 *            The key in question
	 * @return Whether or not we are currently the root
	 */
	public boolean isOwner(Id key) {
		NodeHandleSet set = endPoint.replicaSet(key, 1);

		if (set==null)
			return false;
		else {
			
			Id x = endPoint.getId();

			NodeHandle nh = set.getHandle(0);
			if (nh==null) 	{
				return false;
			}
			return nh.getId().equals(x);
		}


	}

	/**
	 * Sends to the first node of the replicaSet a special message to check if
	 * that have the replicated keys. Remaining nodes are going to receive a
	 * normal message.
	 * 
	 */
	public void leave () {
        timer.cancel();    
	}
	
	
	/**
	 * ContextInfo = (Context, Bucket) Bucket = (Key, Values) Values = (Field,
	 * Object)
	 */
	 
	public Hashtable getStorageInfo() {

       Hashtable buckets = storage.getBuckets();        	  
       Hashtable nodeInfo = new Hashtable();        	       	      	
       
       Iterator itContext = buckets.keySet().iterator();
       // For each Context
       while(itContext.hasNext()) {        	
        	  
         String context = (String) itContext.next();
         Bucket bucket = (Bucket) buckets.get(context);
         Hashtable contextInfo = bucket.getMap();
        	  
         if (contextInfo.size()>0) nodeInfo.put(context,contextInfo);        	  
       }      
       
       return nodeInfo;
		
	}
	
	/**
	 * StorageInfo = (NodeNH, ContextInfo) ContextInfo = (Context, Bucket)
	 * Bucket = (Key, Values) Values = (Field, Object)
	 */
	 
	public Hashtable getReplicaInfo() {

        Hashtable storageInfo = new Hashtable();
        
        // For each NodeId
        Iterator itNode = replicaBuckets.keySet().iterator();
        while(itNode.hasNext()) {        	
        	
        	NodeHandle nh = (NodeHandle) itNode.next();
        	StorageManager replicaNode = (StorageManager) replicaBuckets.get(nh);
        	Hashtable buckets = replicaNode.getBuckets();
        	  
            Hashtable nodeInfo = new Hashtable();        	       	      	
        	Iterator itContext = buckets.keySet().iterator();
        	// For each Context
        	while(itContext.hasNext()) {        	
        	  
        	  String context = (String) itContext.next();
        	  Bucket bucket = (Bucket) buckets.get(context);
        	  Hashtable contextInfo = bucket.getMap();
        	  
        	  if (contextInfo.size()>0) nodeInfo.put(context,contextInfo);        	  
        	}        	
        	if (nodeInfo.size()>0)  storageInfo.put(nh,nodeInfo);          	        	
        }
        return storageInfo;
		
	}	


	/**
	 * Sets the number of replicas
	 * 
	 * @param the
	 *            number of replicas
	 */
	public void setReplicationFactor(int replicas) {
		replicationFactor=replicas;
	}

	/**
	 * Returns the number of replicas used
	 * 
	 * @return the number of replicas
	 */
	public int getReplicationFactor()
	{
		return replicationFactor;
	}
	

	/**
	 * Checks that is the owner all keys are already
	 */
	private void checkMyKeys() {
		
		Hashtable buckets = storage.getBuckets();
		Iterator it = buckets.keySet().iterator();
		while (it.hasNext()) {
			
			// for each bucket
			String context = (String) it.next();
			Bucket store = storage.getBucket(context);

			Vector v = new Vector(store.keySet());
			for(int i=0; i<v.size();i++)
			{
				Id key = (Id) v.get(i);
				if (!isOwner(key)) 	{
				
						undoReplica(context,key);
						Object obj = storage.extract(context,key);
						// put the all values

						Password pwd = null;
						if (obj instanceof Values) {
							Values values = (Values) obj;
							String field = (String) values.get("#field");
							pwd = (Password) values.get("Password");
							if (field!=null) {
							  putSecure(context,key,values,field, pwd, new BunshinAutoPutClient(context,key)); 
							}
							else putSecure(context,key,values, Context.DEFAULT_FIELD, pwd, new BunshinAutoPutClient(context,key)); 
						}
						else putSecure(context,key,obj,Context.DEFAULT_FIELD,pwd, new BunshinAutoPutClient(context,key));


				}
			}
		}
	}

	private Collection remFailNH(Collection c) {

	    NodeHandle localNH = endPoint.getLocalNodeHandle();
		Collection c2 = new Vector();
		Iterator it2 = c.iterator();
		while (it2.hasNext()) {
			
			NodeHandle nh = (NodeHandle) it2.next();

			if (nh.isAlive() || nh.equals(localNH))  {
				c2.add(nh);
			}
		}
		return c2;
	}

	private void addReplica(String context,Id key, Collection c, int tries)	{

		NodeHandle localNH = endPoint.getLocalNodeHandle();
		Values values = storage.extract(context,key);
		Collection v = Utilities.convert(endPoint.replicaSet(key,c.size()+tries+1));
		v.remove(localNH);
		v.removeAll(c);		
		// System.out.println("NHs : "+v);
		
		Iterator it = v.iterator();
		if (it.hasNext()) {
			
			NodeHandle nh = (NodeHandle) it.next();
						
			if (nh!=null && !nh.equals(localNH)){
				
				// new replica try sended
				// endPoint.route(null,new
				// ReplicaMessage(key,values,localNH),nh);
				ReplicaMessage msg = new ReplicaMessage(key,values,localNH);
				msg.setContext(context);
				endPoint.route(null,msg,nh);
			}
		}
	}

	/**
	 * Checks the replica nodes that stores my keys/values
	 */
	private void checkMyRemoteReplicas() {
		
		Hashtable buckets = storage.getBuckets();
		Iterator it = buckets.keySet().iterator();
		while (it.hasNext()) {
			
			// for each bucket
			String context = (String) it.next();			
			Bucket store = storage.getBucket(context);
			
			Vector v = new Vector(store.keySet());
			int size = v.size();
			for(int i=0; i<size;i++)
			{
				Id key = (Id) v.get(i);
				Collection nhs = (Collection) replicaNHs.get(context,key);
				
				if (nhs==null) nhs = new HashSet();
				else {
				  nhs = new HashSet(nhs);
				}
				replicaNHs.put(context,key,(Collection)nhs);
				
				
				if (nhs!=null)
				{
					// Check to find some failed node
					HashSet nhs1 = new HashSet(nhs); 
					Collection nhs2 = remFailNH((Collection)nhs1.clone());
					// Chech to find a not suficient number of copies of some
					// key
					 
					//System.out.println(key+" : "+nhs2.size()+" < "+replicationFactor+"?");				
										
					int tries = replicationFactor - nhs2.size();
					if (tries>0) {
					  //System.out.println("Try "+tries+" times to complete :"+nhs2);					
					  addReplica(context,key,nhs2,tries);					
					}
					
				}
			} //System.out.println("-------- "+size);
		}
	}

	private void checkReplicas() {

		Vector v = new Vector(replicaBuckets.keySet());
		for(int i=0; i<v.size();i++) {
			
			NodeHandle nh = (NodeHandle) v.get(i);
			if (!nh.isAlive())	{
				
				// System.out.println("id "+endPoint.getId()+" checkReplica of
				// "+nh+" is alive ?"+nh.isAlive());


				StorageManager replicas = (StorageManager) replicaBuckets.remove(nh);
                if (replicas!=null) {
				  Hashtable buckets = replicas.getBuckets();
				  Iterator it = buckets.keySet().iterator();
				  while (it.hasNext()) 	{
					// for each bucket
					String context = (String) it.next();
					Bucket bucket = replicas.getBucket(context);
	                         if (bucket!=null) {

					  Vector c = new Vector(bucket.keySet());
					  while(!c.isEmpty())  {

						try {
						  Id key = (Id) c.firstElement();
						  Object obj = replicas.remove(context,key);
						  c.remove(key);
						  replicaKeyOwner.remove(key);
						  Password pwd = null;
						  boolean getcheck = true;
						  
						  if (obj instanceof Values) {
							Values values = (Values) obj;
							String field = (String) values.get("#field");
							pwd = (Password) values.get("Password");							
							if (field!=null) {
								putSecure(context,key,values,field,pwd);
							}
							else putSecure(context,key,obj,Context.DEFAULT_FIELD,pwd);
						  }
						  else if (obj != null) {
                             putSecure (context, key, obj, Context.DEFAULT_FIELD,pwd);
                          }
						}
						catch(StorageException ex) 	{
						  ex.printStackTrace();
						}
					  }
					}
				}
			  }
			}
		}
	}

	/**
	 * This method is invoked to inform the application that the given node has
	 * either joined or left the neighbor set of the local node, as the set
	 * would be returned by the neighborSet call.
	 * 
	 * @param handle
	 *            The handle that has joined/left
	 * @param joined
	 *            Whether the node has joined or left
	 */
	public void update(NodeHandle handle, boolean joined) {

		if (joined)	{
			// only checkMyKeys();
			refresh(1);
			// System.out.println ("Update node ["+endPoint.getId()+"] :
			// "+handle+" joined ");
		}
		else {
			// call to checkMyRemoteReplicas() & checkReplicas()
			refresh(2);
			// System.out.println ("Update node ["+endPoint.getId()+"] :
			// "+handle+" left ");
		}

	}

	/**
	 * This method is invoked on applications when the underlying node is about
	 * to forward the given message with the provided target to the specified
	 * next hop. Applications can change the contents of the message, specify a
	 * different nextHop (through re-routing), or completely terminate the
	 * message.
	 * 
	 * @param message
	 *            The message being sent, containing an internal message along
	 *            with a destination key and nodeHandle next hop.
	 * 
	 * @return Whether or not to forward the message further
	 */
	// public boolean forward(Message message) {
	// System.out.println ("Forward app ["+appId+"] node
	// ["+endPoint.getId()+"]");
	// return true;
	// }

	public boolean forward(RouteMessage message) {
		
		// System.out.println ("Forward app ["+appId+"] node
		// ["+endPoint.getId()+"]");
		// System.out.println ("RouteMessage, prev :
		// ["+message.getPrevNodeHandle()+"], next :
		// ["+message.getNextHopHandle()+"]");

		boolean forward = true;

		BunshinMessage msg = (BunshinMessage) message.getMessage();
		String context = msg.getContext();

		if (caching && msg instanceof GetMessage) {
			
			GetMessage gmsg = (GetMessage) msg;
			if (gmsg.getPhase()==1) {

				Id key = gmsg.getKey();
			    String field = gmsg.getField();
				Object value = null;
				boolean haveValue = false;

				if (isOwner(key)) {
					
					if (field!=null) value = storage.retrieve(context,key,field);
			        else value = storage.extract(context,key);
					
					if (value!=null) haveValue = true;
				}

					// I'm not owner but I have replica
				else if (replicaKeyOwner.containsKey(key))	{

                   Password pwd = gmsg.getPassword();
  			       Values values = getReplicaValue(context,key);
  			       
				   if (values!=null) {
					 Password oldPwd = (Password) values.get("Password");
					 boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	  	 
					 if (validate) {
					   if (field!=null) value = values.get(field);
			           else value = values;
					 }
				   }	 
					 
					GetMessage fmsg = new GetMessage(key,value);
					fmsg.setContext(context);
					endPoint.route (null, fmsg, gmsg.getNH());

					forward = false;
					haveValue = true;
					log ("Replica response "+key);
				}
					// I'm not owner and don't have any replica, but I have
					// value-cache
				else if (cache.contains(context,key))	{
					
				   Password pwd = gmsg.getPassword();
   			       Values values = cache.get(context,key);
  			       
				   if (values!=null) {
					 Password oldPwd = (Password) values.get("Password");
					 boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	  	 
					 if (validate) {
					   if (field!=null) value = values.get(field);
			           else value = values;
					 }
				   }	 
					 
					GetMessage fmsg = new GetMessage(key,value);
					fmsg.setContext(context);
					endPoint.route (null, fmsg, gmsg.getNH());

					forward = false;
					haveValue = true;
					// CACHE_LOG
					log ("Cache response "+key);
				}

				if (haveValue)		{

					NodeHandle previous = gmsg.getPreviousNH();
					Id obj_key = gmsg.getKey();

					// CONTROL VERSION
					Hashtable pairs = (Hashtable) control.get(context);
					if (pairs==null) pairs = new Hashtable();

					PriorityList list = (PriorityList) pairs.get(obj_key);
					if (list==null) list = new PriorityList();

					list.update(previous);

					pairs.put(obj_key,list);
					control.put(context,pairs);

					log ("Request updated from "+previous.getId()+" key "+obj_key);

				}
				gmsg.setPreviousNH(endPoint.getLocalNodeHandle());
			}
		}

		return forward;
	}

	/**
	 * This method is called on the application at the destination node for the
	 * given id.
	 * 
	 * @param id
	 *            The destination id of the message
	 * @param message
	 *            The message being sent
	 */
	public void deliver (Id id, Message message){

		Id key;
		Object value;

		BunshinMessage bm = (BunshinMessage) message;
		String context = bm.getContext();

		log ("Deliver app ["+appId+"] node ["+endPoint.getId()+"] \nmessage : "+message);

		if (message instanceof PutMessage) {

			PutMessage mesg = (PutMessage) message;
			key = mesg.getKey();
			value = mesg.getValue();
			String field = mesg.getField();
            
			try {
				
			  Password pwd = mesg.getPassword();	      
		      Password oldPwd = (Password) storage.retrieve(context,key,"Password");
		      
		      boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	
		      
		      boolean addVersion = false;
		      if (value instanceof Values) validate = validate & CVS(value,storage.extract(context,key));
		      else addVersion = true;
		      
		      // #DEBUG
		      System.out.println("PUT REQUEST :"+key);
			  if (validate) {
			  	
			    storage.write(context,key,value,field);
				remoteListeners(context,key,value,field,true);
				
                if (addVersion) {
                  Values oldValues = (Values)storage.extract(context,key);
                  if (oldValues.containsKey("CVS")) {
                    int oldVersion = ((Integer) oldValues.get("CVS")).intValue();
                    storage.write(context,key,new Integer(oldVersion+1),"CVS");
                  }
                }                
                
                if (oldPwd == null && pwd!=null) storage.write(context,key,pwd,"Password"); 
                
				Values values = storage.extract(context,key);
				// send replicas
				replicate(context,key,values);
				log("I'm the owner of the "+value);
			       	
			  }
			  else  {
			  	log("Wrong validation of put key : "+key); 
			  }	
			  			
			  
			    //originator is replica
				Collection c = (Collection) replicaNHs.get(context,key);
				if (c==null) {
					c = new Vector();
				}
				c.add(mesg.getNH());
				replicaNHs.put(context,key,c);
			    
			  
				PutAckMessage msg = new PutAckMessage(key,field,validate,endPoint.getLocalNodeHandle());
				msg.setContext(context);
				endPoint.route (null, msg, mesg.getNH());
			}
			catch(StorageException ex)	{
				ex.printStackTrace();
				// activate threshold?
				// send reject ack ?
			}
		
		
		
		}
		else if (message instanceof PutAckMessage) {
		  
		  PutAckMessage mesg = (PutAckMessage) message;
		  key = mesg.getKey();		  
		  String field = mesg.getField();
		  boolean successful = mesg.getSuccessful();
		 		  	
		  if (put_clients!=null) {
		    
		    // Client is only for one time
			Vector id_clients = (Vector) put_clients.remove(key);
			if (id_clients!=null) {
			
			  for(int i=0; i<id_clients.size(); i++)	{
			    BunshinPutClient client = (BunshinPutClient) id_clients.get(i);
			    client.put(successful,mesg.getSource());
			  }
			}
		  }							
		}	
			
		else if (message instanceof PutURLMessage)	{
			
			PutURLMessage mesg = (PutURLMessage) message;
			key = mesg.getKey();
			URL url = mesg.getURL();
			String field = mesg.getField();

			Values values = storage.extract(context,key);
			if (values == null) values = new Values();
			// TODO : url --> own url
			replicateURL(context,key,values,url,field);

			Object file = download(url);
			if (file!=null)	{
				
				try	{
					
					String filename = url.getFile();
			        filename = filename.substring(filename.lastIndexOf('/') + 1);
					values.put("#filename",filename);
		            // values.put("#field",field);
		            values.put(field,file);


					storage.write(context,key,values,field);

					log("I'm the owner of the url value : "+url);
				}
				catch(StorageException ex)
				{
					ex.printStackTrace();
					// activate threshold?
					// send reject ack ?
				}
			}
		}
		else if (message instanceof ModifyMessage)	{

			ModifyMessage mesg = (ModifyMessage) message;
			key = mesg.getKey();
			value = (Collection) mesg.getValue(); // MUST BE COLLECTION!
			String field = mesg.getField();

			try
			{
				Collection c = null;
				undoReplica(context,key);
				if (mesg.isAdd())
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					c.addAll((Collection)value);
					storage.write(context,key,c,field);
					remoteListeners(context,key,value,field,true);
				}
				else
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					else c.removeAll((Collection)value);
					storage.write(context,key,c,field);
					remoteListeners(context,key,value,field,false);
				}
				// update replicas
				Values values = storage.extract(context,key);
				replicate(context,key,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else if (message instanceof ListenerMessage)
		{
			try
			{
				ListenerMessage mesg = (ListenerMessage) message;
				key = mesg.getKey();
				String field = mesg.getField();

				undoReplica(context,key);
				Listeners ls = (Listeners) storage.retrieve(context,key,"Listeners");
				if (mesg.isAdd())
				{
					if (ls==null)
					{
						ls = new Listeners();
					}
					ls.add(mesg.getNH(),mesg.getField());
				}
				else
				{
					if (ls!=null)
					{
						ls.remove(mesg.getNH(),mesg.getField());
					}
					else ls = new Listeners();
				}
				storage.write(context,key,ls,"Listeners");
				Values values = storage.extract(context,key);
				replicate(context,key,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
			}
		}
		
		else if (message instanceof NotifyMessage)	{

			NotifyMessage mesg = (NotifyMessage) message;
			key = mesg.getKey();
			value = mesg.getValue();

			RemoteListener rl = (RemoteListener) listeners.get(key);
			if (rl!=null) rl.eventArrived(value,mesg.isAdd());

		}

		else if (message instanceof RemoveMessage)	{
			
			RemoveMessage mesg = (RemoveMessage) message;
			key = mesg.getKey();
			String field = mesg.getField();
			try
			{


              Password pwd = mesg.getPassword();	      
		      Password oldPwd = (Password) storage.retrieve(context,key,"Password");
		      
		      boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	
		  
			  if (validate) {
			  	
				// remove replicas
				undoReplica(context,key);
				log("Removing "+key);
				if (field!=null)
				{
					value = storage.retrieve(context,key,field);
					remoteListeners(context,key,value,field,false);
				    storage.delete(context,key,field);
					
						Values values = storage.extract(context,key);
					if (values!=null) {
					
                    if ( values!=null &&
                    	(values.size()==2&&values.containsKey("CVS")&&values.containsKey("Password"))
                        ||
                        (values.size()==1&&(values.containsKey("CVS")||values.containsKey("Password")))
                        ||
                        (values.size()==0)
                        )
					
					storage.remove(context,key);
					}
				}
				else
				{
					// value = storage.extract(context,key);
					storage.remove(context,mesg.getKey());
					// remoteListeners(key,value,null,false);
				}
              }  
			    RemoveAckMessage msg = new RemoveAckMessage(key,field,validate);
				msg.setContext(context);
				endPoint.route (null, msg, mesg.getNH());

			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// send reject ack ?
			}
		}
		else if (message instanceof RemoveAckMessage) {
			  
			  RemoveAckMessage mesg = (RemoveAckMessage) message;
			  key = mesg.getKey();		  
			  String field = mesg.getField();
			  boolean successful = mesg.getSuccessful();
			 		  	
			  if (remove_clients!=null) {
			    
			    // Client is only for one time
				Vector id_clients = (Vector) remove_clients.remove(key);
				if (id_clients!=null) {
				
				  for(int i=0; i<id_clients.size(); i++)	{
				    BunshinRemoveClient client = (BunshinRemoveClient) id_clients.get(i);
				    client.remove(successful);
				  }
				}
			  }							
			}	
		else if (message instanceof RemoveReplicaMessage)
		{

			RemoveReplicaMessage mesg = (RemoveReplicaMessage) message;

			NodeHandle owner = (NodeHandle) replicaKeyOwner.remove(mesg.getKey()); // remove
																					// key
																					// of
																					// the
																					// owner
																					// node
																					// key
																					// space

			if (owner!=null && replicaBuckets.containsKey(owner)) {
				
				StorageManager replicaStorage = (StorageManager) replicaBuckets.remove(owner);
				Bucket replicas = (Bucket) replicaStorage.getBucket(context);

				log("Replicas "+replicas);
				if (replicas!=null) {
					
					try {

					  replicas.remove(mesg.getKey());					
					  if (replicas.size()>0)  replicaStorage.put(context,replicas);
					  else  replicaStorage.removeBucket(context);

					  replicaBuckets.put(mesg.getNH(),replicaStorage);
					  log("Replicas Final "+replicaBuckets);

					}
					catch(StorageException ex)
					{
						ex.printStackTrace();
					}
				}
			}

		}

		else if (message instanceof GetMessage) {
			GetMessage mesg = (GetMessage) message;
			int phase = mesg.getPhase();
			switch(phase)
			{
				case 1 :
				
				    key = mesg.getKey();
					String field = mesg.getField();
					value=null;
										
				    Password pwd = mesg.getPassword();	      
		            Password oldPwd = (Password) storage.retrieve(context,key,"Password");
		            
		            boolean validate = true;
		            // getcheck system
		            if (oldPwd != null) {
		            	System.out.println("Getcheck key "+key+": "+oldPwd.isGetCheck());
		            	if (oldPwd.isGetCheck()) validate = (pwd!=null && oldPwd.equals(pwd));
		            	else validate = true;
		            }	
		            
		            if (validate) {
		        
			  		  if (field!=null) value = storage.retrieve(context,key,field);
			          else value = storage.extract(context,key);
                    }      
                    else log("Wrong validation of get key : "+key);            
				
					
					GetMessage msg = new GetMessage(key,value);
					msg.setContext(context);
					endPoint.route (null, msg, mesg.getNH());

					break;
				case 2 :
					key = mesg.getKey();
					value = mesg.getValue();

					if (get_clients!=null)
					{

						// Client is only for one time
						Vector id_clients = (Vector) get_clients.remove(key);
                        if (id_clients!=null) {
                        
						  for(int i=0; i<id_clients.size(); i++) {
						  
							BunshinGetClient client = (BunshinGetClient) id_clients.get(i);
							client.get(value);
						  }
						}  
					}

					break;
			}

		}
		else if (message instanceof GetURLMessage) {
			GetURLMessage mesg = (GetURLMessage) message;
			int phase = mesg.getPhase();
			switch(phase) {
				
				case 1 :
				
				    key = mesg.getKey();
					String field = mesg.getField();
					URL url = null;		
					
					Values values = storage.extract(context,key);           
        
                    String f = (String)values.get("#field");
                    if (f!=null && f.equals(field)) {
           	
                      String filename = (String)values.get("#filename");
		              if (filename!=null) {          
                        String url_base = (String) urlContext.get(context);
                        try {
                        
                        url = new URL(url_base+'/'+filename);                        	
                        } catch(Exception e) {e.printStackTrace();
                        }
                      }            
           	
                    } 
                
					
					GetURLMessage msg = new GetURLMessage(key,url);
					msg.setContext(context);
					endPoint.route (null, msg, mesg.getNH());

			    break;					
				case 2 :
					key = mesg.getKey();
					URL url_value = mesg.getURL();

					if (url_clients!=null)
					{

						// Client is only for one time
						Vector id_clients = (Vector) url_clients.remove(key);
                        if (id_clients!=null) {
                        
						  for(int i=0; i<id_clients.size(); i++) {
						  
							BunshinURLClient client = (BunshinURLClient) id_clients.get(i);
							client.get(url_value);
						  }
						}  
					}

					break;
			}

		}
		else if (message instanceof ReplicaMessage)	{
			
			ReplicaMessage mesg = (ReplicaMessage) message;
			key = mesg.getKey();
			Values values = mesg.getValues();

			if (mesg.isURLReplica()){
				
				value = download(mesg.getURL());
				values.put(mesg.getField(),value);

				String filename = mesg.getURL().getFile();
			    filename = filename.substring(filename.lastIndexOf('/') + 1);
				values.put(filename,"filename");
			}

			if (!storage.exists(context,key)) {
				

				putReplica(mesg.getNH(),context,key,values);

				ReplicaAckMessage msg = new ReplicaAckMessage(key,endPoint.getLocalNodeHandle());
				msg.setContext(context);
				endPoint.route (null, msg, mesg.getNH());

				// #DEBUG
				System.out.println("REPLICA REQUEST :"+key);
				log("Saving replica in context"+context+" values : "+values+" of "+mesg.getNH().getId());
			}
		}
		else if (message instanceof ReplicaAckMessage)	{
			
			ReplicaAckMessage mesg = (ReplicaAckMessage) message;
			key = mesg.getKey();
			Collection c = (Collection) replicaNHs.get(context,key);
			if (c==null)
			{
				c = new Vector();
			}
			c.add(mesg.getNH());
			replicaNHs.put(context,key,c);
			log("Replica ack "+key+" of "+mesg.getNH().getId());
		}

		else if (message instanceof LeaveMessage)
		{
			LeaveMessage mesg = (LeaveMessage) message;
			NodeHandle source = mesg.getNH();

			// extract content : Hashtable(context(Hashtable(key,value))
			if (mesg.isContent())
			{
				Hashtable newBuckets = (Hashtable) mesg.getContent();
				Iterator itBucket = newBuckets.keySet().iterator();
				while(itBucket.hasNext())
				{

					context = (String) itBucket.next();
					Bucket newKeys = (Bucket) newBuckets.get(context);

					if (newKeys.size()>0)
					{
						try
						{
							storage.add(context,newKeys);
							log("New keys received : "+newKeys);
						}
						catch(StorageException ex)
						{
							ex.printStackTrace();
							// activate threshold?
							// send reject ack ?
						}
					}
				}
			}

			// System.out.println(replicaBuckets);
			// System.out.println("Deleting "+mesg.getNH().getId()+" entries");

			// remove replicas of a given owner

			StorageManager replicas = (StorageManager) replicaBuckets.remove(mesg.getNH());
			if (replicas!=null)
			{
				Hashtable buckets = replicas.getBuckets();
				if (buckets!=null)
				{
					Iterator it =  buckets.values().iterator();
					while(it.hasNext())
					{
						Bucket bucket = (Bucket) it.next();
						Iterator it2 = bucket.keySet().iterator();
						while(it2.hasNext())
						{
							Id k = (Id) it2.next();
							NodeHandle owner = (NodeHandle) replicaKeyOwner.get(k);
							if (owner!=null && owner.equals(mesg.getNH())) replicaKeyOwner.remove(k);
						}
					}
				}
			}

		}
		else if (message instanceof CacheMessage)
		{
			if (caching)
			{
				CacheMessage mesg = (CacheMessage) message;
				key = mesg.getKey();
				Values values = mesg.getValues();

				if (values!=null) cache.put(context,key,values);

				// CACHE_LOG
				log("Cacheando key "+key);
			}
		}
		else
		{
			log ("Deliver Error : Wrong message : "+message);
		}
	}


	/** ******************************************************************************* */
	/* BUNSHIN METHODS */
	/** ******************************************************************************* */

  /**
	 * Stores the (key,value) pair in the Bunshin DHT in the default context and
	 * with default field
	 * 
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 */
	public void put(Id key, Object value)
	{
		put(Context.DEFAULT_CONTEXT,key,value,Context.DEFAULT_FIELD);
	}

   /**
	 * Stores the (key,value,field) in the Bunshin DHT in the default context
	 * 
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 */  
	public void put(Id key, Object value, String field)	{
		put(Context.DEFAULT_CONTEXT,key,value,field);
	}
	
	/**
	 * Stores the (context,key,value) in the Bunshin DHT with default field
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 */  
	public void put(String context, Id key, Object value)	{
		put(context,key,value,Context.DEFAULT_FIELD);
	}
  
   /**
	 * Stores the (context,key,value,field) in the Bunshin DHT
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 */  
  public void put(String context, Id key, Object value, String field) {
       
       Password pwd = null;
       putSecure(context,key,value,field,pwd);
    }
    
  /**
	 * Stores the (context,key,value,field,pass) in the Bunshin DHT
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 * @param pass
	 *            password
	 */  
	public void putSecure(String context, Id key, Object value, String field, String pass) {
     	
     	Password pwd = null;
		if (pass!=null) pwd = new Password(pass);		
		
		putSecure(context, key, value, field,pwd);
		
	}
	
	  /**
		 * Stores the (context,key,value,field,pass) in the Bunshin DHT
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param pass
		 *            password
		 */  
		public void putSecure(String context, Id key, Object value, String field, String pass, boolean getcheck) {
	     	
	     	Password pwd = null;
			if (pass!=null) pwd = new Password(pass,getcheck);		
			
			putSecure(context, key, value, field,pwd);
			
		}

	  /**
		 * Stores the (context,key,value,field,pass) in the Bunshin DHT
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param Password
		 *            password
		 * @param getcheck
		 *            if true, the password will be only for remove
		 */  
	private void putSecure(String context, Id key, Object value, String field, Password pwd) {
     	
		if (isOwner(key)) {
			
			try {             		      
		      	      
		      
		      Password oldPwd = (Password) storage.retrieve(context,key,"Password");
		      
		      boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	
		      
		      boolean addVersion = false;
		      if (value instanceof Values) validate = validate & CVS(value,storage.extract(context,key));
		      else addVersion = true;
		      
			  if (validate) {
			  	
			  	storage.write(context,key,value,field);
				remoteListeners(context,key,value,field,true);
                	
                if (addVersion) {
                  Values oldValues = (Values)storage.extract(context,key);
                  if (oldValues.containsKey("CVS")) {
                  	int oldVersion = ((Integer) oldValues.get("CVS")).intValue();
                    storage.write(context,key,new Integer(oldVersion+1),"CVS"); 
                  }
                  
                }  
                
                if (oldPwd == null && pwd!=null) storage.write(context,key,pwd,"Password"); 
                 
				// send replicas
				Values values = storage.extract(context,key);
				replicate(context,key,values);
				log("I'm the owner of the "+value);
				
				
				// Client is only for one time
			    Vector id_clients = (Vector) put_clients.remove(key);
			    if (id_clients!=null) {
			
			      for(int i=0; i<id_clients.size(); i++)	{
			        BunshinPutClient client = (BunshinPutClient) id_clients.get(i);
			        client.put(true,endPoint.getLocalNodeHandle()); //fake
			      }
			    }
			  
			       	
			  }
			  else  log("Wrong validation of put key : "+key); 			  
			
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else {
			
			PutMessage msg = new PutMessage(key,value,field,pwd,endPoint.getLocalNodeHandle());
			msg.setContext(context);
			endPoint.route(key,msg, null);
		}
	}



  /**
	 * Stores the (key,value) pair in the Bunshin DHT in a default field
	 * 
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param client
	 *            listener BunshinPutClient
	 */
	public void put(Id key, Object value, BunshinPutClient client) {
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
		
		put(Context.DEFAULT_CONTEXT,key,value,Context.DEFAULT_FIELD);
	}

	   /**
		 * Stores the (context,key,value) in the Bunshin DHT with result
		 * listener
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param client
		 *            listener BunshinPutClient
		 */  
	public void put(String context, Id key, Object value, BunshinPutClient client)
	{
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
		
		put(context,key,value,Context.DEFAULT_FIELD);
	}

  /**
	 * Stores the (key,value,field) in the Bunshin DHT with result listener
	 * 
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 * @param client
	 *            listener BunshinPutClient
	 */ 
	public void put(Id key, Object value, String field, BunshinPutClient client)
	{
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
		
		put(Context.DEFAULT_CONTEXT,key,value,field);
	}
	
	   /**
		 * Stores the (context,key,value,field) in the Bunshin DHT with result
		 * listener
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param client
		 *            listener BunshinPutClient
		 */  
	public void put(String context, Id key, Object value, String field, BunshinPutClient client){
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
			
		put(context,key,value,field);
	}
	
	 /**
		 * Stores the (context,key,value,field,pass) in the Bunshin DHT with
		 * result listener
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param pass
		 *            password
		 * @param client
		 *            listener BunshinPutClient
		 */  
	public void putSecure(String context, Id key, Object value, String field, String pass, BunshinPutClient client){
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
			
		putSecure(context,key,value,field,pass,true);
	}
	
	 /**
		 * Stores the (context,key,value,field,pass) in the Bunshin DHT with
		 * result listener
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param pass
		 *            password
		 * @param client
		 *            listener BunshinPutClient
		 * @param getcheck
		 *            if true, the password will be only for remove
		 */  
		public void putSecure(String context, Id key, Object value, String field, String pass, boolean getcheck, BunshinPutClient client){
			
			Vector id_clients;

		    if (!put_clients.containsKey(key)) id_clients = new Vector();
			else id_clients =  id_clients = (Vector) put_clients.get(key);

			id_clients.add(client);
			put_clients.put(key,id_clients);
				
			putSecure(context,key,value,field,pass,getcheck);
		}
	
	  /**
		 * Stores the (context,key,value,field,pass) in the Bunshin DHT with
		 * result listener
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 * @param Password
		 *            password
		 * @param client
		 *            listener BunshinPutClient
		 * @param getcheck
		 *            if true, the password will be only for remove
		 */  
	private void putSecure(String context, Id key, Object value, String field, Password pwd, BunshinPutClient client){
		
		Vector id_clients;

	    if (!put_clients.containsKey(key)) id_clients = new Vector();
		else id_clients =  id_clients = (Vector) put_clients.get(key);

		id_clients.add(client);
		put_clients.put(key,id_clients);
			
		putSecure(context,key,value,field,pwd);
	}
	
	  /**
		 * Stores the URL and this content (key,url,field) in the Bunshin DHT
		 * and the value is downloaded, in the default context.
		 * 
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to store
		 * @param field
		 *            String key
		 */
	public void putURL(Id key, URL url, String field)	{
		putURL(Context.DEFAULT_CONTEXT,key,url,field);
	}

  /**
	 * Stores the URL and this content (key,url,field) in the Bunshin DHT and
	 * the value is downloaded.
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 */
	public void putURL(String context, Id key, URL url, String field){
		// URL url = new URL(urlName);

		if (isOwner(key))
		{
			try
			{
				Values values = storage.extract(context,key);
				if (values == null) values = new Values();

				// replicateURL(context,key,values,url,field); //TODO --> move

				Object file = download(url);
				if (file!=null)
				{

					String filename = url.getFile();
			        filename = filename.substring(filename.lastIndexOf('/') + 1);
		            	values.put("#filename",filename);
		            // values.put("#field",field);
		            values.put(field,file);

					storage.write(context,key,values,field);

					remoteListeners(context,key,url,field,true);
					log("I'm the owner of the url value : "+url);
				}
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else
		{
			// endPoint.route(key, new PutURLMessage(key,url,field), null);
			PutURLMessage msg = new PutURLMessage(key,url,field);
			msg.setContext(context);
			endPoint.route(key,msg, null);
		}
	}

  

  
	private Object download(URL url){
		byte[] bite = null;
		try	{

             String protocol = url.getProtocol();
             int contentLength;
             String contentType;      
             
             InputStream in;
             
             System.out.println("URL protocol : "+protocol);
                        
             if (protocol.equals("file")) {
               File file = new File(url.getFile());  	
               contentType = "file";
			   contentLength = (int) file.length(); 
               in = new FileInputStream(file);                                                   
             }
             else {	
             
			   URLConnection uc = url.openConnection();
			   contentType = uc.getContentType();
			   contentLength = uc.getContentLength();
	
			   InputStream raw = uc.getInputStream();
			   in  = new BufferedInputStream(raw);
			 }  

			   bite = new byte[contentLength];
			   int bytesRead = 0;
			   int offset = 0;
			   while (offset < contentLength) {
				 bytesRead = in.read(bite, offset, bite.length-offset);
				 if (bytesRead == -1) break;
				 offset += bytesRead;
			   }
			   in.close();

			   if (offset != contentLength)	{
				 throw new IOException("Only read " + offset + " bytes; Expected " + contentLength + " bytes");
			   }			 

		}
		catch(Exception ex)	{
			
			ex.printStackTrace();
		}

		return bite;
	}

  /**
	 * Modifies the content of a specific field
	 * 
	 * @param key
	 *            Id of the value
	 * @param value
	 *            Object to store
	 * @param field
	 *            String key
	 * @param add
	 *            if true for insertion mode, and false to erase mode
	 */ 

	public void modify(Id key, Object value, String field, boolean add)	{
		modify(Context.DEFAULT_CONTEXT,key,value,field,add);
	}

     /**
		 * Modifies the content of a specific field
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param value
		 *            Object to modify
		 * @param field
		 *            String key
		 * @param add
		 *            if true for insertion mode, and false to erase mode
		 */
	public void modify(String context, Id key, Object value, String field, boolean add) {

		if (isOwner(key))
		{
			try
			{
				Collection c = null;
				undoReplica(context,key);
				if (add)
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					c.add(value);
					storage.write(context,key,c,field);
					remoteListeners(context,key,value,field,true);

				}
				else
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					else c.remove(value);
					storage.write(context,key,c,field);
					remoteListeners(context,key,value,field,false);
				}
				// update replicas
				Values values = storage.extract(context,key);
				replicate(context,key,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else
		{
			Vector values = new Vector();
			values.add(value);
			// endPoint.route(key, new ModifyMessage(key,values,field,add),
			// null);
			ModifyMessage msg = new ModifyMessage(key,values,field,add);
			msg.setContext(context);
			endPoint.route(key,msg, null);
		}
	}

   /**
	 * Modifies the content of a specific field
	 * 
	 * @param key
	 *            Id of the value
	 * @param values
	 *            Collection to modify
	 * @param field
	 *            String key
	 * @param add
	 *            if true for insertion mode, and false to erase mode
	 */
	public void modify(Id key, Collection list, String field, boolean add)	{
		modify(Context.DEFAULT_CONTEXT,key,list,field,add);
	}

  /**
	 * Modifies the content of a specific field
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param values
	 *            Collection to modify
	 * @param field
	 *            String key
	 * @param add
	 *            if true for insertion mode, and false to erase mode
	 */
	public void modify(String context,Id key, Collection list, String field, boolean add)	{

		if (isOwner(key))
		{
			try
			{
				Collection c = null;
				undoReplica(context,key);
				if (add)
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					c.addAll(list);
					storage.write(context,key,c,field);
					remoteListeners(context,key,list,field,true);

				}
				else
				{
					c = (Collection) storage.retrieve(context,key,field);
					if (c==null) c = new Vector();
					else c.removeAll(list);
					storage.write(context,key,c,field);
					remoteListeners(context,key,list,field,false);
				}
				// update replicas
				Values values = (Values)storage.extract(context,key);
				replicate(context,key,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else
		{
			// endPoint.route(key, new ModifyMessage(key,list,field,add), null);
			ModifyMessage msg = new ModifyMessage(key,list,field,add);
			msg.setContext(context);
			endPoint.route(key,msg, null);
		}
	}

  /**
	 * Returns the value of the specific key using the Client application
	 * 
	 * @param key
	 *            Id of the value
	 * @param client
	 *            listener that implements BunshinClient
	 */
	public void get(Id key, BunshinGetClient client) {
		get(Context.DEFAULT_CONTEXT,key,Context.DEFAULT_FIELD,client);
	}

  /**
	 * Returns the values of the specific key and sets the Client application
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param client
	 *            listener that implements BunshinClient
	 */  
	public void get(String context, Id key, BunshinGetClient client) {
		get(context,key, Context.DEFAULT_FIELD,client);
	}

  /**
	 * Returns the values of the specific key and sets the Client application
	 * 
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param client
	 *            listener that implements BunshinClient
	 */ 
	public void get(Id key, String field, BunshinGetClient client) {
		get(Context.DEFAULT_CONTEXT,key,field,client);
	}

   /**
	 * Returns the values of the specific key and sets the Client application
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param client
	 *            listener that implements BunshinClient
	 */  
	public void get(String context,Id key, String field, BunshinGetClient client) {
		getSecure(context,key,field,null,client);
	}
	
  /**
	 * Returns the values of the specific key and sets the Client application
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param pass
	 *            password
	 * @param client
	 *            listener that implements BunshinClient
	 */  
	public void getSecure(String context,Id key, String field, String pass, BunshinGetClient client) {

        Object value = null;
        Values values = null;
        boolean found = false;
       
        Password pwd = null;
		if (pass!=null) pwd = new Password(pass);		
		
        
		if (isOwner(key)) {
		    
			// if (field!=null) value = storage.retrieve(context,key,field);
			// else value = storage.extract(context,key);
			values = storage.extract(context,key);
            if (values!=null) found = true;
			
		}
		// I'm not owner but I have replica
		else if (replicaKeyOwner.containsKey(key))
		{
			values = getReplicaValue(context,key);
			// if (field!=null && value instanceof Values) value =
			// ((Values)value).get(field);
			if (values!=null) found = true;
			
		}
		// I'm not owner and don't have any replica, but I have value-cache
		else if (cache.contains(context,key))
		{
			values = cache.get(context,key);
			// if (field!=null && value instanceof Values) value =
			// ((Values)value).get(field);
			if (values!=null) found = true;
		}		
		
		if (found) {		  
		   
            
		  Password oldPwd = (Password) values.get("Password");
		  if (field!=null && values!=null) value = values.get(field);
		  else value = values;
		  
		  boolean validate = true;
          // getcheck system
          if (oldPwd != null) {
          	System.out.println("Getcheck key "+key+": "+oldPwd.isGetCheck());
          	if (oldPwd.isGetCheck()) validate = (pwd!=null && oldPwd.equals(pwd));
          	else validate = true;
          }	
		  
		  if (validate)  client.get(value);	
		  else client.get(null);
		}
		
		else{

			Vector id_clients;

			if (!get_clients.containsKey(key)) id_clients = new Vector();
			else id_clients =  id_clients = (Vector) get_clients.get(key);

			id_clients.add(client);
			get_clients.put(key,id_clients);

			GetMessage msg = new GetMessage(key,field,pwd,endPoint.getLocalNodeHandle());
			msg.setContext(context);
			endPoint.route(key, msg, null);
		}
	}
	
	  /**
		 * Returns the value of the specific key if the values is stored in the
		 * local node
		 * 
		 * @param key
		 *            Id of the value
		 */   
	  public Object getLocal(Id key) {
		  return getLocal(Context.DEFAULT_CONTEXT,key,Context.DEFAULT_FIELD);
	  }
	  
	  /**
		 * Returns the value of the specific key if the values is stored in the
		 * local node
		 * 
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 */   
	  public Object getLocal(Id key, String field) {
		  return getLocal(Context.DEFAULT_CONTEXT,key,field);
	  }
	  
	  /**
		 * Returns the value of the specific key if the values is stored in the
		 * local node
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * 
		 */   
	  public Object getLocal(String context, Id key) {
		  return getLocal(context,key,Context.DEFAULT_FIELD);
	  }
	  
	  /**
		 * Returns the value of the specific key if the values is stored in the
		 * local node
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 * 
		 */   
	  public Object getLocal(String context, Id key, String field) {
		  return getLocalSecure(context,key,field,null);
	  }
	  
	 
	  /**
		 * Returns the value of the specific key if the values is stored in the
		 * local node
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 * @param pass
		 *            password
		 * 
		 */   
	  public Object getLocalSecure(String context, Id key, String field, String pass) {
		  
		    Object value = null;
	        Values values = null;
	        boolean found = false;
	       
	        Password pwd = null;
			if (pass!=null) pwd = new Password(pass);				
	        
			if (isOwner(key)) {
			    
				// if (field!=null) value = storage.retrieve(context,key,field);
				// else value = storage.extract(context,key);
				values = storage.extract(context,key);
	            if (values!=null) found = true;
				
			}
			// I'm not owner but I have replica
			else if (replicaKeyOwner.containsKey(key))
			{
				values = getReplicaValue(context,key);
				// if (field!=null && value instanceof Values) value =
				// ((Values)value).get(field);
				if (values!=null) found = true;
				
			}
			// I'm not owner and don't have any replica, but I have value-cache
			else if (cache.contains(context,key))
			{
				values = cache.get(context,key);
				// if (field!=null && value instanceof Values) value =
				// ((Values)value).get(field);
				if (values!=null) found = true;
			}			
		
			if (found) {		  
				   
	            
				  Password oldPwd = (Password) values.get("Password");
				  if (field!=null && values!=null) value = values.get(field);
				  else value = values;
				  
				  boolean validate = true;
		          // getcheck system
		          if (oldPwd != null) {
		          	System.out.println("Getcheck key "+key+": "+oldPwd.isGetCheck());
		          	if (oldPwd.isGetCheck()) validate = (pwd!=null && oldPwd.equals(pwd));
		          	else validate = true;
		          }	
				  
				  if (validate) return value;
				  else return null;
				}
			
			return null;
		  
	  }
	

	
	
  /**
	 * Returns the URL of the specific key
	 * 
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param client
	 *            listener that implements BunshinClient
	 */ 
	
	public void getURL(Id key, String field, BunshinURLClient client){
	  getURL(Context.DEFAULT_CONTEXT,key,field,client);
    }
  
   /**
	 * Returns the URL of the specific key
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param client
	 *            listener that implements BunshinClient
	 */
  public void getURL(String context, Id key, String field, BunshinURLClient client) {
           
           
       if (isOwner(key)) {		    
		
		  URL url = null;
          Values values = storage.extract(context,key);           
        
          String f = (String)values.get("#field");
          if (f!=null && f.equals(field)) {
           	
            String filename = (String)values.get("#filename");
		    if (filename!=null) {          
              String url_base = (String) urlContext.get(context);
              try{
                                       
              url = new URL(url_base+'/'+filename);
                           } catch(Exception e) {e.printStackTrace();
                        }
              client.get(url);	
            }
            else client.get(null);
           	
          } 
          else client.get(null);        
			
		}
		else{		

			Vector id_clients;

			if (!url_clients.containsKey(key)) id_clients = new Vector();
			else id_clients =  id_clients = (Vector) url_clients.get(key);

			id_clients.add(client);
			url_clients.put(key,id_clients);

			GetURLMessage msg = new GetURLMessage(key,field,endPoint.getLocalNodeHandle());
			msg.setContext(context);
			endPoint.route(key, msg, null);
		}	      	        
		      	
  	
    }	
        


	/**
	 * Removes the key/value of the Bunshin DHT
	 * 
	 * @param key
	 *            Id of the value
	 */ 
	public void remove(Id key)	{
		remove(Context.DEFAULT_CONTEXT,key,Context.DEFAULT_FIELD);
	}
   /**
	 * Removes the (context,key)/value of the Bunshin DHT
	 * 
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 */  
	public void remove(Id key, String field) {
		remove(Context.DEFAULT_CONTEXT,key,field);
	}

   /**
	 * Removes the (context,key)/value of the Bunshin DHT
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 */  
	public void remove(String context, Id key) {
		remove(context,key,Context.DEFAULT_FIELD);
	}
   /**
	 * Removes the (context,key,field)/value of the Bunshin DHT
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 */ 
	public void remove(String context,Id key, String field) {		
		removeSecure(context,key,field,null);
	}
	
	/**
	 * Removes the (context,key,field,pass)/value of the Bunshin DHT
	 * 
	 * @param context
	 *            Id of the context
	 * @param key
	 *            Id of the value
	 * @param field
	 *            String key
	 * @param pass
	 *            password
	 */ 
	public void removeSecure(String context,Id key, String field, String pass) {
		
		Password pwd = null;
		if (pass!=null) pwd = new Password(pass);		
		
		if (isOwner(key)) {
			
			try {    	      	      
		      
		      Password oldPwd = (Password) storage.retrieve(context,key,"Password");
		      
		      boolean validate = (oldPwd == null || (pwd!=null && oldPwd.equals(pwd)));		      	
		  
			  if (validate) {

				undoReplica(context,key);
				log("Removing "+key);
				
				if (field!=null) {
					Object value = storage.retrieve(context,key,field);
					remoteListeners(context,key,value,field,false);
					storage.delete(context,key,field);
			        
			        // is empty values?
					Values values = storage.extract(context,key);
					if (values!=null) {
					
                    if ( values!=null &&
                    	(values.size()==2&&values.containsKey("CVS")&&values.containsKey("Password"))
                        ||
                        (values.size()==1&&(values.containsKey("CVS")||values.containsKey("Password")))
                        ||
                        (values.size()==0)
                        )
					
					storage.remove(context,key);						
					}
				}
				else storage.remove(context,key);
			  }
			  else  log("Wrong validation of remove key : "+key); 
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
				// send local reject ack
			}
		}
		else {
			
			RemoveMessage msg =  new RemoveMessage(key,field,pwd,endPoint.getLocalNodeHandle());
			msg.setContext(context);
			endPoint.route(key,msg, null);
		}
	}
	
	 /**
		 * Removes the key/value of the Bunshin DHT
		 * 
		 * @param key
		 *            Id of the value
		 * @param client
		 *            listener BunshinRemoveClient
		 */  
	  public void remove(Id key, BunshinRemoveClient client) {
		  remove(Context.DEFAULT_CONTEXT,key,Context.DEFAULT_FIELD, client);
	  }

	   /**
		 * Removes the (context,key)/value of the Bunshin DHT
		 * 
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 * @param client
		 *            listener BunshinRemoveClient
		 */  
	  public void remove(Id key, String field, BunshinRemoveClient client) {
		  remove(Context.DEFAULT_CONTEXT,key, field, client);
	  }
	   /**
		 * Removes the (context,key)/value of the Bunshin DHT
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param client
		 *            listener BunshinRemoveClient
		 */  
	  public void remove(String context, Id key, BunshinRemoveClient client) {
		  remove(context,key,Context.DEFAULT_FIELD, client);
	  }

	   /**
		 * Removes the (context,key,field)/value of the Bunshin DHT
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 * @param client
		 *            listener BunshinRemoveClient
		 */  
	  public void remove(String context,Id key, String field, BunshinRemoveClient client) {
		  removeSecure(context,key,field, null, client);
	  }
	  
	   /**
		 * Removes the (context,key,field,pass)/value of the Bunshin DHT
		 * 
		 * @param context
		 *            Id of the context
		 * @param key
		 *            Id of the value
		 * @param field
		 *            String key
		 * @param pass
		 *            password
		 * @param client
		 *            listener BunshinRemoveClient
		 */  
	  public void removeSecure(String context,Id key, String field, String pass, BunshinRemoveClient client) {
		    Vector id_clients;

		    if (!remove_clients.containsKey(key)) id_clients = new Vector();
			else id_clients =  id_clients = (Vector) remove_clients.get(key);

			id_clients.add(client);
			remove_clients.put(key,id_clients);
				
			removeSecure(context,key,field,pass);
	  }

	
   /**
	 * Sets the listener of the specific key and this notifies about the
	 * incoming links
	 * 
	 * @param id
	 *            Id of the value
	 * @param listener
	 *            that implements RemoteListener listener
	 * @param field
	 *            String key
	 */
	public void setRemoteListener(Id id, RemoteListener listener, String field)	{
		setRemoteListener(Context.DEFAULT_CONTEXT,id,listener,field);
	}

  /**
	 * Sets the listener of the specific key and this notifies about the
	 * incoming links
	 * 
	 * @param context
	 *            Id of the context
	 * @param id
	 *            Id of the value
	 * @param listener
	 *            that implements RemoteListener listener
	 * @param field
	 *            String key
	 */
	public void setRemoteListener(String context,Id id, RemoteListener listener, String field)	{

		listeners.put(id,listener);

		if (isOwner(id))
		{
			try
			{
				undoReplica(context,id);
				Listeners ls = (Listeners) storage.retrieve(context,id,"Listeners");
				if (ls==null) ls = new Listeners();
				ls.add(endPoint.getLocalNodeHandle(),field);

				storage.write(context,id,ls,"Listeners");
				Values values = storage.extract(context,id);
				replicate(context,id,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
			}
		}
		else
		{
			// endPoint.route(id, new
			// ListenerMessage(id,field,endPoint.getLocalNodeHandle(),true),
			// null);
			ListenerMessage mesg = new ListenerMessage(id,field,endPoint.getLocalNodeHandle(),true);
			mesg.setContext(context);
			endPoint.route(id,mesg, null);
		}
	}

   /**
	 * Removes the listener of the specific key
	 * 
	 * @param id
	 *            Id of the value
	 * @param field
	 *            String key
	 */
	public void removeRemoteListener(Id id, String field)	{
		removeRemoteListener(Context.DEFAULT_CONTEXT,id,field);
	}
	
	   /**
		 * Removes the listener of the specific key
		 * 
		 * @param context
		 *            Id of the context
		 * @param id
		 *            Id of the value
		 * @param field
		 *            String key
		 */
	public void removeRemoteListener(String context, Id id, String field)	{

		listeners.remove(id);
		if (isOwner(id))
		{
			try
			{

				undoReplica(context,id);
				Listeners ls = (Listeners) storage.retrieve(context,id,"Listeners");
				if (ls!=null) ls.remove(endPoint.getLocalNodeHandle(),field);
				else ls = new Listeners();

				storage.write(context,id,ls,"Listeners");
				Values values = storage.extract(context,id);
				replicate(context,id,values);
			}
			catch(StorageException ex)
			{
				ex.printStackTrace();
				// activate threshold?
			}
		}
		else
		{
			// endPoint.route(id, new
			// ListenerMessage(id,field,endPoint.getLocalNodeHandle(),false),
			// null);
			ListenerMessage mesg = new ListenerMessage(id,field,endPoint.getLocalNodeHandle(),false);
			mesg.setContext(context);
			endPoint.route(id,mesg, null);
		}
	}



	/**
	 * Inserts in the structure of replicas a new (key,value) from a specific
	 * node
	 * 
	 * @param own
	 * @param key
	 * @param value
	 */
	private void putReplica(NodeHandle own, String context, Id key, Values value)
	{
		StorageManager replicas;
		try
		{
			if (replicaBuckets.containsKey(own))
			{
				replicas = (StorageManager) replicaBuckets.get(own);
			}
			else
			{
				replicas = new MemStorage(); // not persistence for replicas
			}
			replicas.write(context,key,value,null);

			log("Put Replica in context "+context+" key : "+key+" value "+value+" --> "+replicas.getBucket(context));

			replicaBuckets.put(own,replicas);
			replicaKeyOwner.put(key,own);

		}
		catch(StorageException ex)
		{
			ex.printStackTrace();

		}
	}

	/**
	 * Sends a replica of the pair (key,value) to replicationFactor nodes from
	 * replicaSet
	 * 
	 * @param key
	 * @param value
	 */
	private void replicate(String context, Id key, Values values)
	{
		// reset possible replicas
		replicaNHs.remove(context,key);

		NodeHandle nh = endPoint.getLocalNodeHandle();
		Collection c = Utilities.convert(endPoint.replicaSet(key,replicationFactor+1));
		if (c!=null)
		{

			Iterator it = c.iterator();
			while(it.hasNext())
			{

				NodeHandle to = (NodeHandle) it.next();
				if (!to.getId().equals(nh.getId()))
				{
					log("Sending replica to "+to.getId());
					// endPoint.route(null,new
					// ReplicaMessage(key,values,nh),to);
					ReplicaMessage msg = new ReplicaMessage(key,values,nh);

					msg.setContext(context);
					endPoint.route(null,msg,to);
				}
			}
		}
	}

	private void replicateURL(String context, Id key, Values values, URL url, String field)
	{

		// reset possible replicas
		replicaNHs.remove(context,key);

		NodeHandle nh = endPoint.getLocalNodeHandle();
		Collection c = Utilities.convert(endPoint.replicaSet(key,replicationFactor+1));
		if (c!=null)
		{
			Iterator it = c.iterator();
			while(it.hasNext())
			{
				NodeHandle to = (NodeHandle) it.next();
				if (!to.getId().equals(nh.getId()))
				{
					log("Sending replica to "+to.getId());
					// endPoint.route(null,new
					// ReplicaMessage(key,values,nh,url,field),to);
					ReplicaMessage msg = new ReplicaMessage(key,values,nh,url,field);
					msg.setContext(context);
					endPoint.route(null,msg,to);

				}
			}
		}
	}

	/**
	 * Sends remove replica of the all keys to replica nodes
	 * 
	 * @param key
	 */
	private void undoReplica(String context,Id key)
	{

		Collection c = replicaNHs.remove(context,key);
		if (c!=null) {

			Iterator it = c.iterator();
			while(it.hasNext())	{
				
				NodeHandle nh = (NodeHandle) it.next();
				log("Removing replica from "+nh.getId());
				// endPoint.route(null,new RemoveReplicaMessage(nh,key),nh);
				RemoveReplicaMessage msg = new RemoveReplicaMessage(nh,key);
				msg.setContext(context);
				endPoint.route(null,msg,nh);

			}
		}
	}


	/**
	 * Finds in the replicated buckets a specific value in the context bucket
	 * 
	 * @param context
	 * @param key
	 * @return Values values
	 */
	private Values getReplicaValue(String context, Id key)
	{

		boolean found = false;
		Values values = null;

		Iterator it = replicaBuckets.values().iterator();

		while(it.hasNext() && !found)
		{
			StorageManager s = (StorageManager) it.next();
			values = s.extract(context,key);
			found = (values!=null);
		}
		return values;
	}



	private void remoteListeners(String context, Id key, Object value, String field, boolean add)
	{
		Listeners ls = (Listeners) storage.retrieve(context,key,"Listeners");
		if (ls!=null && field!=null)
		{
			Collection c = ls.getSubscribers(field);
			if (c!=null)
			{
				Iterator it = c.iterator();
				while (it.hasNext())
				{
					NodeHandle destiny = (NodeHandle) it.next();
					// endPoint.route(null, new
					// NotifyMessage(key,value,field,true), destiny);
					NotifyMessage msg = new NotifyMessage(key,value,field,true);
					msg.setContext(context);
					endPoint.route(null, msg, destiny);

				}
			}
		}
	}
	
	private boolean CVS(Object v1, Object v2) {
		
	  Values values = (Values) v1;
	  Values oldValues = (Values) v2;
	  
	 // System.out.println("CVS : new:"+v1+" vs old:"+v2);
	  	
	  if (values!=null && values.get("CVS")!=null) { // new version with
														// version
	    
	  	int version = ((Integer) values.get("CVS")).intValue();
	  	
	  	if (oldValues!=null && oldValues.get("CVS")!=null) { // exists old
																// version ?
	  	  
	  	  int oldVersion = ((Integer) oldValues.get("CVS")).intValue(); 
	  	  
	  	// System.out.println("CVS : new:"+version+" vs old:"+oldVersion);
	  	  
	  	  return version>oldVersion; 		  	  
	  	}	  	
	  	else {
	  	  values.put("CVS",new Integer(0)); // first version
	  	  // System.out.println("CVS : new version");
	  	  return true;
	  	}  
	  }	
	  else { // new value without version
	  
	  	if (values==null) return false;
	  	
	  	if (oldValues!=null && oldValues.get("CVS")!=null) { // exists old
																// version ?
	  		
	  	  int oldVersion = ((Integer) oldValues.get("CVS")).intValue(); 
	  	  
	  	  int newVersion = oldVersion+1;
	  	  
	  	  values.put("CVS",new Integer(newVersion));  		  	  
	  	}	  	
	  	else values.put("CVS",new Integer(0));
	  	
	  	// System.out.println("CVS : no version -> new version");
	  	
	  	return true;	  	
	  }	
	}

	private synchronized void refresh(int mode)
	{
		if ( mode==0)
		{
			checkMyKeys();
			checkMyRemoteReplicas();
			checkReplicas();
		}
		else if (mode==1)
		{
			checkMyKeys();
		}
		else
		{
			checkMyRemoteReplicas();
			checkReplicas();
		}

	}
	
	
	class BunshinAutoPutClient implements BunshinPutClient {
   	  	
   	  	String context;   	  	
   	  	Id key;
   	  	
   	  	public BunshinAutoPutClient(String context, Id key) {
   	  	  this.context = context;
   	  	  this.key = key;
   	  	}
   	  	
  	    public void put(boolean ack, NodeHandle owner) {  	  	
  	      if (ack) {
  	       	try {
  	       		Values values = storage.extract(context, key);
  	       		storage.remove(context,key);  	       	
  	       	    if (owner!=null && values!=null) putReplica(owner,context,key,values);
  	       	} catch(Exception e) {
  	       		e.printStackTrace();
  	       	}  	         
  	      }
  	    	
	    }
    }
    
    

	class PeriodicTask extends TimerTask
	{

		public void run()
		{
			// keys and replicas
			refresh(0);

			// caches
			if (caching)
			{
				cache.incTime();

				// caching?
				Iterator it = control.keySet().iterator();
				// for each context
				while (it.hasNext())
				{
					String context = (String) it.next();
					Hashtable pairs = (Hashtable) control.get(context);
					if (pairs!=null)
					{
						Iterator it2 = pairs.keySet().iterator();

						// for each key
						while (it2.hasNext())
						{
							Id key = (Id) it2.next();
							Values values = null;

							if (isOwner(key))
							{
								values = storage.extract(context,key);
							}
							else if (replicaKeyOwner.containsKey(key))
							{
								values = getReplicaValue(context,key);
							}
							else if (cache.contains(context,key))
							{
								values = cache.get(context,key);
							}

							PriorityList pl = (PriorityList) pairs.get(key);
							log("cache key "+key+" max request : "+pl.getMax());
							if (pl.getMax()>MAX_REQUEST_TIME)
							{

								NodeHandle nh = (NodeHandle) pl.peak();
								log("send CacheMessage (key : "+key+", values : "+values);
								// endPoint.route (null, new
								// CacheMessage(key,values), nh);
								CacheMessage msg = new CacheMessage(key,values);
								msg.setContext(context);
								endPoint.route (null, msg, nh);
							}
							pl.clear();
						}
					}
				}
			}
		}

	}




}

