udpconnectionset.java

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

		}
	}
	
	
	protected int[]
	writeHeaderStart(
		ByteBuffer	buffer,
		byte		command,
		byte		flags )
	
		throws IOException
	{		
		// packet format
	
		// int[3] sequence numbers
		// short header length starting from version
		// byte version
		// byte flags
		// short timer base in 100ths second
		// byte command: data, close, re-request
		// 
		// command specific data
		//     	data: int connection_id
		//		close: int connection_id....
		// 		...
		// for non-data packets: random 0->7 bytes
		// int checksum
		// for data packets: payload
	
		// ***** Change this for DATA and you must change PROTOCOL_DATA_HEADER_SIZE ****
		
		sendTimerBase();
		
		stats_packets_unique_sent++;
		total_packets_unique_sent++;
		
		int[]	sequence_numbers = out_seq_generator.getNextSequenceNumber();
		
		int	seq = sequence_numbers[1];
		
		buffer.putInt( sequence_numbers[0] );
		buffer.putInt( seq );
		buffer.putInt( sequence_numbers[2] );
		
			// insert space for length added later
				
		buffer.putShort((short)0 );
				
		buffer.put((byte)UDPPacket.PROTOCOL_VERSION);
		buffer.put( flags );
		buffer.putShort((short)(current_timer_base/10));
		buffer.put((byte)command);

		return( sequence_numbers );
	}

	protected int
	writeHeaderEnd(
		ByteBuffer	buffer,
		boolean		randomise_size )
	
		throws IOException
	{
		if ( randomise_size ){
			
			int	pad = random.nextInt( 8 );
			
			for (int i=0;i<pad;i++){
				
				buffer.put((byte)0);
			}
		}
		
		short	total_length = (short)buffer.position();
		
		buffer.position( 12 );
		
		buffer.putShort((short)( total_length + 4 ));
		
			// hash includes real sequence + header content but obviously not including the hash
		
		byte[] buffer_bytes = buffer.array();
		
		SHA1Hasher	hasher = new SHA1Hasher();
		
		hasher.update( buffer_bytes, 4, 4 );
		hasher.update( buffer_bytes, 12, total_length - 12 );
			
		byte[] hash = hasher.getDigest();

		buffer.position( total_length );
		
		buffer.put( hash, 0, 4 );
		
		total_length += 4;
		
			// don't encrypt the sequence numbers
		
		header_cipher_out.processBytes( buffer_bytes, 12, total_length-12, buffer_bytes, 12 );
		
		if ( total_length > MAX_HEADER ){
			
			Debug.out( "MAX_HEADER exceeded!!!!" );
			
			throw( new IOException( "MAX_HEADER exceeded" ));
		}
		
		return( total_length );
	}
		
	protected int 
	write( 
		UDPConnection	connection,
		ByteBuffer[]	buffers,
		int				offset,
		int				length )
	
		throws IOException
	{
		if ( !canWrite( connection )){
			
			return( 0 );
		}
		
		synchronized( connection_writers ){
			
			int	size = connection_writers.size();
			
			if ( size == 0 ){
				
				connection_writers.add( connection );
				
			}else if ( connection_writers.size() == 1 && connection_writers.get(0) == connection ){
				
			}else{
				
				connection_writers.remove( connection );
			
				connection_writers.addLast( connection );
			}
		}
		
		if ( total_packets_sent == 0 ){
			
			return( sendCrypto( buffers, offset, length ));
			
		}else{
		
			return( sendDataCommand( connection, buffers, offset, length ));
		}
	}
	
	protected boolean
	canWrite(
		UDPConnection	connection )
	{
		if ( !crypto_done ){
			
			if ( connection != lead_connection ){
				
				return( false );
			}
			
			if ( total_packets_sent > 0 ){
				
				return( false );
			}
		}
		
		boolean	space =  transmit_unack_packets.size() < MAX_TRANSMIT_UNACK_DATA_PACKETS;
		
		/*
		boolean	old_log = LOG;
		
		LOG = !space;
		
		if ( manager.trace() != old_log ){
			
			System.out.println( "Log: " + (LOG?"On":"Off"));
		}
		*/
		
		return( space );
	}
	
	public void
	close(
		UDPConnection	connection,
		String			reason )
	{
		if ( manager.trace() ){
			trace( connection, "close: " + reason );
		}
		
		boolean	found;
		
		synchronized( connections ){
			
			found = connections.containsValue( connection );
		}
		
		if ( found ){
			
			try{
				sendCloseCommand( connection );
				
			}catch( Throwable e ){
				
				failed( e );
			}
		}

			// final poll incase there are ignorant listeners
		
		connection.poll();

		manager.remove( this, connection );
	}
	
	public void
	failed(
		UDPConnection	connection,
		Throwable		reason )
	{
		if ( manager.trace() ){
			trace( connection, "Failed: " + Debug.getNestedExceptionMessage(reason));
		}
		
			// run a final poll operation to inform any selector listeners of the failure
			
		connection.poll();
		
		manager.remove( this, connection );
	}
	
	protected void
	failed(
		Throwable e )
	{
		List	conns = null;
		
		synchronized( connections ){
			
			if ( !failed ){
				
				if ( manager.trace() ){
					trace( "Connection set failed: " + Debug.getNestedExceptionMessage( e ));
				}
			
				failed	= true;

				conns = new ArrayList( connections.values());
			}
		}
		
		if ( conns != null ){
				
			for (int i=0;i<conns.size();i++){
				
				try{
					((UDPConnection)conns.get(i)).failed( e );
					
				}catch( Throwable f ){
					
					Debug.printStackTrace(f);
				}
			}
			
			manager.failed( this );
		}
	}
	
	protected boolean
	hasFailed()
	{			
		return( failed );
	}
	
	protected void
	removed()
	{
		logStats();
	}
	
	static void
	forDocumentation()
	{
			// this is here to draw attention to the fact that there's a dependency between packet formats...
		
		PRUDPPacketReply.registerDecoders( new HashMap());
	}
	

	
	protected int
	cipherInt(
		RC4Engine	cipher,
		int			i )
	{
		byte[]	bytes = intToBytes( i );
		
		cipher.processBytes( bytes, 0, bytes.length, bytes, 0 );
		
		return( bytesToInt( bytes, 0 ));
	}
	
	protected int
	bytesToInt(
		byte[]	bytes,
		int		offset )
	{
		int 	res = 	(bytes[offset++]<<24)&0xff000000 | 
						(bytes[offset++]<<16)&0x00ff0000 | 
						(bytes[offset++]<<8)&0x0000ff00 | 
						bytes[offset++]&0x000000ff;
				
		return( res );
	}
	
	protected byte[]
	intToBytes(
		int		i )
	{
		byte[] res = new byte[]{ (byte)(i>>24), (byte)(i>>16), (byte)(i>>8), (byte)i };
		
		return( res );
	}
	
	protected long
	bytesToLong(
		byte[]	bytes )
	{
		return( bytesToLong( bytes, 0 ));
	}
	
	protected long
	bytesToLong(
		byte[]	bytes,
		int		offset )
	{
		long 	i1 = 	(bytes[offset++]<<24)&0xff000000L | 
						(bytes[offset++]<<16)&0x00ff0000L | 
						(bytes[offset++]<<8)&0x0000ff00L | 
						bytes[offset++]&0x000000ffL;
						
		long 	i2 = 	(bytes[offset++]<<24)&0xff000000L | 
						(bytes[offset++]<<16)&0x00ff0000L | 
						(bytes[offset++]<<8)&0x0000ff00L | 
						bytes[offset++]&0x000000ffL;				

		long	res = ( i1 << 32 ) | i2;
				
		return( res );
	}
	
	protected String
	getName()
	{
		return( "loc="+local_port + " - " + remote_address );
	}
	
	protected void
	logStats()
	{
		if ( Logger.isEnabled()){
				
			synchronized( this ){
				String	str = "sent: tot=" + total_packets_sent + ",uni=" + total_packets_unique_sent +
								",ds=" + total_data_sent + ",dr=" + total_data_resent +
								",ps=" + total_protocol_sent + ",pr=" + total_protocol_resent + 
								",rt=" + total_packets_resent_via_timer + ",ra=" + total_packets_resent_via_ack;
				
				str += " recv: tot=" + total_packets_received + ",uni=" + total_packets_unique_received +
								",du=" + total_packets_duplicates + ",oo=" + total_packets_out_of_order;
				
				str += " timer=" + current_timer_base + ",adj=" + timer_is_adjusting;
				
				Logger.log(new LogEvent(LOGID, "UDP " + getName() + " - " + str ));
			}
		}
	}
	
	protected void
	trace(
		String		str )
	{
		if ( manager.trace()){
			
			manager.trace( "UDP " + getName() + ": " + str );
		}
	}
	
	protected void
	trace(
		UDPConnection	connection,
		String			str )
	{
		if ( manager.trace()){
			
			manager.trace(  "UDP " + getName() + " (" + connection.getID() + "): " + str );
		}
	}
	
	protected class
	SequenceGenerator
	{
		private Random		generator;
		private RC4Engine	cipher;
		private boolean		in;
		
		private final int[]	seq_memory;
		private final int[]	alt_seq_memory;
		private int seq_memory_pos;
		
		private int debug_seq_in_next	= outgoing?0:1000000;
		private int debug_seq_out_next	= outgoing?1000000:0;

		protected
		SequenceGenerator(
			Random		_generator,
			RC4Engine	_cipher,
			boolean		_in )
		{
			generator	= _generator;
			cipher		= _cipher;
			in			= _in;
			
			seq_memory		= new int[MAX_SEQ_MEMORY];
			alt_seq_memory	= new int[MAX_SEQ_MEMORY];
			
			Arrays.fill( seq_memory, -1 );
			Arrays.fill( alt_seq_memory, -1 );
		}
		
		protected synchronized int[]
      	getNextSequenceNumber()
      	{
      			// damn tracker udp protocol has:
      			// request: long (random connection id) int (action)
      			// reply: int (action) int (random txn id)
      		
      			// now action is always < 2048 so all other uses of udp packets will have either
      			// 0x000007ff in either bytes 9 or 0 onwards. So we're forced to use 12 byte sequence numbers
      		
      			// internally we use the middle integer as the packet sequence
      		
      			// a secondary identifier for the sequence is also generated to be used in header position
      			// 0-2 and 8-10 when reporting last sequences in the clear
      				
      		final int	mask = 0xfffff800;
      		
      		while( true ){
      			
      			int	seq1;
      			int	seq2;
      			int	seq3;
      			int	seq4;
      			
      			if ( DEBUG_SEQUENCES ){
      				
      				if ( in ){
      					seq1 = 0xffffffff;
      					seq2 = debug_seq_in_next;
      					seq3 = 0xffffffff;
      					seq4 = debug_seq_in_next;
      					
      					debug_seq_in_next++;
      				}else{
      					seq1 = 0xffffffff;
      					seq2 = debug_seq_out_next;
      					seq3 = 0xffffffff;
      					seq4 = debug_seq_out_next;
      					
      					debug_seq_out_next++;
      				}
      				
      				
      			}else{
      				seq1 = generator.nextInt();
      				seq2 = generator.nextInt();
      				seq3 = generator.nextInt();
      				seq4 = generator.nextInt();
      			
      				seq1 = cipherInt( cipher, seq1 );
      				seq2 = cipherInt( cipher, seq2 );
      				seq3 = cipherInt( cipher, seq3 );
      				seq4 = cipherInt( cipher, seq4 );
      			}
      			
  				if (( seq1 & mask ) != 0 && seq2 != -1 && ( seq3 & mask ) != 0){
  					
  					if ( (seq4 & 0xffff0000) != 0 && (seq4 & 0x0000ffff) != 0 ){
  						
  						boolean	bad	= false;
  						
  						for (int i=0;i<MAX_SEQ_MEMORY;i++){
  							
  							if ( seq_memory[i] == seq2 || alt_seq_memory[i] == seq4 ){
  								
  								bad	= true;
  								
  								break;
  							}
  						}
  						
  						if ( !bad ){
  							
  							seq_memory[seq_memory_pos]			= seq2;
  							
  							alt_seq_memory[seq_memory_pos++]	= seq4;
  							
  							if ( seq_memory_pos == MAX_SEQ_MEMORY ){
  								
  								seq_memory_pos = 0;
  							}
  							
  							return( new int[]{ seq1, seq2, seq3, seq4 });
  						}
  					}
  				}
  			}
      	}
      	
		protected boolean
      	isValidAlterativeSequence(
      		int		seq )
		{
			for (int i=0;i<MAX_SEQ_MEMORY;i++){
				
				if ( alt_seq_memory[i] == seq ){
					
					return( true );
				}
			}
			
			return( false );
      	}
	}
}