udpconnectionmanager.java

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

			//System.out.println( "recv:" + ByteFormatter.encodeString( data, 0, data_length>64?64:data_length ) + (data_length>64?"...":""));
			
			connection_set.receive( data, data_length );
			
		}catch( IOException e ){
			
			connection_set.failed( e );
			
		}catch( Throwable e ){
			
			Debug.printStackTrace( e );
			
			connection_set.failed( e );
		}
	}
	
	protected synchronized boolean
	rateLimitIncoming(
		InetSocketAddress	s_address )
	{
		byte[]	address = s_address.getAddress().getAddress();
		
		int	hit_count = incoming_bloom.add( address );
		
		long	now = SystemTime.getCurrentTime();

			// allow up to 10% bloom filter utilisation
		
		if ( incoming_bloom.getSize() / incoming_bloom.getEntryCount() < 10 ){
			
			incoming_bloom = BloomFilterFactory.createAddRemove4Bit(incoming_bloom.getSize() + BLOOM_INCREASE );
			
			incoming_bloom_create_time	= now;
			
     		Logger.log(	new LogEvent(LOGID, "UDP connnection bloom: size increased to " + incoming_bloom.getSize()));

		}else if ( now < incoming_bloom_create_time || now - incoming_bloom_create_time > BLOOM_RECREATE ){
			
			incoming_bloom = BloomFilterFactory.createAddRemove4Bit(incoming_bloom.getSize());
			
			incoming_bloom_create_time	= now;
		}
			
		if ( hit_count >= 15 ){
			
     		Logger.log(	new LogEvent(LOGID, "UDP incoming: too many recent connection attempts from " + s_address ));
     		
			return( false );
		}
		
		long	since_last = now - last_incoming;
		
		long	delay = 100 - since_last;
		
			// limit to 10 a second
		
		if ( delay > 0 && delay < 100 ){
			
			try{
				Thread.sleep( delay );
				
			}catch( Throwable e ){
			}
		}
		
		last_incoming = now;	
		
		return( true );
	}
	
	public int
	send(
		int					local_port,
		InetSocketAddress	remote_address,
		byte[]				data )
	
		throws IOException
	{
		return( network_glue.send( local_port, remote_address, data ));
	}
	
	protected void
	accept(
		final int				local_port,
		final InetSocketAddress	remote_address,
		final UDPConnection		connection )
	{
		final UDPTransportHelper	helper = new UDPTransportHelper( this, remote_address, connection );

		try{
			connection.setTransport( helper );
			
			TransportCryptoManager.getSingleton().manageCrypto( 
				helper, 
				null, 
				true, 
				null,
				new TransportCryptoManager.HandshakeListener() 
				{
					public void 
					handshakeSuccess( 
						ProtocolDecoder	decoder,
						ByteBuffer		remaining_initial_data ) 
					{
						TransportHelperFilter	filter = decoder.getFilter();
						
						ConnectionEndpoint	co_ep = new ConnectionEndpoint( remote_address);
	
						ProtocolEndpointUDP	pe_udp = new ProtocolEndpointUDP( co_ep, remote_address );
	
						UDPTransport transport = new UDPTransport( pe_udp, filter );
								
						helper.setTransport( transport );
						
						incoming_manager.addConnection( local_port, filter, transport );
	        		}
	
					public void 
					handshakeFailure( 
	            		Throwable failure_msg ) 
					{
						if (Logger.isEnabled()){
							Logger.log(new LogEvent(LOGID, "incoming crypto handshake failure: " + Debug.getNestedExceptionMessage( failure_msg )));
						}
	 
						connection.close( "handshake failure: " + Debug.getNestedExceptionMessage(failure_msg));
					}
	            
					public void
					gotSecret(
						byte[]				session_secret )
					{
						helper.getConnection().setSecret( session_secret );
					}
					
					public int
					getMaximumPlainHeaderLength()
					{
						return( incoming_manager.getMaxMinMatchBufferSize());
					}
	    		
					public int
					matchPlainHeader(
							ByteBuffer			buffer )
					{
						Object[]	match_data = incoming_manager.checkForMatch( helper, local_port, buffer, true );
	    			
						if ( match_data == null ){
	    				
							return( TransportCryptoManager.HandshakeListener.MATCH_NONE );
	    				
						}else{
							
								// no fallback for UDP
	    				
							return( TransportCryptoManager.HandshakeListener.MATCH_CRYPTO_NO_AUTO_FALLBACK );
	    				}
	    			}
	        	});
			
		}catch( Throwable e ){
			
			Debug.printStackTrace( e );
			
			helper.close( Debug.getNestedExceptionMessage(e));
		}
	}
	
	protected synchronized int
	allocationConnectionID()
	{
		int	id = next_connection_id++;
		
		if ( id < 0 ){
			
			id					= 0;
			next_connection_id	= 1;
		}
		
		return( id );
	}
	
	protected void
	timeoutDeadKeys()
	{
		Iterator	it = recently_dead_keys.values().iterator();
		
		long	now = SystemTime.getCurrentTime();
		
		while( it.hasNext()){
			
			long	dead_time = ((Long)it.next()).longValue();
		
			if ( dead_time > now || now - dead_time > DEAD_KEY_RETENTION_PERIOD ){
								
				it.remove();
			}
		}
	}
	
	protected class
	ProtocolTimer
	{
		private volatile boolean	destroyed;
		
		protected 
		ProtocolTimer()
		{
			new AEThread( "UDPConnectionManager:timer", true )
			{
				private int	tick_count;
				
				public void
				runSupport()
				{
					Thread.currentThread().setPriority( Thread.NORM_PRIORITY + 1 );
					
					while( !destroyed ){
						
						try{
							Thread.sleep( TIMER_TICK_MILLIS );
							
						}catch( Throwable e ){
							
						}
							
						tick_count++;
						
						if ( tick_count % STATS_TICKS == 0 ){
							
							logStats();
						}
						
						List	failed_sets = null;
						
						synchronized( connection_sets ){
								
							int	cs_size = connection_sets.size();
							
							checkThreadDeath( cs_size > 0 );
								
							if ( cs_size > 0 ){
								
								Iterator	it = connection_sets.values().iterator();
								
								while( it.hasNext()){
									
									UDPConnectionSet	set = (UDPConnectionSet)it.next();
									
									try{
										set.timerTick();
										
										if ( set.idleLimitExceeded()){
																					
											if (Logger.isEnabled()){
												
												Logger.log(new LogEvent(LOGID, "Idle limit exceeded for " + set.getName() + ", removing" ));
											}
											
											recently_dead_keys.put( set.getKey(), new Long( SystemTime.getCurrentTime()));
											
											it.remove();
											
											set.removed();
										}
									}catch( Throwable e ){
										
										if ( failed_sets == null ){
											
											failed_sets = new ArrayList();
										}
										
										failed_sets.add( new Object[]{ set, e });
									}
								}
							}
						}
						
						if ( failed_sets != null ){
							
							for (int i=0;i<failed_sets.size();i++){
								
								Object[]	entry = (Object[])failed_sets.get(i);
								
								((UDPConnectionSet)entry[0]).failed((Throwable)entry[1]);
							}
						}
					}
					
					logStats();
				}
			}.start();
		}
		
		protected void
		destroy()
		{
			destroyed	= true;
		}
	}
	
	protected void 
	logStats()
	{
		if (Logger.isEnabled()){
			
			long[]	nw_stats = network_glue.getStats();
		
			String	str = "UDPConnection stats: sent=" + nw_stats[0] + "/" + nw_stats[1] + ",received=" + nw_stats[2] + "/" + nw_stats[3];
			
			str += ", setup discards=" + setup_discard_packets + "/" + setup_discard_bytes;
			str += ", rate discards=" + rate_limit_discard_packets + "/" + rate_limit_discard_bytes;
			
			Logger.log(new LogEvent(LOGID, str ));
		}
	}
	
	protected boolean
	trace()
	{
		return( LOG );
	}
	
	protected void
	trace(
		String				str )
	{
		if ( LOG ){
			
			if ( FORCE_LOG ){
				
				System.out.println( str );
			}
			
			if (Logger.isEnabled()){
				
				Logger.log(new LogEvent(LOGID, str ));
			}
		}
	}
}