httpnetworkconnection.java

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

				}
			}else{
				
				decoder.addMessage( request );
			}
		}
	}
	
	protected RawMessage
	encodePiece(
		Message		message )
	{
		last_http_activity_time	= SystemTime.getCurrentTime();
		
		BTPiece	piece = (BTPiece)message;
		
		List	ready_requests = new ArrayList();
		
		boolean	found = false;
		
		synchronized( outstanding_requests ){

			if ( destroyed ){
				
				return( getEmptyRawMessage( message ));
			}
		
			for (int i=0;i<outstanding_requests.size();i++){
				
				pendingRequest	req = (pendingRequest)outstanding_requests.get(i);
				
				if ( 	req.getPieceNumber() == piece.getPieceNumber() &&
						req.getStart() 	== piece.getPieceOffset() &&
						req.getLength() == piece.getPieceData().remaining( DirectByteBuffer.SS_NET )){
		
					if ( req.getBTPiece() == null ){
					
						req.setBTPiece( piece );
						
						found	= true;
						
						if ( i == 0 ){
							
							Iterator	it = outstanding_requests.iterator();
							
							while( it.hasNext()){
								
								pendingRequest r = (pendingRequest)it.next();
								
								BTPiece	btp = r.getBTPiece();
								
								if ( btp == null ){
									
									break;
								}
								
								it.remove();
								
								ready_requests.add( r );
							}
						}
					
						break;
					}
				}
			}
		}
		
		if ( !found ){
			
			Debug.out( "request not matched" );
			
			return( getEmptyRawMessage( message ));
		}
		
		if ( ready_requests.size() == 0 ){
			
			return( getEmptyRawMessage( message ));
		}
		
		try{
			submitBTRequests();
			
		}catch( IOException e ){
			
		}
		
		pendingRequest req	= (pendingRequest)ready_requests.get(0);
		
		DirectByteBuffer[]	buffers;
		int					buffer_index	= 0;
		
		httpRequest	http_request = req.getHTTPRequest();
		
		buffers = new DirectByteBuffer[ ready_requests.size() + 1 ];

		if ( !http_request.hasSentFirstReply()){
		
			http_request.setSentFirstReply();
						
			String	header = encodeHeader( http_request );
			
			buffers[buffer_index++] = new DirectByteBuffer( ByteBuffer.wrap( header.getBytes()));
			
		}else{
			
				// we have to do this as core code assumes buffer entry 0 is protocol
			
			buffers[buffer_index++] = new DirectByteBuffer( ByteBuffer.allocate(0));
		}
		
		for (int i=0;i<ready_requests.size();i++){
			
			req	= (pendingRequest)ready_requests.get(i);

			BTPiece	this_piece = req.getBTPiece();
			
			int	piece_number = this_piece.getPieceNumber();
			
			if ( !piece_map.get( piece_number )){
				
					// kinda crappy as it triggers on first block of piece, however better
					// than nothing
				
				piece_map.set( piece_number );
				
				decoder.addMessage( new BTHave( piece_number ));
			}
			
			buffers[buffer_index++] = this_piece.getPieceData();
		}
		
		return(	new RawMessageImpl( 
						message, 
						buffers,
						RawMessage.PRIORITY_HIGH, 
						true, 
						new Message[0] ));
	}
	
	protected int
	getRequestCount()
	{
		synchronized( outstanding_requests ){

			return( http_requests.size());
		}
	}
	
	protected boolean
	isClosing()
	{
		return( closing );
	}
	
	protected void
	close(
		String	reason )
	{
		closing	= true;
		
		peer.getControl().removePeer( peer );
	}
	
	protected void
	destroy()
	{
		synchronized( http_connection_map ){

			List	connections = (List)http_connection_map.get( network_connection_key );
			
			if ( connections != null ){
				
				connections.remove( this );
				
				if ( connections.size() == 0 ){
					
					http_connection_map.remove( network_connection_key );
				}
			}
		}
		
		synchronized( outstanding_requests ){

			destroyed	= true;
			
			for (int i=0;i<outstanding_requests.size();i++){
				
				pendingRequest	req = (pendingRequest)outstanding_requests.get(i);
				
				BTPiece	piece = req.getBTPiece();
				
				if ( piece != null ){
					
					piece.destroy();
				}
			}
			
			outstanding_requests.clear();
			
			for (int i=0;i<choked_requests.size();i++){
				
				BTRequest	req = (	BTRequest)choked_requests.get(i);
								
				req.destroy();
			}
			
			choked_requests.clear();
		}
	}
	
	protected long
	getTimeSinceLastActivity()
	{
		long	now = SystemTime.getCurrentTime();
		
		if ( now < last_http_activity_time ){
			
			last_http_activity_time = now;
		}
		
		return( now - last_http_activity_time );
	}
	
	protected void
	log(
		String	str )
	{
		if (Logger.isEnabled()){
			Logger.log(new LogEvent( getPeer(),LOGID, str));
		}   
	}
	
	protected RawMessage
	getEmptyRawMessage(
		Message	message )
	{		
		return( 
			new RawMessageImpl( 
					message, 
					new DirectByteBuffer[]{ new DirectByteBuffer( ByteBuffer.allocate(0))},
					RawMessage.PRIORITY_HIGH, 
					true, 
					new Message[0] ));
	}
	
	protected void
	sendAndClose(
		String		data )
	{
		final Message	http_message = new HTTPMessage( data );
		
		getConnection().getOutgoingMessageQueue().registerQueueListener(
			new OutgoingMessageQueue.MessageQueueListener()
			{
				public boolean 
				messageAdded( 
					Message message )
				{	
					return( true );
				}
				   
				public void 
				messageQueued( 
					Message message )
				{			
				}
				    
				public void 
				messageRemoved( 
					Message message )
				{
				}
				    
				public void 
				messageSent( 
					Message message )
				{
					if ( message == http_message ){
						
						close( "Close after message send complete" );
					}
				}
				    
			    public void 
			    protocolBytesSent( 
			    	int byte_count )
			    {	
			    }
				 
			    public void 
			    dataBytesSent( 
			    	int byte_count )
			    {	
			    }
			});
		
		getConnection().getOutgoingMessageQueue().addMessage( http_message, false );
	}
	
	protected class
	httpRequest
	{
		private long[]	offsets;
		private long[]	lengths;
		private boolean	partial_content;
		
		private int		index;
		private long	total_length;
		private boolean	sent_first_reply;
		
		private boolean	keep_alive;
		
		protected
		httpRequest(
			long[]		_offsets,
			long[]		_lengths,
			boolean		_partial_content,
			boolean		_keep_alive )
		{
			offsets	= _offsets;
			lengths	= _lengths;
			partial_content	= _partial_content;
			keep_alive		= _keep_alive;
			
			/*
			String	str ="";
			for (int i=0;i<lengths.length;i++){	
				str += (i==0?"":",") +"[" + offsets[i] + "/" + lengths[i] + "]";
			}
			System.out.println( network_connection_key.getName() + ": requested " + str + ",part=" + partial_content +",ka=" + keep_alive );
			*/
			
			for (int i=0;i<lengths.length;i++){
				
				total_length += lengths[i];
			}
		}
		
		protected boolean
		isPartialContent()
		{
			return( partial_content );
		}
		
		protected boolean
		hasSentFirstReply()
		{
			return( sent_first_reply );
		}
		
		protected void
		setSentFirstReply()
		{
			sent_first_reply	= true;
		}
		
		protected long[]
		getOffsets()
		{
			return( offsets );
		}
		
		protected long[]
   		getLengths()
   		{
   			return( lengths );
   		}
		
		protected int
		getIndex()
		{
			return( index );
		}
		
		protected void
		setIndex(
			int	_index )
		{
			index = _index;
		}
		
		protected long
		getTotalLength()
		{
			return( total_length );
		}
		
		protected boolean
		keepAlive()
		{
			return( keep_alive );
		}
	}
	
	protected class
	pendingRequest
	{
		private int	piece;
		private int	start;
		private int	length;
				
		private httpRequest	http_request;
		
		private BTPiece	bt_piece;
		
		protected
		pendingRequest(
			BTRequest		_request,
			httpRequest		_http_request )
		{
			piece	= _request.getPieceNumber();
			start	= _request.getPieceOffset();
			length	= _request.getLength();
			
			http_request	= _http_request;
		}
		
		protected int
		getPieceNumber()
		{
			return( piece );
		}
		
		protected int
		getStart()
		{
			return( start );
		}
		
		protected int
		getLength()
		{
			return( length );
		}
		
		protected httpRequest
		getHTTPRequest()
		{
			return( http_request );
		}
		
		protected BTPiece
		getBTPiece()
		{
			return( bt_piece );
		}
		
		protected void
		setBTPiece(
			BTPiece	_bt_piece )
		{
			bt_piece	= _bt_piece;
		}
	}
	
	protected class
	networkConnectionKey
	{
		public boolean 
		equals(Object obj) 
		{
			networkConnectionKey	other = (networkConnectionKey)obj;
			
			return( Arrays.equals( getAddress(), other.getAddress()) &&
					Arrays.equals(getHash(),other.getHash()));	
		}
			
		protected String
		getName()
		{
			return( peer.getControl().getDisplayName() + ": " + connection.getEndpoint().getNotionalAddress().getAddress().getHostAddress());
		}
		
		protected byte[]
		getAddress()
		{
			return( connection.getEndpoint().getNotionalAddress().getAddress().getAddress());
		}
		
		protected byte[]
		getHash()
		{
			return( peer.getControl().getHash());
		}
		
		public int 
		hashCode() 
		{
			return( peer.getControl().hashCode());

		}
	}
}