messagedispatcher.java

Java Communicating Agents是一个用于开发网络反应式信息agent的框架

/*
 * $Source: /home/data/cvsroot/src/jacomma/platform/engine/MessageDispatcher.java,v $
 * $Revision: 1.7 $
 * $Date: 2000/10/28 20:09:08 $
 *
 * This file is part of the jacomma framework
 * Copyright (c) 2000 	Dimitrios Vyzovitis
 *			mailto:[e-mail]
 *			
 *
 *			
 *			
 *			
 *
 *	This library is free software; you can redistribute it and/or modify
 *	it under the terms of the GNU Library General Public License as published by
 *	the Free Software Foundation; either version 2 of the License, or
 *	(at your option) 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 Library General Public License for more details.
 *
 *	You should have received a copy of the GNU Library 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 jacomma.platform.engine;

import jacomma.icm.type.Message;
import jacomma.platform.core.Agent;
import jacomma.platform.core.MessagePattern;
import jacomma.util.Cache;
import jacomma.util.Visitor;

import java.util.List;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Set;
import java.util.Iterator;

/**
 * TBA
 **/
class MessageDispatcher implements Runnable, Visitor {
	
	Agent agent_;
	MessageQueue mque_;
	HashMap pkeys_;
	HashMap hdls_;
	
	Thread thread_;
	boolean active_;

	MessageDispatcher( MessageQueue mque ) {
		mque_ = mque;
		pkeys_ = new HashMap(); // use the hdl_ monitor
		hdls_ = new HashMap();
		// agent must be set externally (circ. reference)
	}
	
	// Handler management
	public void registerHandler( MessagePattern ptrn, 
											  Agent.MessageHandler hdlr ) {
		synchronized ( hdls_ ) {

			for ( Iterator it = ptrn.keys().iterator(); it.hasNext(); ) {
				String key = (String)it.next();
				List pl = (List)pkeys_.get( key );
				if ( pl == null ) {
					pl = new ArrayList();
					pkeys_.put( key, pl );
				}
				
				pl.add( ptrn );
				hdls_.put( ptrn, hdlr );
			}
			
		}

		// notify the dispatcher to return-examine the queue
		mque_.signal();
	}

	public void unregisterHandler( MessagePattern ptrn, 
												Agent.MessageHandler hdlr ) {
		synchronized ( hdls_ ) {
			for ( Iterator it = ptrn.keys().iterator(); it.hasNext(); ) {
				List pl = (List)pkeys_.get( it.next() );
				if ( pl != null ) pl.remove( ptrn );
			}
			
			hdls_.remove( ptrn );
		}
	}
	
	public void unregisterHandler( Agent.MessageHandler hdlr ) {
		synchronized ( hdls_ ) {
			for ( Iterator it = hdls_.entrySet().iterator();
				  it.hasNext(); ) {
				Map.Entry next = (Map.Entry)it.next();
				if ( hdlr.equals( next.getValue() ) ) {
					unregisterHandler( (MessagePattern)next.getKey(), 
									   (Agent.MessageHandler)next.getValue() );
					break;
				}
			}
		}
	}

	public Map handlers() {
		synchronized ( hdls_ ) {
			return (Map)hdls_.clone();
		}
	}

	public void addHandlers( Map map ) {
		synchronized ( hdls_ ) {
			for ( Iterator it = map.entrySet().iterator();
				  it.hasNext(); ) {
				Map.Entry next = (Map.Entry)it.next();
				registerHandler( (MessagePattern)next.getKey(), 
							 (Agent.MessageHandler)next.getValue() );
			}
		}
			
	}


	public Map setHandlers( Map map ) {
		synchronized ( hdls_ ) {
			Map old = clearHandlers();
			addHandlers( map );
			
			return old;
		}
	}

	public Map clearHandlers() {
		synchronized ( this ) {
			Map m = (Map)hdls_.clone();
			hdls_.clear();
			pkeys_.clear();
		
			return m;
		}
	}

	// Visitor Interface
	public boolean canVisit( Class c ) {
		return c.equals( List.class );
	}

	public Object visit( Object q ) {
		for ( Iterator it = ((List)q).iterator();
			  it.hasNext(); ) {
			MessageQueue.Entry e = (MessageQueue.Entry)it.next();
			for ( Iterator kit = e.patterns().iterator();
				  kit.hasNext(); ) {
				String key = (String)kit.next();
				
				synchronized ( hdls_ ) {
					List pl = (List)pkeys_.get( key );
					if ( pl != null ) for ( Iterator pit = pl.iterator();
											pit.hasNext(); ) {
						MessagePattern pat = (MessagePattern)pit.next();
						if ( pat.matches( e.getMessage() ) ) {
							it.remove();
							return new Rule( e.getMessage(), 
											 (Agent.MessageHandler)hdls_.get( pat ) );
						}
					}
				}
				
			}
		}
		
		return null;
	}
	
	// Dispatcher stuff
	synchronized Thread startDispatcher( ThreadGroup tg ) {
		if ( !active_ ) {
			active_ = true;
			(thread_ = new Thread( tg, this, toString() + " - Dispatcher Thread" )).start();
			return thread_;
		} else return null;
	}
	
	synchronized void stopDispatcher() {
		if ( active_ ) {
			active_ = false;
			thread_.interrupt();
		}
	}
	
	synchronized boolean isActive() { 
		return active_ && !thread_.isInterrupted(); 
	}
	
	// Dispatched code
	public void run() {
		while( isActive() ) try {
			Rule r = (Rule)mque_.accept( this );
			if ( r != null ) r.exec();
			// Attempt to fix the 100% CPU utilization behavior incurred by LinuxThreads
			Thread.currentThread().yield();
		} catch ( Exception exc ) {
			thread_.getThreadGroup().uncaughtException( thread_, exc );
		}
	}
	
	class Rule {
			
		Message msg_;
		Agent.MessageHandler hdlr_;

		Rule( Message msg, Agent.MessageHandler hdlr ) {
			hdlr_ = hdlr;
			msg_ = msg;
		}

		void exec() {
			hdlr_.onMessage( agent_, msg_ );
		}
		
	}

}