messagedispatcher.java

ALGAE是一个快速创建算法演示的框架。目前支持的算法实现语言包括java和c

/**
 * MessageDispatcher.java
 *
 *
 * Created: Fri Apr 17 22:14:01 1998
 *
 * @author 
 * @version
 */

package edu.odu.cs.zeil.AlgAE.Client;

import edu.odu.cs.zeil.AlgAE.Debug;
import edu.odu.cs.zeil.AlgAE.Queue;
import edu.odu.cs.zeil.AlgAE.Termination;

import edu.odu.cs.zeil.AlgAE.Message;
import edu.odu.cs.zeil.AlgAE.Messages;
import edu.odu.cs.zeil.AlgAE.MessageIO;
import edu.odu.cs.zeil.AlgAE.ParameterizedMessage;

import edu.odu.cs.zeil.AlgAE.Client.Displayer;
import edu.odu.cs.zeil.AlgAE.Client.GraphicsInfo;

import java.util.Vector;

public class MessageDispatcher 
  extends Thread
{
  private MessageIO io;
  private MessageIO logOut;
  
  private Queue deferredMsgs;
  private Queue priorityMsgs;
  private int forcingMessageCount;
  private boolean canContinue;
  private Reader reader;
  private Displayer displayer;
  private GraphicsInfo graphics;
  private boolean quitting;
  private int interpolationInterval = 250;
  private boolean waiting = false;
  private boolean hasBeenPainted = false;    
    

  public MessageDispatcher(MessageIO mainIO, MessageIO log)
  {
    // super (application.algaeThreads, "ClientMessageDispatcher");
    super ("ClientMessageDispatcher");
    
    io = mainIO;
    logOut = log;

    deferredMsgs = new Queue();
    priorityMsgs = new Queue();
    forcingMessageCount = 0;
    canContinue = true;
    reader = null;
    graphics = new GraphicsInfo ();
    quitting = false;
    Debug.watchForGC();
  }



  /**
   *   Send a message to the AlgAE server
   */
  public void send (ParameterizedMessage pmsg)
  {
    io.send (pmsg);
    if (logOut != null)
      logOut.send (pmsg);
  }
  

    /**
     *  Return the graphics information for this client.
     */
  public GraphicsInfo getGraphics() {return graphics;}
		     

  public int getInterpolationInterval()
  {
      return interpolationInterval;
  }
    
  public void setInterpolationInterval(double time)
  {
      interpolationInterval = (int)(time * 1000.0);
  }
    

  class Reader extends Thread 
  {
    
    public Reader()
    {
      // super (application.algaeThreads, "Client Message Reader");
      super ("Client Message Reader");
    }
    
    public void run()
    {
      while ((!quitting) && readMessage()) {}
    }
  }





  public boolean readMessage()
  {
    final int queueLimit = 4;  // suspend reading if we already have this
                               // many complete frames in the queue
    while (forcingMessageCount >= queueLimit)
	{
	    try {
		sleep (interpolationInterval);
	    } catch (java.lang.InterruptedException e) {}
	}
    
    ParameterizedMessage pmsg = io.read();
    Debug.show (Debug.clientread, "Read msg: " + pmsg);
    
    if (pmsg != null)
      {
	deferMessage (pmsg);
      }
    return (pmsg != null);
    
  }


    private synchronized void restart()
    {
	if (waiting)
	    {
	      waiting = false;
	      notifyAll();
	    }
    }
    
    public synchronized void requestDisplay()
    {
	graphics.setInterpolationStep (0);
	restart();
    }
    

  public void run()
  {
    reader = new Reader();
    // appl.register (reader);
    reader.start();

    boolean lastWasNotForcing = true;
    
    while (!quitting)
      {
	synchronized (this) {
	  if (Debug.messageSynch)
	    Debug.show (Debug.messageSynch, "dispatcher: fMC="
		+ forcingMessageCount + " canC=" + canContinue 
		+ " quitting=" + quitting
		+ " g.iC=" + graphics.interpolationCompleted()
		+ " q.size=" + deferredMsgs.size());
	  if (((forcingMessageCount == 0) || (!canContinue))
	      && (!quitting)
	      && (priorityMsgs.isEmpty())
	      && (graphics.interpolationCompleted()))
	    {
	      Debug.show (Debug.messageSynch, "pausing dispatcher");
	      try {
		  waiting = true;
		  wait();
	      }
	      catch (java.lang.InterruptedException e) {
		  quitting = true;
	      }
	    }
	}      
	if (!quitting)
	  {
	    if (!priorityMsgs.isEmpty())
	      {
		ParameterizedMessage pmsg;
		synchronized (this) {
		  pmsg = (ParameterizedMessage)priorityMsgs.getFront();
		  priorityMsgs.removeFront();
		}
		Message msg = pmsg.getMessage();
		Debug.show (Debug.messageSynch, "dispatching priority: " + pmsg);
	
		if (logOut != null)
		    logOut.send (pmsg);
	
		msg.received (pmsg.getParams());
	      }
	    else if (!graphics.interpolationCompleted())
	      {
		 synchronized (this) {
		     graphics.advanceInterpolation();
		     if (graphics.doubleBuffering)
			 displayer.display(graphics);
		     hasBeenPainted = false;
		 }
		 graphics.getCanvas().repaint();
		 while (!hasBeenPainted) {
		     try {
			 sleep (interpolationInterval);
		     } catch (java.lang.InterruptedException e) {}
		 }
	      }
	    else if (forcingMessageCount > 0)
	      {
		boolean lastWasForcing = false;
		while (!lastWasForcing)
		  {
		    ParameterizedMessage pmsg;
		    synchronized (this) {
		      pmsg = (ParameterizedMessage)deferredMsgs.getFront();
		      Message msg = pmsg.getMessage();
		      if (msg.isForcing())
		        {
			  --forcingMessageCount;
			  lastWasForcing = true;
			  // System.gc();
			  // System.runFinalization();
			}
		      deferredMsgs.removeFront();

		      Debug.show (Debug.messageSynch, "dispatching: " + pmsg);
	
		      if (logOut != null)
			  logOut.send (pmsg);
	
		      msg.received (pmsg.getParams());
		    }
		  }
	      }
	  }
      }
  }



  public synchronized void addPriorityMessage (ParameterizedMessage pmsg)
    {
	priorityMsgs.addToRear (pmsg);
	restart();
    }
    
	


  public synchronized void exit ()
  {
    quitting = true;
    restart();
  }
    

  public synchronized void permitContinuation ()
  {
    if (!canContinue)
      {
	Debug.show (Debug.messageSynch, "permitContinuation");
	canContinue = true;
	restart();
      }
  }
  

  public synchronized void denyContinuation ()
  {
    canContinue = false;
  }
  

  private synchronized void waitForPermissionToContinue ()
  {
    if (!canContinue)
      {
	try {
	  Debug.show (Debug.messageSynch, "waiting for continue signal");
	  wait();
	  Debug.show (Debug.messageSynch, "allowed to continue");
	}
	catch (java.lang.InterruptedException e) {
	}
      }
    else
      Debug.show (Debug.messageSynch, "can continue without waiting");
  }  

  private synchronized void waitForForcingMessage ()
  {
    if (forcingMessageCount <= 0) 
      {
	try {
	  Debug.show (Debug.messageSynch, "waiting for a marked message");
	  wait();
	  Debug.show (Debug.messageSynch, "marked message detected");
	}
	catch (java.lang.InterruptedException e) {
	}
    }
    else
      Debug.show (Debug.messageSynch, "no need to wait for marked message");
  }


  private synchronized void deferMessage (ParameterizedMessage pmsg)
  {
    deferredMsgs.addToRear (pmsg);
    if (pmsg.getMessage().isForcing()) {
      ++forcingMessageCount;
      restart();
    }
  }


  public void setDisplayer (Displayer d)
  {
    displayer = d;
  }
    

    public synchronized void painted()
    {
	hasBeenPainted = true;
    }
    
 } // MessageDispatcher