trhostimpl.java

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

/*
 * File    : TRHostImpl.java
 * Created : 24-Oct-2003
 * By      : parg
 * 
 * Azureus - a Java Bittorrent client
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License.
 *
 * This program 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 General Public License for more details ( see the LICENSE file ).
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */
 
package org.gudy.azureus2.core3.tracker.host.impl;

/**
 * @author parg
 */

import java.util.*;
import java.io.*;
import java.net.*;

import org.gudy.azureus2.core3.logging.*;
import org.gudy.azureus2.core3.config.*;
import org.gudy.azureus2.core3.util.*;
import org.gudy.azureus2.core3.tracker.host.*;
import org.gudy.azureus2.core3.tracker.server.*;
import org.gudy.azureus2.core3.tracker.util.TRTrackerUtils;
import org.gudy.azureus2.core3.tracker.client.*;
import org.gudy.azureus2.core3.torrent.*;

public class 
TRHostImpl
	implements 	TRHost, TRTrackerAnnouncerFactoryListener, 
				TRTrackerServerListener, TRTrackerServerFactoryListener,
				TRTrackerServerRequestListener, TRTrackerServerAuthenticationListener
{
	private static final LogIDs LOGID = LogIDs.TRACKER;
	private static final int URL_DEFAULT_PORT		= 80;	// port to use if none in announce URL
	private static final int URL_DEFAULT_PORT_SSL	= 443;	// port to use if none in announce URL
	
	public static final int STATS_PERIOD_SECS		= 60;
	private static final int TICK_PERIOD_SECS			= 10;
	private static final int TICKS_PER_STATS_PERIOD	= STATS_PERIOD_SECS/TICK_PERIOD_SECS;
		
	private static TRHostImpl	singleton;
	private static AEMonitor 	class_mon 	= new AEMonitor( "TRHost:class" );

	private TRHostConfigImpl		config;
		
	private Hashtable				server_map 	= new Hashtable();
	
	private List	host_torrents			= new ArrayList();
	private Map	host_torrent_hash_map	= new HashMap();
	
	private Map	host_torrent_map		= new HashMap();
	private Map	tracker_client_map		= new HashMap();
	
	private static final int LDT_TORRENT_ADDED			= 1;
	private static final int LDT_TORRENT_REMOVED		= 2;
	private static final int LDT_TORRENT_CHANGED		= 3;
	
	private ListenerManager	listeners 	= ListenerManager.createAsyncManager(
		"TRHost:ListenDispatcher",
		new ListenerManagerDispatcher()
		{
			public void
			dispatch(
				Object		_listener,
				int			type,
				Object		value )
			{
				TRHostListener	target = (TRHostListener)_listener;
		
				if ( type == LDT_TORRENT_ADDED ){
					
					target.torrentAdded((TRHostTorrent)value);
					
				}else if ( type == LDT_TORRENT_REMOVED ){
						
					target.torrentRemoved((TRHostTorrent)value);
						
				}else if ( type == LDT_TORRENT_CHANGED ){
					
					target.torrentChanged((TRHostTorrent)value);
				}
			}
		});	
	
	private static boolean host_add_announce_urls;
	
	static{
		COConfigurationManager.addAndFireParameterListener(
				"Tracker Host Add Our Announce URLs",
				new ParameterListener()
				{
					public void 
					parameterChanged(
						String name )
					{
						host_add_announce_urls = COConfigurationManager.getBooleanParameter( name );
					}
				});
	}
	
	private List	auth_listeners		= new ArrayList();
	
	private boolean	server_factory_listener_added;
	
	protected AEMonitor this_mon 	= new AEMonitor( "TRHost" );

	private volatile boolean	closed;
	
	public static TRHost
	create()
	{
		try{
			class_mon.enter();
		
			if ( singleton == null ){
				
				singleton = new TRHostImpl();
			}
			
			return( singleton );
			
		}finally{
			
			class_mon.exit();
		}
	}
	
	protected
	TRHostImpl()
	{	
			// we need to synchronize this so that the async (possible) establishment of
			// a server within the stats loop (to deal with public trackers with no locally
			// hosted torrents) doesn't get ahead of the reading of persisted torrents
			// If we allow the server to start early then it can potentially receive an
			// announce/scrape and result in the creation of an "external" torrent when
			// it should really be using an existing torrent 
			 
		try{
			this_mon.enter();
					
			config = new TRHostConfigImpl(this);	
			
			TRTrackerAnnouncerFactory.addListener( this );
			
			Thread t = new AEThread("TRHost::stats.loop")
						{
							private int	tick_count = 0;
							
							private Set	failed_ports = new HashSet();
							
							public void
							runSupport()
							{
								while(true){
									
									try{	
										
										URL[][]	url_sets = TRTrackerUtils.getAnnounceURLs();
										
										for (int i=0;i<url_sets.length;i++){
											
											URL[]	urls = url_sets[i];
											
											for (int j=0;j<urls.length;j++){
												
												URL	url = urls[j];
														
												int port = url.getPort();
												
												if ( port == -1 ){
													
													port = url.getDefaultPort();
												}
												
												String	protocol = url.getProtocol().toLowerCase();
												
												try{
													if ( protocol.equals( "http" )){
												
														startServer( TRTrackerServerFactory.PR_TCP, port, false );
													
													}else if ( protocol.equals( "udp" )){
														
														startServer( TRTrackerServerFactory.PR_UDP, port, false );

													}else if ( protocol.equals( "https" )){
														
														startServer( TRTrackerServerFactory.PR_TCP, port, true );

													}else{
														
														Debug.out( "Unknown protocol '" + protocol + "'" );
													}
													
												}catch( Throwable e ){
											
													Integer port_i = new Integer(port);
													
													if ( !failed_ports.contains(port_i)){
														
														failed_ports.add( port_i );
														
														Logger.log(
																new LogEvent(LOGID,
																"Tracker Host: failed to start server", e));
													}
												}
											}
										}
										
										Thread.sleep( TICK_PERIOD_SECS*1000 );
										
										if ( closed ){
											
											break;
										}
										
										if ( tick_count % TICKS_PER_STATS_PERIOD == 0 ){
											
											try{
												this_mon.enter();
												
												for (int i=0;i<host_torrents.size();i++){
					
													TRHostTorrent	ht = (TRHostTorrent)host_torrents.get(i);
													
													if ( ht instanceof TRHostTorrentHostImpl ){
																						
														((TRHostTorrentHostImpl)ht).updateStats();
														
													}else{
														
														((TRHostTorrentPublishImpl)ht).updateStats();
														
													}
												}
											}finally{
												
												this_mon.exit();
											}
											
											config.saveConfig( true );
											
										}else{
											
											config.saveConfig( false );
										}
										
									}catch( InterruptedException e ){
										
										Debug.printStackTrace( e );
										
										break;
									}finally{
										
										tick_count++;
									}
								}
							}
						};
			
			t.setDaemon(true);
			
				// try to ensure that the tracker stats are collected reasonably
				// regularly
			
			t.setPriority( Thread.MAX_PRIORITY -1);
			
			t.start();
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	public void
	initialise(
		TRHostTorrentFinder	finder )
	{
		config.loadConfig( finder );
	}

	public String
	getName()
	{
		return( TRTrackerServer.DEFAULT_NAME );
	}

	public TRHostTorrent
	hostTorrent(
		TOTorrent		torrent,
		boolean			persistent,
		boolean			passive )
	
		throws TRHostException
	{
		return( addTorrent( torrent, TRHostTorrent.TS_STARTED, persistent, passive, SystemTime.getCurrentTime() ));
	}
	
	public TRHostTorrent
	publishTorrent(
		TOTorrent		torrent )
		
		throws TRHostException
	{
		return( addTorrent( torrent, TRHostTorrent.TS_PUBLISHED, true, false, SystemTime.getCurrentTime()));
	}
	
	protected TRHostTorrent
	addTorrent(
		TOTorrent		torrent,
		int				state,
		boolean			persistent,
		boolean			passive,
		long			date_added )
		
		throws TRHostException
	{
		try{
			this_mon.enter();
		
				// non-persistent additions should know what they're doing regarding
				// announce URL
			
			if ( persistent && state != TRHostTorrent.TS_PUBLISHED ){

				if ( host_add_announce_urls ){
					
					addTrackerAnnounce( torrent );
				}
			}
			
			TRHostTorrent	ht = lookupHostTorrent( torrent );
			
			if ( ht != null ){
						
				// check that this isn't the explicit publish/host of a torrent already there
				// as an external torrent. If so then just replace the torrent
				
				try{
				
					ht = lookupHostTorrentViaHash( torrent.getHash());
				
					if ( ht instanceof TRHostTorrentHostImpl ){
						
						TRHostTorrentHostImpl hti = (TRHostTorrentHostImpl)ht;
						
						if ( hti.getTorrent() != torrent ){
							
							hti.setTorrent( torrent );	
						
							if ( persistent && !hti.isPersistent()){
								
								hti.setPersistent( true );
							}
							
							if ( passive && !hti.isPassive()){
								
								hti.setPassive( true );
							}
							
							if ( state != TRHostTorrent.TS_PUBLISHED ){
					
								startHosting( hti );
					
								if ( state == TRHostTorrent.TS_STARTED ){
								
									hti.start();
								}
							}	
							
							listeners.dispatch( LDT_TORRENT_CHANGED, ht );
						}
					}
				}catch( TOTorrentException e ){
					
					Debug.printStackTrace( e );	
				}
				
				return( ht );
			}
			
			int		port;
			boolean	ssl;
			int		protocol	= TRTrackerServerFactory.PR_TCP;
			
			if ( state == TRHostTorrent.TS_PUBLISHED ){
			
				port = COConfigurationManager.getIntParameter("Tracker Port", TRHost.DEFAULT_PORT );
				
				ssl	= false;		
			}else{
			
				URL	announce_url = torrent.getAnnounceURL();
				
				String	protocol_str = announce_url.getProtocol();
				
				ssl = protocol_str.equalsIgnoreCase("https");
				
				if ( protocol_str.equalsIgnoreCase("udp")){
					
					protocol = TRTrackerServerFactory.PR_UDP;
					
				}else if ( TorrentUtils.isDecentralised( torrent )){
					
					protocol = TRTrackerServerFactory.PR_DHT;
				}
				
				boolean force_external = COConfigurationManager.getBooleanParameter("Tracker Port Force External", false );
				
				port = announce_url.getPort();
				
				if ( force_external ){
					
					String 	tracker_ip 		= COConfigurationManager.getStringParameter("Tracker IP", "");
		
					if ( 	tracker_ip.length() > 0 &&
							!announce_url.getHost().equalsIgnoreCase( tracker_ip )){
							
						if ( ssl ){
			
							port = COConfigurationManager.getIntParameter("Tracker Port SSL", TRHost.DEFAULT_PORT_SSL );
							
						}else{
							
							port = COConfigurationManager.getIntParameter("Tracker Port", TRHost.DEFAULT_PORT );
							
						}
					}
				}
				
				if ( port == -1 ){
					
					port = ssl?URL_DEFAULT_PORT_SSL:URL_DEFAULT_PORT;
				}
			}
			
			TRTrackerServer server = startServer( protocol, port, ssl );
			
			TRHostTorrent host_torrent;
		
			if ( state == TRHostTorrent.TS_PUBLISHED ){
	
				TRHostTorrentPublishImpl new_torrent = new TRHostTorrentPublishImpl( this, torrent, date_added );
	
				new_torrent.setPersistent( persistent );
				
				host_torrent	= new_torrent;
			}else{
			
				TRHostTorrentHostImpl	new_torrent = new TRHostTorrentHostImpl( this, server, torrent, port, date_added );
				
				new_torrent.setPersistent( persistent );
				
				new_torrent.setPassive( passive );

				host_torrent	= new_torrent;
			}
						
			host_torrents.add( host_torrent );
			
			try{
				host_torrent_hash_map.put( new HashWrapper( torrent.getHash()), host_torrent );
				
			}catch( TOTorrentException e ){
				
				Debug.printStackTrace( e );
			}
					
			host_torrent_map.put( torrent, host_torrent );
			
			if ( state != TRHostTorrent.TS_PUBLISHED ){
			
				startHosting((TRHostTorrentHostImpl)host_torrent );
		
				if ( state == TRHostTorrent.TS_STARTED ){
						
					host_torrent.start();
				}
				
					// if not persistent, see if we can recover the stats
				
				if ( !persistent ){
					
					config.recoverStats( (TRHostTorrentHostImpl)host_torrent );
				}
			}
	
			listeners.dispatch( LDT_TORRENT_ADDED, host_torrent );
			
			config.saveRequired();
			
			return( host_torrent );
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected TRTrackerServer
	startServer(
		int		protocol,
		int		port,
		boolean	ssl )
		
		throws TRHostException
	{
		try{
			this_mon.enter();
		
			String	key = ""+protocol+ ":" + port;
			
			TRTrackerServer	server = (TRTrackerServer)server_map.get( key );
				
			if ( server == null ){
					
				try{
					
					if ( ssl ){
						
						server = TRTrackerServerFactory.createSSL( "tracker", protocol, port, true, true );
					
					}else{
					
						server = TRTrackerServerFactory.create( "tracker", protocol, port, true, true );
					}
						
					server_map.put( key, server );
						
					if ( auth_listeners.size() > 0 ){
						
						server.addAuthenticationListener( this );
					}
					
					server.addListener( this );
							
				}catch( TRTrackerServerException e ){
								
					throw( new TRHostException( "startServer failed", e ));
				}
			}
			
			return( server );
			
		}finally{
			
			this_mon.exit();
		}
	}
	
	protected TRHostTorrent
	lookupHostTorrent(
		TOTorrent	torrent )
	{
	  if (torrent == null)
	    return null;

		try{
			return((TRHostTorrent)host_torrent_hash_map.get( torrent.getHashWrapper()));
			
		}catch( TOTorrentException e ){
			
			Debug.printStackTrace( e );
		}
		
		return( null );
	}
	
	protected void
	startHosting(
		TRHostTorrentHostImpl	host_torrent )
	{
		TOTorrent	torrent = host_torrent.getTorrent();
		
		TRTrackerAnnouncer tc = (TRTrackerAnnouncer)tracker_client_map.get( torrent );
		
		if ( tc != null ){
			
			startHosting( host_torrent, tc );
		}
	}
	
	protected void
	startHosting(
		TRTrackerAnnouncer	tracker_client )
	{
		TRHostTorrent	host_torrent = (TRHostTorrent)host_torrent_map.get( tracker_client.getTorrent());
			
		if ( host_torrent instanceof TRHostTorrentHostImpl ){
			
			startHosting( (TRHostTorrentHostImpl)host_torrent, tracker_client );
		}
	}
	
	protected void
	startHosting(
		TRHostTorrentHostImpl	host_torrent,
		final TRTrackerAnnouncer 	tracker_client )
	{
		final TOTorrent	torrent = host_torrent.getTorrent();	

			// set the ip override so that we announce ourselves to other peers via the 
			// real external address, not the local one used to connect to the tracker 
			
		URL	announce = torrent.getAnnounceURL();
		
		if ( host_add_announce_urls ){
		
			tracker_client.setIPOverride( announce.getHost());
			
		}else{
			
				// prolly a backup tracker, we only want to override the IP if we're hosting it
			
			if ( TRTrackerUtils.isHosting( announce )){
				
				tracker_client.setIPOverride( announce.getHost());

			}
		}
		
			// hook into the client so that when the announce succeeds after the refresh below
			// we can force a rescrape to pick up the new status 
		
		TRTrackerAnnouncerListener	listener = 
			new TRTrackerAnnouncerListener()
			{
				public void
				receivedTrackerResponse(
					TRTrackerAnnouncerResponse	response	)
				{	
					try{
						TRTrackerScraperFactory.getSingleton().scrape( torrent, true );
					
					}finally{
						
						tracker_client.removeListener( this );
					}
				}

				public void
				urlChanged(
					String		url,
					boolean		explicit )
				{	
				}
					
				public void