pepeertransportprotocol.java

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

  	public BitFlags getAvailable()
	{
		return peerHavePieces;
	}
  public boolean isPieceAvailable(int pieceNumber)
  {
	  if (peerHavePieces !=null)
		  return peerHavePieces.flags[pieceNumber];
	  return false;
  };
  public boolean isChokingMe() {  return choked_by_other_peer;  }
  public boolean isChokedByMe() {  return choking_other_peer;  }
  /**
   * @return true if the peer is interesting to us
   */
  public boolean isInteresting() {  return interested_in_other_peer;  }
  /**
   * @return true if the peer is interested in what we're offering
   */
  public boolean isInterested() {  return other_peer_interested_in_me;  }
  public boolean isSeed() {  return seed;  }
  public boolean isSnubbed() {  return snubbed !=0;  }

	public long getSnubbedTime()
	{
		if (snubbed ==0)
			return 0;
		final long now =SystemTime.getCurrentTime();
		if (now <snubbed)
			snubbed =now -26; // odds are ...
		return now -snubbed;
	}

	public void setSnubbed(boolean b)
	{
		if (!b)
			snubbed =0;
		else if (snubbed ==0)
			snubbed =SystemTime.getCurrentTime();
	}
  public void setUploadHint(int spreadTime) {  spreadTimeHint = spreadTime;  }
  public int getUploadHint() {  return spreadTimeHint;  }
  public void setUniqueAnnounce(int _uniquePiece) {  uniquePiece = _uniquePiece;  }
  public int getUniqueAnnounce() {  return uniquePiece;  }



  /** To retreive arbitrary objects against a peer. */
  public Object getData (String key) {
  	if (data == null) return null;
    return data.get(key);
  }

  /** To store arbitrary objects against a peer. */
  public void setData (String key, Object value) {
  	try{
      data_mon.enter();
  	
	  	if (data == null) {
	  	  data = new HashMap();
	  	}
	    if (value == null) {
	      if (data.containsKey(key))
	        data.remove(key);
	    } else {
	      data.put(key, value);
	    }
  	}finally{
      data_mon.exit();
  	}
  }




	public String
	getIPHostName()
	{
		if ( ip_resolved == null ){
			
			ip_resolved = ip;
		
			ip_resolver_request = IPToHostNameResolver.addResolverRequest( 
				ip_resolved,
				new IPToHostNameResolverListener()
				{
					public void
					IPResolutionComplete(
						String		res,
						boolean		ok )
					{
						ip_resolved	= res;
					}
				});
		}
		
		return( ip_resolved );
	}
	


	private void cancelRequests() {
	  if ( requested != null ) {
	    try{
	      requested_mon.enter();
		        
	      for (int i = requested.size() - 1; i >= 0; i--) {
	        DiskManagerReadRequest request = (DiskManagerReadRequest) requested.remove(i);
	        manager.requestCanceled(request);
	      }
	    }finally{
		        	
	      requested_mon.exit();
	    }
	  }
	  if( !closing ) {
	    //cancel any unsent requests in the queue
	    Message[] type = { new BTRequest( -1, -1, -1 ) };
	    connection.getOutgoingMessageQueue().removeMessagesOfType( type, false );
	  }
	}
  

		public int 
		getNbRequests() {
			return requested.size();
		}

		/**
		 * 
		 * @return	may be null for performance purposes
		 */
		
		public List 
		getExpiredRequests() {
			List result = null;
			
				// this is frequently called, hence we operate without a monitor and
				// take the hit of possible exceptions due to concurrent list
				// modification (only out-of-bounds can occur)
				
			try{
		    	for (int i = 0; i < requested.size(); i++){
		    		
	    			DiskManagerReadRequest request = (DiskManagerReadRequest) requested.get(i);
		    			
	    			if (request.isExpired()){
	    				
	    				if ( result == null ){
	    					
	    					result = new ArrayList();
	    				}
	    				
	    				result.add(request);
	    			}
		    	}
		    	
		    	return( result );
		    	
		    }catch(Throwable e ){
		    	
		    	return( null );
		    }
		}
		
    
		private boolean	hasBeenRequested( DiskManagerReadRequest request ) {
			try{  requested_mon.enter();
	    
				return requested.contains( request );
      }
      finally{  requested_mon.exit();  }
		}
    
		
		protected void
		addRequest(
			DiskManagerReadRequest	request )
		{
			try{
				requested_mon.enter();
			
				requested.add(request);
			}finally{
				
				requested_mon.exit();
			}
	        _lastPiece =request.getPieceNumber();
		}
		
		protected void
		removeRequest(
			DiskManagerReadRequest	request )
		{
	    	try{
	    		requested_mon.enter();
	    	
	    		requested.remove(request);
	    	}finally{
	    		
	    		requested_mon.exit();
	    	}
	    	BTRequest msg = new BTRequest( request.getPieceNumber(), request.getOffset(), request.getLength() );
	    	connection.getOutgoingMessageQueue().removeMessage( msg, false );
        msg.destroy();
		}
		
		protected void 
		reSetRequestsTime() 
		{
			try{
			  requested_mon.enter();
	    
			  for (int i = 0; i < requested.size(); i++) {
			  	DiskManagerReadRequest request = null;
			  	try {
			  		request = (DiskManagerReadRequest) requested.get(i);
			  	}
			  	catch (Exception e) { Debug.printStackTrace( e );}
	        
			  	if (request != null)
			  		request.reSetTime();
			  }
			}finally{
				
				requested_mon.exit();
			}
		}
    

	public String toString() {
    if( connection != null && connection.isConnected() ) {
      return connection + " [" + client+ "]";
    }
    return ip + ":" + port + " [" + client+ "]";
	}
	
  
  
  
  public void doKeepAliveCheck() {
    long wait_time = SystemTime.getCurrentTime() - last_message_sent_time;
    
    if( last_message_sent_time == 0 || wait_time < 0 ) {
      last_message_sent_time = SystemTime.getCurrentTime(); //don't send if brand new connection
      return;
    }
    
    if( wait_time > 2*60*1000 ) {  //2min keep-alive timer
      sendKeepAlive();
      last_message_sent_time = SystemTime.getCurrentTime();  //not quite true, but we don't want to queue multiple keep-alives before the first is actually sent
    }
  }

  
  public boolean doTimeoutChecks() {
    //Timeouts when in states PEPeerTransport.CONNECTION_PENDING and
    //PEPeerTransport.CONNECTION_CONNECTING are handled by the ConnectDisconnectManager
    //so we don't need to deal with them here.
    
    //make sure we time out stalled connections
    if( connection_state == PEPeerTransport.CONNECTION_FULLY_ESTABLISHED ) {
      long dead_time = SystemTime.getCurrentTime() - last_message_received_time;
      
      if( dead_time < 0 ) {  //oops, system clock went backwards
        last_message_received_time = SystemTime.getCurrentTime();
        return false;
      }
      
      if( dead_time > 5*60*1000 ) { //5min timeout
        closeConnectionInternally( "timed out while waiting for messages" );
        return true;
      }
    }
    //ensure we dont get stuck in the handshaking phases
    else if( connection_state == PEPeerTransport.CONNECTION_WAITING_FOR_HANDSHAKE ) {
      long wait_time = SystemTime.getCurrentTime() - connection_established_time;
      
      if( wait_time < 0 ) {  //oops, system clock went backwards
        connection_established_time = SystemTime.getCurrentTime();
        return false;
      }
      
      if( wait_time > 3*60*1000 ) { //3min timeout
        closeConnectionInternally( "timed out while waiting for handshake" );
        return true;
      }
    }
    
    return false;
  }
  
  
  
  public void doPerformanceTuningCheck() {
    if( peer_stats != null && outgoing_piece_message_handler != null ) {

      //send speed -based tuning
      long send_rate = peer_stats.getDataSendRate() + peer_stats.getProtocolSendRate();
      
      if( send_rate >= 3125000 ) {  // 25 Mbit/s
        connection.getTCPTransport().setTransportMode( TCPTransport.TRANSPORT_MODE_TURBO );
        outgoing_piece_message_handler.setRequestReadAhead( 256 );
      }
      else if( send_rate >= 1250000 ) {  // 10 Mbit/s
        connection.getTCPTransport().setTransportMode( TCPTransport.TRANSPORT_MODE_TURBO );
        outgoing_piece_message_handler.setRequestReadAhead( 128 );
      }
      else if( send_rate >= 125000 ) {  // 1 Mbit/s
        if( connection.getTCPTransport().getTransportMode() < TCPTransport.TRANSPORT_MODE_FAST ) {
          connection.getTCPTransport().setTransportMode( TCPTransport.TRANSPORT_MODE_FAST );
        }
        outgoing_piece_message_handler.setRequestReadAhead( 32 );
      }
      else if( send_rate >= 62500 ) {  // 500 Kbit/s
        outgoing_piece_message_handler.setRequestReadAhead( 16 );
      }
      else if( send_rate >= 31250 ) {  // 250 Kbit/s
        outgoing_piece_message_handler.setRequestReadAhead( 8 );
      }
      else if( send_rate >= 12500 ) {  // 100 Kbit/s
        outgoing_piece_message_handler.setRequestReadAhead( 4 );
      }
      else {
        outgoing_piece_message_handler.setRequestReadAhead( 2 );
      }
      
      
      //receive speed -based tuning
      long receive_rate = peer_stats.getDataReceiveRate() + peer_stats.getProtocolReceiveRate();
      
      if( receive_rate >= 1250000 ) {  // 10 Mbit/s
        connection.getTCPTransport().setTransportMode( TCPTransport.TRANSPORT_MODE_TURBO );
      }
      else if( receive_rate >= 125000 ) {  // 1 Mbit/s
        if( connection.getTCPTransport().getTransportMode() < TCPTransport.TRANSPORT_MODE_FAST ) {
          connection.getTCPTransport().setTransportMode( TCPTransport.TRANSPORT_MODE_FAST );
        }
      }
      
    }
  }
  
  
  
  
  public int getConnectionState() {  return connection_state;  }
  
  
  
  
  public long getTimeSinceLastDataMessageReceived() {
    if( last_data_message_received_time == -1 ) {  //never received
      return -1;
    }
    
    long time_since = SystemTime.getCurrentTime() - last_data_message_received_time;
    
    if( time_since < 0 ) {  //time went backwards
      last_data_message_received_time = SystemTime.getCurrentTime();
      time_since = 0;
    }
    
    return time_since;    
  }
  
	public long getTimeSinceGoodDataReceived()
	{
		if (last_good_data_time ==-1)
			return -1;	// never received
		long now =SystemTime.getCurrentTime();
		long time_since =now -last_good_data_time;
		if (time_since <0)
		{	// time went backwards
			last_good_data_time =now;
			time_since =0;
		}
		return time_since;
	}
  
  
  public long getTimeSinceLastDataMessageSent() {
    if( last_data_message_sent_time == -1 ) {  //never sent
      return -1;
    }
    
    long time_since = SystemTime.getCurrentTime() - last_data_message_sent_time;
    
    if( time_since < 0 ) {  //time went backwards
      last_data_message_sent_time = SystemTime.getCurrentTime();
      time_since = 0;
    }
    
    return time_since;    
  }
  
  
  
  
  public long getTimeSinceConnectionEstablished() {
    if( connection_established_time == 0 ) {  //fudge it while the transport is being connected
      return 0;
    }
    
    long time = SystemTime.getCurrentTime() - connection_established_time;
    
    if( time < 0 ) {  //time went backwards
      connection_established_time = SystemTime.getCurrentTime();
      time = 0;
    }
    return time;
  }