/* Learning Net - shlomy boshy 031868912
 * Reinforcement Learning network simulation  
 */
package learnnet;

public class Node extends Thread {
	protected Queue incomingMsgQueue; /* queue is synchronized */
	protected Queue nodeQueue; /* queue is synchronized */
	protected lookupTable Q;  /* lookup table is synchronized */
	protected Router R;
  public BellmanFordEstimator BF;
  public Network NT; 
    /* used only to determine network start */
  protected String routerType;
  public int nodeNumber;
  
  final static private int NONE=-1;	
      /* indicates no reply received yet */
	private boolean waitForReply=false;
  
  Node(Network NT,int nodeNumber) {
    this.NT=NT;    
    this.nodeNumber=nodeNumber;	       
    incomingMsgQueue = new FIFOQueue();
      /* the short-term queue until reply */    
    nodeQueue = new FIFOQueue(); 
       /* queue for waiting until routing */
    
    if (NT.lookupTableType.equals("ARRAY")) 
              Q = new ARRAYlookupTable(NT,this);            
     else if (NT.lookupTableType.equals("HASHTABLE")) 
         Q = new HASHlookupTable();   
     else throw new java.util.NoSuchElementException();         
      
    if ((NT.routerType.equals("QROUTING"))
        || (NT.routerType.equals("DUALQROUTING")) )
            R=new QROUTINGRouter();
     else if (NT.routerType.equals("SHORTEST_PATHS")) 
            R=new BellmanFordRouter();    
     else throw new java.util.NoSuchElementException();         
  }
	
public void	run(){
		
    Message M;      
    double reply;
	  Node N;	 

   if (NT!=null)
     while (!NT.networkStarted)  {
         try {sleep(50);} catch(InterruptedException e){break;};
    } 
 
    Network.debugPrint("Node "+this+":Starting to work on packets...");
    		
	while (!isInterrupted())
	{		     
		/* take packet from queue  */    
		 M = nodeQueue.getMessage();      
     if (M==null) continue;  /* queue empty */
     if (Network.Debug_Nodes)
       Network.debugPrint("Node "+this+" :packet "+(String)M.Data+" from "+M.sender+" with dest "+M.dest+" was read from queue");    
		  /* select neighbor to route through */		
      if (Network.Debug_Nodes)
         Network.debugPrint("Node "+this+" :get routing for send");
		 N = R.getBestRouting(Q,M.sender,M.dest,this); 
		  /* send packet to it */		
     if (Network.Debug_Nodes)
        Network.debugPrint("Node ="+this+" :packet "+(String)M.Data+" sent through "+N+"(with queue time here="+M.queueTime+")");
     M.transferTime = Q.getTransferTimeTo(N);    
     M.totalTime += M.transferTime + M.queueTime;
       /* adding queue time of current + transfer time to next */
       
     if ((NT.routerType.equals("DUALQROUTING")) && (M.sender!=this) &&(M.sender!=M.source)){       
        /* get estimate from M.sender to M.source (backwards) */
        if (Network.Debug_Routing)
          Network.debugPrint("Node "+this+":DUAL-QROUTING:getting backwards estimate for source "+M.source+" through neighbor "+M.sender);
        reply = M.sender.Q.getValue(M.sender.R.getBestRouting(M.sender.Q,this,M.source,M.sender),M.source); 
        if (Network.Debug_Routing)
          Network.debugPrint("Node "+this+":DUAL-QROUTING:reply is "+reply+" for estimate to source "+M.source+" from neighbor "+M.sender+"with transfer time "+M.transferTime+" and queue time on sender= "+M.queueTimeOnSender);
        R.updateFromReply(Q,reply,M.transferTime,M.queueTimeOnSender,M.sender,M.source,this);             
      }  
       
     if (!NT.routerType.equals("SHORTEST_PATHS"))        
        waitForReply=true;    
                     
		 M.sendMessage(this,N,Message.TYPE_DATA);			              
     while (waitForReply) ;     
	}	
		if (Network.Debug_Nodes)
      Network.debugPrint("Node "+this+"ended.");
	
}	
		
	

  public void putMessage(Message M) {   
    /* puts message into incoming-msg queue.
     *  activated by other nodes 
     */    
    incomingMsgQueue.putMessage(M);	   	
  }
   

  protected void noteMessage() {
  	/* get a message from incoming queue:
  	 * if reply -> update the reply
  	 * else send a reply and put in routing queue
  	 */ 
  	 
  	Node N;
  	double val;
  	Message newM;
  	double reply;
  	Message M = incomingMsgQueue.getMessage();    
    
    if (M == null) return;
    if (Network.Debug_Nodes)
       Network.debugPrint("Node "+this+" :received packet "+(String)M.Data+" from "+M.sender+"with dest "+M.dest);    
            
    if (NT.routerType.equals("SHORTEST_PATHS")){
       /* simpler routine (no reply needed) */
       noteMessageNoReplies(M);  
       return;
    }                                   
        
  	if ( M.type == Message.TYPE_REPLY) {
  		/* this is a reply -> update lookup table by reply */                        
      reply = new Double((String)M.Data).doubleValue();            
		  R.updateFromReply(Q,reply,M.transferTime,M.queueTime,M.sender,M.dest,this);
      waitForReply=false;
  	}
  	else { 
  	/* routing message - send estimate as reply 
  	 * and put in queue 
  	 */     
      if (Network.Debug_Nodes)
         Network.debugPrint("Node "+this+" :get routing for reply");
      if (M.dest != this) {
  	    N = R.getBestRouting(Q,M.sender,M.dest,this);        
  	    val = Q.getValue(N,M.dest);
      }
      else val =0; /* reached dest -> estimate to dest=0 */
  	  newM = new Message(null,this,M.dest,new String(String.valueOf(val)),Message.TYPE_REPLY);  	              
      newM.transferTime = M.transferTime;
  	  /* send reply back */      
      if (Network.Debug_Nodes)
           Network.debugPrint("Node "+this+" :sending reply "+ (String)M.Data+" to sender "+M.sender);                   
      newM.queueTime = (double) nodeQueue.numItems();     
      M.queueTimeOnSender = M.queueTime; /* last queue time */
      M.queueTime = newM.queueTime;      
      if (Network.Debug_Routing)          
        Network.debugPrint("Node "+this+":Message "+(String)M.Data+"received with queue size="+M.queueTime);     
  	  newM.sendMessage(this,M.sender,Message.TYPE_REPLY); 
      
      if (M.dest != this) /* not final dest */
         nodeQueue.putMessage(M);         
      else { 
            PacketCreator.addToTotalTime(M.totalTime);                        
         /* no queueing time in dest */
            if (Network.Debug_Routing)
              Network.debugPrint("Packet "+(String)M.Data+" from "+M.source+" to "+M.dest+" reached Total time: "+M.totalTime);
           }
    }  
    
  }  
     
  protected void addNeighbor(Node N,double transferTime){   
    Q.addNeighbor(N,transferTime);
  }
  
  protected void noteMessageNoReplies(Message M) {  
     /** for shortest paths routing */     
     M.queueTimeOnSender = M.queueTime; /* last queue time */     
     M.queueTime = (double) nodeQueue.numItems();     
     if (Network.Debug_Routing)          
        Network.debugPrint("Node "+this+":Message "+(String)M.Data+"received with queue size="+M.queueTime);
     if (M.dest != this) /* not final dest */
         nodeQueue.putMessage(M);         
     else { 
            PacketCreator.addToTotalTime(M.totalTime);            
         /* no queueing time in dest */
            if (Network.Debug_Routing)
              Network.debugPrint("Packet "+(String)M.Data+" from "+M.source+" to "+M.dest+" reached Total time: "+M.totalTime);
           }   
    }
 }