agentimp.java

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

		Message rq = Request.createCancelForwardRequest( id, handle_, to, at );
		_doUnFwd( id, to, rq );
	}

	public synchronized int getRegisterCount() { return rcount_; }


	synchronized void _doRegister( Handle id, Message rq ) 
		throws RequestFailedException {
		try {

			mon_.onRegistering( this, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmRegisterAgent, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				// update our handle + add the host to the registered list
				handle_ = (Handle)cnt.get( 2 ); 
				Handle cs = resp.getSender();
				orgcs_.add( cs );
				rgcs_.put( cs, cs.getHome() );
				oatcs_.add( cs );
				atcs_.put( cs, cs.getHome() );
				rcount_++; // update count before notifying
				mon_.onRegistered( this, cs.getHome() );
			} else 
				throw new RequestFailedException( Request.icmRegisterAgent, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmRegisterAgent, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmRegisterAgent, exc );
		}
	}

	synchronized void _doDeregister( Handle id, Message rq ) 
		throws RequestFailedException {
		try {

			mon_.onDeregistering( this, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmDeregisterAgent, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				handle_ = (Handle)cnt.get( 2 ); 
				Handle cs = resp.getSender();
				orgcs_.remove( cs );
				rgcs_.remove( cs );
				oatcs_.remove( cs );
				atcs_.remove( cs );
				odetcs_.remove( cs );
				detcs_.remove( cs );
				rcount_--; // update before notifying
				mon_.onDeregistered( this, cs.getHome() );
			} else 
				throw new RequestFailedException( Request.icmRegisterAgent, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmRegisterAgent, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmRegisterAgent, exc );
		}
	}
	
	synchronized void _doAttach( Handle id, Message rq ) 
		throws RequestFailedException {
		try {

			mon_.onAttaching( this, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmAttachAgent, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				handle_ = (Handle)cnt.get( 2 ); 
				Handle cs = resp.getSender();
				InetAddress cshost = cs.getHome();
				orgcs_.add( cs );
				rgcs_.put( cs, cshost );
				oatcs_.add( cs );
				atcs_.put( cs, cshost );
				odetcs_.add( cs );
				detcs_.remove( cs );
				mon_.onAttached( this, cshost );
			} else 
				throw new RequestFailedException( Request.icmAttachAgent, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmAttachAgent, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmAttachAgent, exc );
		} 
	}
		
	synchronized void _doDetach( Handle id, Message rq ) 
		throws RequestFailedException {
		try {

			mon_.onDetaching( this, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmDetachAgent, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				handle_ = (Handle)cnt.get( 2 ); 
				Handle cs = resp.getSender();
				InetAddress cshost = cs.getHome();
				oatcs_.remove( cs );
				atcs_.remove( cs );
				odetcs_.add( cs );
				detcs_.put( cs, cshost );
				mon_.onDetached( this, cshost );
			} else 
				throw new RequestFailedException( Request.icmDetachAgent, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmDetachAgent, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmDetachAgent, exc );
		}
	}
	
	synchronized void _doFwd( Handle id, Handle to, Message rq ) 
		throws RequestFailedException {
		try {
			mon_.onFwding( this, to, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmRequestForwarding, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				handle_ = (Handle)cnt.get( 2 ); 
				InetAddress cshost = resp.getSender().getHome();
				fwds_.put( cshost, to );
				mon_.onFwded( this, to, cshost );
			} else 
				throw new RequestFailedException( Request.icmRequestForwarding, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmRequestForwarding, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmRequestForwarding, exc );
		}
	}
	
	synchronized void _doUnFwd( Handle id, Handle to, Message rq ) 
		throws RequestFailedException {
		try {
			mon_.onUnFwding( this, to, rq.getRecipient().getHome() );
			Message resp = 
				sendAndWait( rq, 
							  RequestPattern.createListPattern( id, 
																Request.icmCancelForwarding, 
																EngineUtilities.makeIdentifierTest( id ) ) );
			List cnt = (List)resp.getContent();
			if ( cnt.get( 1 ).equals( Request.icmOk ) ) {
				handle_ = (Handle)cnt.get( 2 ); 
				InetAddress cshost = resp.getSender().getHome();
				fwds_.remove( cshost );
				mon_.onUnFwded( this, to, cshost );
			} else 
				throw new RequestFailedException( Request.icmCancelForwarding, rq.toString() );
		} catch ( java.io.IOException exc ) {
			throw new RequestFailedException( Request.icmCancelForwarding, exc );
		} catch ( InterruptedException exc ) {
			throw new RequestFailedException( Request.icmCancelForwarding, exc );
		}
	}

	
	// Message management
	public Object accept( Visitor v ) throws InterruptedException {
		return que_.accept( v );
	}

	public Object try_accept( Visitor v ) {
		return que_.try_accept( v );
	}

	public Message recv() throws InterruptedException {
		return que_.get();
	}

	public Message try_recv() {
		return que_.try_get();
	}
	
	public Message recv( MessagePattern ptrn ) throws InterruptedException {
		Message m = null;
		Visitor pm = new PatternMatch( ptrn );
		for ( m =(Message)que_.accept( pm );
			  m == null;
			  m =(Message)que_.accept( pm ) );
		return m;
	}

	public Message try_recv( MessagePattern ptrn ) {
		return (Message)que_.try_accept( new PatternMatch( ptrn ) );
	}

	public void send( Message msg ) throws java.io.IOException {
		outp_.put( msg );
	}

	public void send( Handle to, Object cnt ) throws java.io.IOException {
		outp_.put( new Message( handle_, to, cnt ) );
	}

	public void send( Collection to, Object cnt ) throws java.io.IOException {
		// multicast is not yet supported byte the icm protocol - so we
		// have to send each message indiviually (argh)
		for ( Iterator it = to.iterator(); it.hasNext(); send( (Handle)it.next(), cnt ) );
	}

	// Message handlers for the agent (reactive behavior)
	public void registerHandler( MessagePattern ptrn, MessageHandler hdlr ) {
		dsp_.registerHandler( ptrn, hdlr );
	}

	public void unregisterHandler( MessagePattern ptrn, MessageHandler hdlr ) {
		dsp_.unregisterHandler( ptrn, hdlr );
	}

	public void unregisterHandler( MessageHandler hdlr ) {
		dsp_.unregisterHandler( hdlr );
	}
	
	public Map handlers() {
		return dsp_.handlers();
	}

	public void addHandlers( Map map ) {
		dsp_.addHandlers( map );
	}

	public Map setHandlers( Map map ) {
		return dsp_.setHandlers( map );
	}

	public Map clearHandlers() {
		return dsp_.clearHandlers();
	}
	
	class PatternMatch implements Visitor {
		MessagePattern ptrn_;
		
		PatternMatch( MessagePattern ptrn ) {
			ptrn_ = ptrn;
		}
		
		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 entry = (MessageQueue.Entry)it.next();
				if ( ptrn_.matches( entry.getMessage() ) ) {
					it.remove();
					return entry.getMessage();
				}
				
			}
			
			return null;
		}
	}

}