pepeertransportprotocol.java

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

/*
 * File    : PEPeerTransportProtocol.java
 * Created : 22-Oct-2003
 * By      : stuff
 * 
 * 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.peer.impl.transport;


import java.net.InetSocketAddress;
import java.util.*;

import org.gudy.azureus2.core3.config.*;
import org.gudy.azureus2.core3.disk.*;
import org.gudy.azureus2.core3.logging.*;
import org.gudy.azureus2.core3.peer.*;
import org.gudy.azureus2.core3.peer.impl.*;
import org.gudy.azureus2.core3.peer.util.*;
import org.gudy.azureus2.core3.util.*;
import org.gudy.azureus2.plugins.PluginInterface;
import org.gudy.azureus2.plugins.network.Connection;
import org.gudy.azureus2.pluginsimpl.local.network.ConnectionImpl;

import com.aelitis.azureus.core.AzureusCoreFactory;
import com.aelitis.azureus.core.networkmanager.*;
import com.aelitis.azureus.core.peermanager.messaging.*;
import com.aelitis.azureus.core.peermanager.messaging.azureus.*;
import com.aelitis.azureus.core.peermanager.messaging.bittorrent.*;
import com.aelitis.azureus.core.peermanager.peerdb.*;
import com.aelitis.azureus.core.peermanager.piecepicker.PiecePicker;
import com.aelitis.azureus.core.peermanager.piecepicker.util.BitFlags;
import com.aelitis.azureus.core.peermanager.utils.*;
import com.aelitis.azureus.plugins.dht.DHTPlugin;

/**
 * 
 * @author MjrTom
 *			2005-2006: lastPiece, bitfield, availabilityAdded
 */

public class 
PEPeerTransportProtocol
	extends LogRelation
	implements PEPeerTransport
{
	protected final static LogIDs LOGID = LogIDs.PEER;
  
	private volatile int	_lastPiece =-1;		//last piece that was requested from this peer (mostly to try to request from same one)
	
	protected final PEPeerControl 	manager;
	protected final DiskManager		diskManager;
	protected final PiecePicker		piecePicker;
	
	protected int		nbPieces =-1;
	
	private String			peer_source;
	private byte[] peer_id;
	private String ip;
	protected String ip_resolved;
	private IPToHostNameResolverRequest	ip_resolver_request;
	
	private int port;
  
  private PeerItem peer_item_identity;
  private int tcp_listen_port = 0;
  private int udp_listen_port = 0;
	
  protected final PEPeerStats peer_stats;
  
  private final ArrayList requested = new ArrayList();
  private final AEMonitor	requested_mon = new AEMonitor( "PEPeerTransportProtocol:Req" );

  private HashMap data;
  	
	private long lastNeededUndonePieceChange;
	
  protected boolean choked_by_other_peer = true;
  protected boolean choking_other_peer = true;
  private boolean interested_in_other_peer = false;
  private boolean other_peer_interested_in_me = false;
  private volatile long snubbed =0;

  private volatile BitFlags	peerHavePieces =null;		// lazy allocation; null until needed
  private volatile boolean	availabilityAdded =false;

  private boolean seed = false;
 
  private boolean incoming;
  
  protected volatile boolean closing = false;
  private volatile int current_peer_state;
  
  protected NetworkConnection connection;
  private OutgoingBTPieceMessageHandler outgoing_piece_message_handler;
  private OutgoingBTHaveMessageAggregator outgoing_have_message_aggregator;
  private Connection	plugin_connection;
  
  private boolean identityAdded = false;  //needed so we don't remove id's in closeAll() on duplicate connection attempts

  protected int connection_state = PEPeerTransport.CONNECTION_PENDING;

  //The client name identification	
	private String client = "";
    	
	//When superSeeding, number of unique piece announced
	private int uniquePiece = -1;
  
  //When downloading a piece in exclusivity mode the piece number being downloaded
  private int reservedPiece = -1;
	
	//Spread time (0 secs , fake default)
	private int spreadTimeHint = 0 * 1000;

	protected long last_message_sent_time = 0;
	protected long last_message_received_time = 0;
	protected long last_data_message_received_time = -1;
	protected long last_good_data_time =-1;			// time data written to disk was recieved
	protected long last_data_message_sent_time = -1;
  
  private long connection_established_time = 0;

  private boolean az_messaging_mode = false;
  private Message[] supported_messages = null;
  
  
  private final AEMonitor	closing_mon	= new AEMonitor( "PEPeerTransportProtocol:closing" );
  private final AEMonitor data_mon  = new AEMonitor( "PEPeerTransportProtocol:data" );
  

  private LinkedHashMap recent_outgoing_requests;
  private AEMonitor	recent_outgoing_requests_mon;
  
  private static final boolean SHOW_DISCARD_RATE_STATS;
  static {
  	String	prop = System.getProperty( "show.discard.rate.stats" );
  	SHOW_DISCARD_RATE_STATS = prop != null && prop.equals( "1" );
  }
  
  private static int requests_discarded = 0;
  private static int requests_discarded_endgame = 0;
  private static int requests_recovered = 0;
  private static int requests_completed = 0;

  private List peer_listeners_cow;
  private final AEMonitor	peer_listeners_mon = new AEMonitor( "PEPeerTransportProtocol:PL" );

  
  //certain Optimum Online networks block peer seeding via "complete" bitfield message filtering
  //lazy mode makes sure we never send a complete (seed) bitfield
  protected static boolean ENABLE_LAZY_BITFIELD;
  
  static {
    
    COConfigurationManager.addAndFireParameterListeners(
    		new String[]{ "Use Lazy Bitfield" },
    		new ParameterListener()
    		{
    			 public void 
    			 parameterChanged(
    				String ignore )
    			 {
    				 String  prop = System.getProperty( "azureus.lazy.bitfield" );
    				 
    				 ENABLE_LAZY_BITFIELD = prop != null && prop.equals( "1" );
 
    				 ENABLE_LAZY_BITFIELD |= COConfigurationManager.getBooleanParameter( "Use Lazy Bitfield" );
    			 }
    		});
  }
  
  
  private boolean is_optimistic_unchoke = false;

  private PeerExchangerItem peer_exchange_item = null;
  private boolean peer_exchange_supported = false;
    
  protected PeerMessageLimiter message_limiter;
  
  
  
  
  
  //INCOMING
  public PEPeerTransportProtocol( PEPeerControl _manager, String _peer_source, NetworkConnection _connection ) {
    manager = _manager;
    diskManager =manager.getDiskManager();
    piecePicker =manager.getPiecePicker();
    nbPieces =diskManager.getNbPieces();
    
    peer_source	= _peer_source;
    ip    = _connection.getAddress().getAddress().getHostAddress();
    port  = _connection.getAddress().getPort();
    
    peer_item_identity = PeerItemFactory.createPeerItem( ip, port, PeerItem.convertSourceID( _peer_source ), PeerItemFactory.HANDSHAKE_TYPE_PLAIN );  //this will be recreated upon az handshake decode
    
    incoming = true;
    connection = _connection;
    plugin_connection = new ConnectionImpl(connection);
    
    peer_stats = manager.createPeerStats();

    changePeerState( PEPeer.CONNECTING );
    
    //"fake" a connect request to register our listener
    connection.connect( new NetworkConnection.ConnectionListener() {
      public void connectStarted() {
        connection_state = PEPeerTransport.CONNECTION_CONNECTING;
      }
      
      public void connectSuccess() {  //will be called immediately
      	if (Logger.isEnabled())
					Logger.log(new LogEvent(PEPeerTransportProtocol.this, LOGID,
							"In: Established incoming connection"));
        initializeConnection();
        
        /*
         * Waiting until we've received the initiating-end's full handshake, before sending back our own,
         * really should be the "proper" behavior.  However, classic BT trackers running NAT checking will
         * only send the first 48 bytes (up to infohash) of the peer handshake, skipping peerid, which means
         * we'll never get their complete handshake, and thus never reply, which causes the NAT check to fail.
         * So, we need to send our handshake earlier, after we've verified the infohash.
         * NOTE:
         * This code makes the assumption that the inbound infohash has already been validated,
         * as we don't check their handshake fully before sending our own.
         */
        sendBTHandshake();
      }
      
      public void connectFailure( Throwable failure_msg ) {  //should never happen
        Debug.out( "ERROR: incoming connect failure: ", failure_msg );
        closeConnectionInternally( "ERROR: incoming connect failure [" + PEPeerTransportProtocol.this + "] : " + failure_msg.getMessage() );
      }
      
      public void exceptionThrown( Throwable error ) {
        if( error.getMessage() == null ) {
          Debug.out( error );
        }
        
        closeConnectionInternally( "connection exception: " + error.getMessage() );
      }
    });
  }
  
  
  
  //OUTGOING
  public PEPeerTransportProtocol( PEPeerControl _manager, String _peer_source, String _ip, int _port, boolean require_crypto_handshake ) {
    manager = _manager;
    diskManager =manager.getDiskManager();
    piecePicker =manager.getPiecePicker();
    nbPieces =diskManager.getNbPieces();
    lastNeededUndonePieceChange =Long.MIN_VALUE;

    peer_source	= _peer_source;
    ip    = _ip;
    port  = _port;
    tcp_listen_port = _port;
    
    peer_item_identity = PeerItemFactory.createPeerItem( ip, tcp_listen_port, PeerItem.convertSourceID( _peer_source ), PeerItemFactory.HANDSHAKE_TYPE_PLAIN );  //this will be recreated upon az handshake decode
    
    incoming = false;
    
    peer_stats = manager.createPeerStats();
    
    if( port < 0 || port > 65535 ) {
      closeConnectionInternally( "given remote port is invalid: " + port );
      return;
    }

    boolean use_crypto = require_crypto_handshake || NetworkManager.REQUIRE_CRYPTO_HANDSHAKE;  //either peer specific or global pref
    
    if( isLANLocal() )  use_crypto = false;  //dont bother with PHE for lan peers
    
    connection = NetworkManager.getSingleton().createConnection( new InetSocketAddress( ip, port ), new BTMessageEncoder(), new BTMessageDecoder(), use_crypto, !require_crypto_handshake, manager.getTorrentHash());
    
    plugin_connection = new ConnectionImpl(connection);
    
    changePeerState( PEPeer.CONNECTING );
    
    connection.connect( new NetworkConnection.ConnectionListener() {
      public void connectStarted() {
        connection_state = PEPeerTransport.CONNECTION_CONNECTING;
      }
 
      public void connectSuccess() {
        if( closing ) {
          //Debug.out( "PEPeerTransportProtocol::connectSuccess() called when closing." );
          return;
        }
        
        if (Logger.isEnabled())
					Logger.log(new LogEvent(PEPeerTransportProtocol.this, LOGID,
							"Out: Established outgoing connection"));
        initializeConnection();
        sendBTHandshake();
      }
        
      public void connectFailure( Throwable failure_msg ) {
        closeConnectionInternally( "failed to establish outgoing connection: " + failure_msg.getMessage() );
      }
        
      public void exceptionThrown( Throwable error ) {
        if( error.getMessage() == null ) {
          Debug.out( error );
        }
        
        closeConnectionInternally( "connection exception: " + error.getMessage() );
      }
    });
      
    if (Logger.isEnabled())
			Logger.log(new LogEvent(this, LOGID,
					"Out: Creating outgoing connection"));
  }
  
  

  
  protected void initializeConnection() {
    if( closing )  return;

    recent_outgoing_requests = new LinkedHashMap( 16, .75F, true ) {
      public boolean removeEldestEntry(Map.Entry eldest) {
        return size() > 16;
      }
    };
    recent_outgoing_requests_mon  = new AEMonitor( "PEPeerTransportProtocol:ROR" );
    
    message_limiter = new PeerMessageLimiter();

    //link in outgoing piece handler
    outgoing_piece_message_handler = new OutgoingBTPieceMessageHandler(diskManager, connection.getOutgoingMessageQueue() );

    //link in outgoing have message aggregator
    outgoing_have_message_aggregator = new OutgoingBTHaveMessageAggregator( connection.getOutgoingMessageQueue() );

    connection_established_time = SystemTime.getCurrentTime();
    
    connection_state = PEPeerTransport.CONNECTION_WAITING_FOR_HANDSHAKE;
    changePeerState( PEPeer.HANDSHAKING );
    
    registerForMessageHandling();
  }
  
  
  
  
  
	public String
	getPeerSource()
	{
		return( peer_source );
	}
  


  /**
   * Close the peer connection from within the PEPeerTransport object.
   * @param reason
   */
  protected void closeConnectionInternally( String reason ) {
    performClose( reason, false );
  }
  
  
  /**
   * Close the peer connection from the PEPeerControl manager side.
   * NOTE: This method assumes PEPeerControl already knows about the close.
   * This method is inteded to be only invoked by select administrative methods.
   * You probably should not invoke this directly.
   */
  public void closeConnection( String reason ) {
    performClose( reason, true );
  }

  
  private void performClose( String reason, boolean externally_closed ) {
    try{
      closing_mon.enter();
    
      if( closing ){ 
    	  
        return;
      }
      
      closing = true;
      
      if( identityAdded ) {  //remove identity
      	if( peer_id != null ) {
      		PeerIdentityManager.removeIdentity( manager.getPeerIdentityDataID(), peer_id );
      	}
      	else {
      		Debug.out( "PeerIdentity added but peer_id == null !!!" );
      	}  
      	
      	identityAdded	= false;
      }
      
      changePeerState( PEPeer.CLOSING );