ddbaseimpl.java

这是一个基于java编写的torrent的P2P源码

	
	public void
	delete(
		final DistributedDatabaseListener		listener,
		final DistributedDatabaseKey			key )
	
		throws DistributedDatabaseException
	{
		throwIfNotAvailable();
		
		getDHT().remove( ((DDBaseKeyImpl)key).getBytes(),
					key.getDescription(),
					new listenerMapper( listener, DistributedDatabaseEvent.ET_VALUE_DELETED, key, 0, false, 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 );
		
		final String	handler_name = type==torrent_transfer?"Torrent Transfer":"Plugin Defined";
		
		getDHT().registerHandler(
			type_key.getHash(),
			new DHTPluginTransferHandler()
			{
				public String
				getName()
				{
					return( handler_name );
				}
				
				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(), -1));
						
					}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(), -1));
		}
	}
	
	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;
		private boolean						high_priority;
		
		private int							continuation_num;
		
		protected
		listenerMapper(
			DistributedDatabaseListener	_listener,
			int							_type,
			DistributedDatabaseKey		_key,
			long						_timeout,
			boolean						_exhaustive,
			boolean						_high_priority )
		{
			listener	= _listener;
			type		= _type;
			key			= _key;
			key_bytes	= ((DDBaseKeyImpl)key).getBytes();
			timeout		= _timeout;
			exhaustive	= _exhaustive;
			high_priority	= _high_priority;
			
			continuation_num	= 1;
		}
		
		private
		listenerMapper(
			DistributedDatabaseListener	_listener,
			int							_type,
			DistributedDatabaseKey		_key,
			byte[]						_key_bytes,
			long						_timeout,
			int							_continuation_num )
		{
			listener	= _listener;
			type		= _type;
			key			= _key;
			key_bytes	= _key_bytes;
			timeout		= _timeout;
			
			continuation_num	= _continuation_num;
		}
		
		public void
		diversified()
		{
		}
		
		public void
		valueRead(
			DHTPluginContact	originator,
			DHTPluginValue		_value )
		{
			if ( type == DistributedDatabaseEvent.ET_KEY_STATS_READ ){
				
				if (( _value.getFlags() & DHTPlugin.FLAG_STATS ) == 0 ){
					
						// skip, old impl
					
					return;
				}
				
				try{
					final DHTPluginKeyStats	stats = getDHT().decodeStats( _value );
					
					DistributedDatabaseKeyStats ddb_stats = new
						DistributedDatabaseKeyStats()
						{
							public int
							getEntryCount()
							{
								return( stats.getEntryCount());
							}
							
							public int
							getSize()
							{
								return( stats.getSize());
							}
							
							public int
							getReadsPerMinute()
							{
								return( stats.getReadsPerMinute());
							}
							
							public byte
							getDiversification()
							{
								return( stats.getDiversification());
							}
						};
						
					listener.event( new dbEvent( type, key, originator, ddb_stats ));
					
				}catch( Throwable e ){
					
					Debug.printStackTrace(e);
				}
			}else{
				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(), _value.getVersion()));
						
						pos += len;
					}				
	
					if ( value[0] == 1 ){
						
							// continuation exists
						
						final	byte[]	next_key_bytes = new SHA1Simple().calculateHash( key_bytes );
						
						complete_disabled	= true;
		
						grabDHT().get(	
							next_key_bytes, 
							key.getDescription() + " [continuation " + continuation_num + "]",
							(byte)0, 
							256, 
							timeout, 
							exhaustive,
							high_priority,
							new listenerMapper( listener, DistributedDatabaseEvent.ET_VALUE_READ, key, next_key_bytes, timeout, continuation_num+1 ));
					}
				}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 DistributedDatabaseKeyStats	key_stats;
		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(), _value.getVersion()); 
		}
		
		protected
		dbEvent(
			int								_type,
			DistributedDatabaseKey			_key,
			DHTPluginContact				_contact,
			DistributedDatabaseKeyStats		_key_stats )
		{
			type		= _type;
			key			= _key;
			
			contact	= new DDBaseContactImpl( DDBaseImpl.this, _contact );

			key_stats	= _key_stats;
		}
		
		protected
		dbEvent(
			int						_type,
			DistributedDatabaseKey	_key,
			DHTPluginContact		_contact,
			byte[]					_value,
			long					_ct,
			long					_v )
		{
			type		= _type;
			key			= _key;
			
			contact	= new DDBaseContactImpl( DDBaseImpl.this, _contact );
			
			value	= new DDBaseValueImpl( contact, _value, _ct, _v ); 
		}
		
		public int
		getType()
		{
			return( type );
		}
		
		public DistributedDatabaseKey
		getKey()
		{
			return( key );
		}
		
		public DistributedDatabaseKeyStats
		getKeyStats()
		{
			return( key_stats );
		}

		public DistributedDatabaseValue
		getValue()
		{
			return( value );
		}
		
		public DistributedDatabaseContact
		getContact()
		{
			return( contact );
		}
	}
}