dhtplugin.java

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

			// changes that occurred during dht init
		
		try{
			port_change_mon.enter();
		
			port_change_outstanding	= _new_port;
			
			if ( port_changing ){
								
				return;
			}
			
			port_changing			= true;
			
		}finally{
			
			port_change_mon.exit();
		}
		
		new AEThread("DHTPlugin:portChanger", true )
		{
			public void
			runSupport()
			{
				while( true ){
					
					int	new_port;
				
					try{
						port_change_mon.enter();

						new_port	= port_change_outstanding;
						
					}finally{
						
						port_change_mon.exit();
					}	
						
					try{
						dht_data_port	= new_port;
						
						if ( upnp_mapping != null ){
							
							if ( upnp_mapping.getPort() != new_port ){
								
								upnp_mapping.setPort( new_port );
							}
						}
						
						if ( status == STATUS_RUNNING ){
							
							if ( dhts != null ){
								
								for (int i=0;i<dhts.length;i++){
									
									if ( dhts[i].getPort() != new_port ){
										
										dhts[i].setPort( new_port );
									}
								}
							}
						}
					}finally{
						
						try{
							port_change_mon.enter();

							if ( new_port == port_change_outstanding ){
								
								port_changing	= false;
								
								break;
							}
							
						}finally{
							
							port_change_mon.exit();
						}						
					}
				}
			}
		}.start();
	}
	
	protected void
	initComplete(
		final UITextField		status_area,
		final boolean			logging,
		final String			override_ip )
	{
		Thread t = 
			new AEThread( "DHTPlugin.init" )
			{
				public void
				runSupport()
				{
					try{							
						
						enabled = VersionCheckClient.getSingleton().DHTEnableAllowed();
						
						if ( enabled ){
							
							status_area.setText( "Initialising" );
							
							List	plugins = new ArrayList();
							
							if ( MAIN_DHT_ENABLE ){
								
								DHTPluginImpl plug = new DHTPluginImpl(
												plugin_interface,
												AzureusCoreFactory.getSingleton().getNATTraverser(),
												DHTTransportUDP.PROTOCOL_VERSION_MAIN,
												DHT.NW_MAIN,
												override_ip,
												dht_data_port,
												reseed,
												logging,
												log, dht_log );
								
								plugins.add( plug );
							}
							
							if ( Constants.isCVSVersion() && CVS_DHT_ENABLE ){
								
								plugins.add( new DHTPluginImpl(
										plugin_interface,
										AzureusCoreFactory.getSingleton().getNATTraverser(),
										DHTTransportUDP.PROTOCOL_VERSION_CVS,
										DHT.NW_CVS,
										override_ip,
										dht_data_port,
										reseed,
										logging,
										log, dht_log ));
							}
							
							DHTPluginImpl[]	_dhts = new DHTPluginImpl[plugins.size()];
							
							plugins.toArray( _dhts );
												
							dhts = _dhts;
							
							status = dhts[0].getStatus();
							
							status_area.setText( dhts[0].getStatusText());
						
							dhts[0].getDHT().getTransport().addListener(
					        		new DHTTransportListener()
					        		{
					        			public void
					        			localContactChanged(
					        				DHTTransportContact	local_contact )
					        			{
					        				for (int i=0;i<listeners.size();i++){
					        					
					        					((DHTPluginListener)listeners.get(i)).localAddressChanged(dhts[0].getLocalAddress());
					        				}
					        			}
					        			
					        			public void
					        			currentAddress(
					        				String		address )
					        			{
					        			}
					        			
					        			public void
					        			reachabilityChanged(
					        				boolean	reacheable )
					        			{
					        			}
					        		});							
							
						}else{
							
							status	= STATUS_DISABLED;

							status_area.setText( "Disabled administratively due to network problems" );
						}
					}catch( Throwable e ){
						
						enabled	= false;
						
						status	= STATUS_DISABLED;

						status_area.setText( "Disabled due to error during initialisation" );

						log.log( e );
						
						Debug.printStackTrace(e);
						
					}finally{
						
						init_sem.releaseForever();
					}
					
						// pick up any port changes that occurred during init
					
					if ( status == STATUS_RUNNING ){
					
						changePort( dht_data_port );
					}
				}
			};
			
		t.setDaemon(true);
		
		t.start();
	}
	
	protected void
	setPluginInfo()
	{
		boolean	reachable	= plugin_interface.getPluginconfig().getPluginBooleanParameter( "dht.reachable." + DHT.NW_MAIN, true );

		plugin_interface.getPluginconfig().setPluginParameter( 
				"plugin.info", 
				reachable?"1":"0" );
	}

	public boolean
	isEnabled()
	{
		init_sem.reserve();
		
		return( enabled );
	}
	
	public boolean
	peekEnabled()
	{
		if ( init_sem.isReleasedForever()){
			
			return( enabled );
		}
		
		return( true );	// don't know yet
	}
	
	public boolean
	isExtendedUseAllowed()
	{
		if ( !isEnabled()){
			
			return( false );
		}
		
		if ( !got_extended_use){
		
			got_extended_use	= true;
			
			extended_use = VersionCheckClient.getSingleton().DHTExtendedUseAllowed();
		}
		
		return( extended_use );
	}
	
	public boolean
	isReachable()
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
				
		return( dhts[0].isReachable());
	}
	
	public boolean
	isDiversified(
		byte[]		key )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
				
		return( dhts[0].isDiversified( key ));
	}
	
	public void
	put(
		final byte[]						key,
		final String						description,
		final byte[]						value,
		final byte							flags,
		final DHTPluginOperationListener	listener)
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
				
		dhts[0].put( key, description, value, flags, listener );
		
		for (int i=1;i<dhts.length;i++){

			final int f_i	= i;
			
			new AEThread( "mutli-dht: put", true )
			{
				public void
				runSupport()
				{
					dhts[f_i].put( key, description, value, flags, 
							new DHTPluginOperationListener()
							{
								public void
								diversified()
								{
								}
								
								public void
								valueRead(
									DHTPluginContact	originator,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":put valueRead" );
									}
								}
								
								public void
								valueWritten(
									DHTPluginContact	target,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":put valueWritten" );
									}
								}
								
								public void
								complete(
									boolean	timeout_occurred )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":put complete, timeout=" + timeout_occurred );
									}
								}
							});
				}
			}.start();
		}
	}
	
	public void
	get(
		final byte[]								key,
		final String								description,
		final byte									flags,
		final int									max_values,
		final long									timeout,
		final boolean								exhaustive,
		final boolean								high_priority,
		final DHTPluginOperationListener			listener )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
				
		dhts[0].get( key, description, flags, max_values, timeout, exhaustive, high_priority, listener );
		
		for (int i=1;i<dhts.length;i++){

			final int f_i	= i;
			
			new AEThread( "mutli-dht: get", true )
			{
				public void
				runSupport()
				{
					dhts[f_i].get( 
							key, description, flags, max_values, timeout, exhaustive, high_priority,
							new DHTPluginOperationListener()
							{
								public void
								diversified()
								{
								}
								
								public void
								valueRead(
									DHTPluginContact	originator,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":get valueRead" );
									}
								}
								
								public void
								valueWritten(
									DHTPluginContact	target,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":get valueWritten" );
									}
								}
								
								public void
								complete(
									boolean	timeout_occurred )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":get complete, timeout=" + timeout_occurred );
									}
								}
							});
				}
			}.start();
		}
	}
	
	public void
	remove(
		final byte[]						key,
		final String						description,
		final DHTPluginOperationListener	listener )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
				
		dhts[0].remove( key, description, listener );
		
		for (int i=1;i<dhts.length;i++){

			final int f_i	= i;
			
			new AEThread( "mutli-dht: remove", true )
			{
				public void
				runSupport()
				{
					dhts[f_i].remove( 
							key, description, 
							new DHTPluginOperationListener()
							{
								public void
								diversified()
								{
								}
								
								public void
								valueRead(
									DHTPluginContact	originator,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":remove valueRead" );
									}
								}
								
								public void
								valueWritten(
									DHTPluginContact	target,
									DHTPluginValue		value )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":remove valueWritten" );
									}
								}
								
								public void
								complete(
									boolean	timeout_occurred )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":remove complete, timeout=" + timeout_occurred );
									}
								}
							});
				}
			}.start();
		}
	}
	
	public DHTPluginContact
	importContact(
		InetSocketAddress				address )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}

			// first DHT will do here
		
		return( dhts[0].importContact( address ));
	}
	
	public DHTPluginContact
	getLocalAddress()
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}

			// first DHT will do here
		
		return( dhts[0].getLocalAddress());
	}
	
		// direct read/write support
	
	public void
	registerHandler(
		byte[]							handler_key,
		final DHTPluginTransferHandler	handler )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
		
		for (int i=0;i<dhts.length;i++){
			
			dhts[i].registerHandler( handler_key, handler );
		}
	}
	
	public byte[]
	read(
		final DHTPluginProgressListener	listener,
		final DHTPluginContact			target,
		final byte[]					handler_key,
		final byte[]					key,
		final long						timeout )
	{
		if ( !isEnabled()){
			
			throw( new RuntimeException( "DHT isn't enabled" ));
		}
		
		for (int i=1;i<dhts.length;i++){

			final int f_i	= i;
			
			new AEThread( "mutli-dht: readXfer", true )
			{
				public void
				runSupport()
				{
					dhts[f_i].read( 
							new DHTPluginProgressListener()
							{
								public void
								reportSize(
									long	size )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":readXfer: size = " + size );
									}
								}
								
								public void
								reportActivity(
									String	str )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":readXfer: act = " + str );
									}
								}
								
								public void
								reportCompleteness(
									int		percent )
								{
									if ( TRACE_NON_MAIN ){
										System.out.println( "DHT_" + f_i + ":readXfer: % = " + percent );
									}
								}
							},
							target, handler_key, key, timeout );
				}
			}.start();
		}
		
		return( dhts[0].read( listener, target, handler_key, key, timeout ));
	}

	public int
	getStatus()
	{
		return( status );
	}
	
	public DHT[]
	getDHTs()
	{
		if ( dhts == null ){
			
			return( new DHT[0] );
		}
		
		DHT[]	res = new DHT[ dhts.length ];
		
		for (int i=0;i<res.length;i++){
			
			res[i] = dhts[i].getDHT();
		}
		
		return( res );
	}
	
	public DHTPluginKeyStats
	decodeStats(
		DHTPluginValue		value )
	{
		return( dhts[0].decodeStats( value ));
	}
	
	public void
	addListener(
		DHTPluginListener	l )
	{
		listeners.add(l);
	}
	
	public void
	removeListener(
		DHTPluginListener	l )
	{
		listeners.remove(l);
	}
	
	public void
	log(
		String	str )
	{
		log.log( str );
	}
}