ddbaseimpl.java

基于JXTA开发平台的下载软件开发源代码

			// TODO: max values?
		
		dht.get(	
			((DDBaseKeyImpl)key).getBytes(), 
			key.getDescription(),
			(byte)0, 
			256, 
			timeout, 
			exhaustive,
			new listenerMapper( listener, DistributedDatabaseEvent.ET_VALUE_READ, key, timeout, exhaustive ));
	}
	
	public void
	delete(
		final DistributedDatabaseListener		listener,
		final DistributedDatabaseKey			key )
	
		throws DistributedDatabaseException
	{
		throwIfNotAvailable();
		
		dht.remove( ((DDBaseKeyImpl)key).getBytes(),
					key.getDescription(),
					new listenerMapper( listener, DistributedDatabaseEvent.ET_VALUE_DELETED, key, 0, false ));
	}
	
	public void
	addTransferHandler(
		final DistributedDatabaseTransferType		type,
		final DistributedDatabaseTransferHandler	handler )
	
		throws DistributedDatabaseException
	{
		throwIfNotAvailable();
		
		final HashWrapper	type_key = DDBaseHelpers.getKey( type.getClass());
		
		if ( transfer_map.get( type_key ) != null ){
			
			throw( new DistributedDatabaseException( "Handler for class '" + type.getClass().getName() + "' already defined" ));
		}
		
		transfer_map.put( type_key, handler );
		
		dht.registerHandler(
			type_key.getHash(),
			new DHTPluginTransferHandler()
			{
				public byte[]
				handleRead(
					DHTPluginContact	originator,
					byte[]				xfer_key )
				{
					try{
						DDBaseValueImpl	res = (DDBaseValueImpl)
							handler.read(
									new DDBaseContactImpl( DDBaseImpl.this, originator ),
									type,
									new DDBaseKeyImpl( xfer_key ));
						
						if ( res == null ){
							
							return( null );
						}
						
						return( res.getBytes());
						
					}catch( Throwable e ){
						
						Debug.printStackTrace(e);
						
						return( null );
					}
				}
				
				public void
				handleWrite(
					DHTPluginContact	originator,
					byte[]				xfer_key,
					byte[]				value )
				{
					try{
						DDBaseContactImpl	contact = new DDBaseContactImpl( DDBaseImpl.this, originator );
						
						handler.write( 
							contact,
							type,
							new DDBaseKeyImpl( xfer_key ),
							new DDBaseValueImpl( contact, value, SystemTime.getCurrentTime()));
						
					}catch( Throwable e ){
						
						Debug.printStackTrace(e);
					}
				}
			});
	}
	
	public DistributedDatabaseTransferType
	getStandardTransferType(
		int		standard_type )
	
		throws DistributedDatabaseException
	{
		if ( standard_type == DistributedDatabaseTransferType.ST_TORRENT ){
			
			return( torrent_transfer );
		}
		
		throw( new DistributedDatabaseException( "unknown type" ));
	}
	
	protected DistributedDatabaseValue
	read(
		DDBaseContactImpl							contact,
		final DistributedDatabaseProgressListener	listener,
		DistributedDatabaseTransferType				type,
		DistributedDatabaseKey						key,
		long										timeout )
	
		throws DistributedDatabaseException
	{
		if ( type == torrent_transfer ){
			
			return( torrent_transfer.read( contact, listener, type, key, timeout ));
			
		}else{
			
			byte[]	data = getDHT().read( 
								new DHTPluginProgressListener()
								{
									public void
									reportSize(
										long	size )
									{
										listener.reportSize( size );
									}
									
									public void
									reportActivity(
										String	str )
									{
										listener.reportActivity( str );
									}
									
									public void
									reportCompleteness(
										int		percent )
									{
										listener.reportCompleteness( percent );
									}
								},
								contact.getContact(),
								DDBaseHelpers.getKey(type.getClass()).getHash(),
								((DDBaseKeyImpl)key).getBytes(),
								timeout );
								
			if ( data == null ){
				
				return( null );
			}
			
			return( new DDBaseValueImpl( contact, data, SystemTime.getCurrentTime()));
		}
	}
	
	protected class
	listenerMapper
		implements DHTPluginOperationListener
	{
		private DistributedDatabaseListener	listener;
		private int							type;
		private DistributedDatabaseKey		key;
		private byte[]						key_bytes;
		private long						timeout;
		private boolean						complete_disabled;
		private boolean						exhaustive;
		
		protected
		listenerMapper(
			DistributedDatabaseListener	_listener,
			int							_type,
			DistributedDatabaseKey		_key,
			long						_timeout,
			boolean						_exhaustive )
		{
			listener	= _listener;
			type		= _type;
			key			= _key;
			key_bytes	= ((DDBaseKeyImpl)key).getBytes();
			timeout		= _timeout;
			exhaustive	= _exhaustive;
		}
		
		private
		listenerMapper(
			DistributedDatabaseListener	_listener,
			int							_type,
			DistributedDatabaseKey		_key,
			byte[]						_key_bytes,
			long						_timeout )
		{
			listener	= _listener;
			type		= _type;
			key			= _key;
			key_bytes	= _key_bytes;
			timeout		= _timeout;
		}
		
		public void
		valueRead(
			DHTPluginContact	originator,
			DHTPluginValue		_value )
		{
			byte[]	value = _value.getValue();
			
			if ( _value.getFlags() == DHTPlugin.FLAG_MULTI_VALUE ){

				int	pos = 1;
				
				while( pos < value.length ){
					
					int	len = (	( value[pos++]<<8 ) & 0x0000ff00 )+
							 	( value[pos++] & 0x000000ff );
					
					if ( len > value.length - pos ){
						
						Debug.out( "Invalid length: len = " + len + ", remaining = " + (value.length - pos ));
						
						break;
					}
					
					byte[]	d = new byte[len];
					
					System.arraycopy( value, pos, d, 0, len );
					
					listener.event( new dbEvent( type, key, originator, d, _value.getCreationTime()));
					
					pos += len;
				}				

				if ( value[0] == 1 ){
					
						// continuation exists
					
					final	byte[]	next_key_bytes = new SHA1Simple().calculateHash( key_bytes );
					
					complete_disabled	= true;
	
					dht.get(	
						next_key_bytes, 
						key.getDescription(),
						(byte)0, 
						256, 
						timeout, 
						exhaustive,
						new listenerMapper( listener, DistributedDatabaseEvent.ET_VALUE_READ, key, next_key_bytes, timeout ));
				}
			}else{
				
				listener.event( new dbEvent( type, key, originator, _value ));
			}
		}
		
		public void
		valueWritten(
			DHTPluginContact	target,
			DHTPluginValue		value )
		{
			listener.event( new dbEvent( type, key, target, value ));
		}

		public void
		complete(
			boolean	timeout_occurred )
		{
			if ( !complete_disabled ){
				
				listener.event( 
					new dbEvent( 
						timeout_occurred?DistributedDatabaseEvent.ET_OPERATION_TIMEOUT:DistributedDatabaseEvent.ET_OPERATION_COMPLETE,
						key ));
			}
		}
	}
	
	protected class
	dbEvent
		implements DistributedDatabaseEvent
	{
		private int							type;
		private DistributedDatabaseKey		key;
		private DistributedDatabaseValue	value;
		private DDBaseContactImpl			contact;
		
		protected 
		dbEvent(
			int						_type,
			DistributedDatabaseKey	_key )
		{
			type	= _type;
			key		= _key;
		}
		
		protected
		dbEvent(
			int						_type,
			DistributedDatabaseKey	_key,
			DHTPluginContact		_contact,
			DHTPluginValue			_value )
		{
			type		= _type;
			key			= _key;
			
			contact	= new DDBaseContactImpl( DDBaseImpl.this, _contact );
			
			value	= new DDBaseValueImpl( contact, _value.getValue(), _value.getCreationTime()); 
		}
		
		protected
		dbEvent(
			int						_type,
			DistributedDatabaseKey	_key,
			DHTPluginContact		_contact,
			byte[]					_value,
			long					_ct )
		{
			type		= _type;
			key			= _key;
			
			contact	= new DDBaseContactImpl( DDBaseImpl.this, _contact );
			
			value	= new DDBaseValueImpl( contact, _value, _ct ); 
		}
		
		public int
		getType()
		{
			return( type );
		}
		
		public DistributedDatabaseKey
		getKey()
		{
			return( key );
		}
		
		public DistributedDatabaseValue
		getValue()
		{
			return( value );
		}
		
		public DistributedDatabaseContact
		getContact()
		{
			return( contact );
		}
	}
}