pepeertransportprotocol.java

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

	 * Checks if a particular piece makes us interested in the peer
	 * @param pieceNumber the piece number that has been received
	 */
	/*
	private void checkInterested(int pieceNumber)
	{
		if (closing)
			return;

		// Do we need this piece and it's not Done?
		if (!interested_in_other_peer &&diskManager.isInteresting(pieceNumber))
        {
			connection.getOutgoingMessageQueue().addMessage(new BTInterested(), false);
			interested_in_other_peer =true;
        }
	}
	*/
	
  /**
	 * Private method to send the bitfield.
	 */
  private void sendBitField()
	{
		if (closing)
			return;

		// In case we're in super seed mode, we don't send our bitfield
		if (manager.isSuperSeedMode())
			return;

		ArrayList lazies =null;

		// create bitfield
		final DirectByteBuffer buffer =DirectByteBufferPool.getBuffer(DirectByteBuffer.AL_MSG, (nbPieces +7) /8);
		final DiskManagerPiece[] pieces =diskManager.getPieces();

		int bToSend =0;
		int i =0;
		for (; i <pieces.length; i++ )
		{
			if ((i %8) ==0)
				bToSend =0;
			bToSend =bToSend <<1;
			if (pieces[i].isDone())
			{
				if (ENABLE_LAZY_BITFIELD)
				{
					if (i <8 ||i >=(pieces.length -(pieces.length %8)))
					{ // first and last bytes
						if (lazies ==null)
							lazies =new ArrayList();
						lazies.add(new Integer(i)); // send as a Have message instead
					} else
						bToSend +=1;
				} else
					bToSend +=1;
			}
			if ((i %8) ==7)
				buffer.put(DirectByteBuffer.SS_BT, (byte) bToSend);
		}
		if ((i %8) !=0)
		{
			bToSend =bToSend <<(8 -(i %8));
			buffer.put(DirectByteBuffer.SS_BT, (byte) bToSend);
		}

		buffer.flip(DirectByteBuffer.SS_BT);

		connection.getOutgoingMessageQueue().addMessage(new BTBitfield(buffer), false);

		if (lazies !=null)
		{
			for (int x =0; x <lazies.size(); x++ )
			{
				final Integer num =(Integer) lazies.get(x);
				connection.getOutgoingMessageQueue().addMessage(new BTHave(num.intValue()), false);
			}
		}

	}

  
  public byte[] getId() {  return peer_id;  }
  public String getIp() {  return ip;  }
  public int getPort() {  return port;  }
  
  public int getTCPListenPort() {  return tcp_listen_port;  }
  public int getUDPListenPort() {  return udp_listen_port;  }
  public int getUDPNonDataListenPort() { return( udp_non_data_port ); }
  
  public String getClient() {  return client;  }
  
  public boolean isIncoming() {  return incoming;  }  
  
  
  public boolean isOptimisticUnchoke() {  return is_optimistic_unchoke && !isChokedByMe();  }
  public void setOptimisticUnchoke( boolean is_optimistic ) {  is_optimistic_unchoke = is_optimistic;  }
  
  
  public PEPeerControl getControl() {  return manager;  }
  public PEPeerManager getManager() {  return manager;  }
  public PEPeerStats getStats() {  return peer_stats;  }
  
	public int[]
	getPriorityOffsets()
	{
			// normal peer has no special priority requirements
		
		return( null );
	}
	
	public boolean
	requestAllocationStarts(
		int[]	base_priorities )
	{
		return( false );
	}
	
	public void
	requestAllocationComplete()
	{	
	}
	
  	/**
  	 * @return null if no bitfield has been recieved yet
  	 * else returns BitFlags indicating what pieces the peer has
  	 */
  	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_set_by_accessor;  }
  private void
  setSeed(
	boolean	s )
  {
	  if ( seed_set_by_accessor != s ){
		  
		  seed_set_by_accessor	= s;
		  
		  if ( peer_exchange_item != null && s){
			  
			  peer_exchange_item.seedStatusChanged();
		  }
	  }
  }
  
  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 (!closing)
		{
			final long now =SystemTime.getCurrentTime();
			if (!b)
			{
				if (snubbed !=0)
				{
					snubbed =0;
					manager.decNbPeersSnubbed();
					if (!choked_by_other_peer)
						unchokedTime =now;
				}
			} else if (snubbed ==0)
			{
				snubbed =now;
				manager.incNbPeersSnubbed();
				if (!choked_by_other_peer)
				{
			    	final long unchoked =now -unchokedTime;
		    		if (unchoked >0)
		    			unchokedTimeTotal +=unchoked;
				}
			}
		}
	}
  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 final void
					IPResolutionComplete(
						String		res,
						boolean		ok )
					{
						ip_resolved	= res;
					}
				});
		}
		
		return( ip_resolved );
	}
	


	private void cancelRequests()
    {
        if (!closing) {   //cancel any unsent requests in the queue
            final Message[] type ={new BTRequest(-1, -1, -1)};
            connection.getOutgoingMessageQueue().removeMessagesOfType(type, false);
        }
        if (requested !=null &&requested.size() >0) {
			try{
				requested_mon.enter();
		        
                if (!closing)
                {   // may have unchoked us, gotten a request, then choked without filling it - snub them
                    // if they actually have data coming in, they'll be unsnubbed as soon as it writes
                    final long timeSinceGoodData =getTimeSinceGoodDataReceived();
                    if (timeSinceGoodData ==-1 ||timeSinceGoodData >60 *1000)
                        setSnubbed(true);
                }
                for (int i =requested.size() -1; i >=0; i--) {
                    final DiskManagerReadRequest request =(DiskManagerReadRequest) requested.remove(i);
                    manager.requestCanceled(request);
                }
			}finally{
			
				requested_mon.exit();
			}
        }
    }
  
		public int 
		getMaxNbRequests() 
		{
			return( -1 );
		}

		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 =requested.size() -1; i >=0; i--)
                {
                    final DiskManagerReadRequest request = (DiskManagerReadRequest) requested.get(i);
		    		
	    			if (request.isExpired()){
	    				
	    				if ( result == null ){
	    				
	    					result = new ArrayList();
	    				}
	    				
	    				result.add(request);
	    			}
		    	}
		    	
		    	return( result );
		    	
		    }catch(Throwable e ){
		    	
		    	return result;
		    }
		}
		
    
		private boolean	hasBeenRequested( DiskManagerReadRequest request ) {
			try{  requested_mon.enter();
	    
				return requested.contains( request );
      }
      finally{  requested_mon.exit();  }
		}
    
		/** @deprecated no longer used by CVS code
		*/		
		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();
	    	}
	    	final BTRequest msg = new BTRequest( request.getPieceNumber(), request.getOffset(), request.getLength() );
	    	connection.getOutgoingMessageQueue().removeMessage( msg, false );
        msg.destroy();
		}
		
		protected void 
		reSetRequestsTime(final long now)
		{
			try{
			  requested_mon.enter();

			  final int requestedSize =requested.size();
			  for (int i =0; i <requestedSize; i++)
			  {
				  final DiskManagerReadRequest request =(DiskManagerReadRequest) requested.get(i);
				  if (request !=null)
					  request.reSetTime(now);
			  }
			}finally{
				
				requested_mon.exit();
			}
		}
    

	public String toString() {
    if( connection != null && connection.isConnected() ) {
      return connection + " [" + client+ "]";
    }
    return ip + ":" + port + " [" + client+ "]";
	}