dhtnatpuncherimpl.java

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

					}
					
					public void
					reportActivity(
						String	str )
					{
					}
					
					public void
					reportCompleteness(
						int		percent )
					{						
					}
				},
				target,
				transfer_handler_key,
				new byte[0],
				data,
				TUNNEL_TIMEOUT );
   					
			return( true );
			
   		}catch( DHTTransportException e ){
   			
   			// log(e); timeout most likely
   			
   			return( false );
   		}		
   	}
	
	protected int
	sendBind(
		DHTTransportContact	target )
	{
		try{
			Map	request = new HashMap();
			
			request.put("type", new Long( RT_BIND_REQUEST ));
			
			Map response = sendRequest( target, request, TRANSFER_TIMEOUT );
			
			if ( response == null ){
				
				return( RESP_FAILED );
			}
			
			if (((Long)response.get( "type" )).intValue() == RT_BIND_REPLY ){
				
				int	result = ((Long)response.get("ok")).intValue();
					
				trace( "received bind reply: " + (result==0?"failed":"ok" ));
					
				if ( result == 1 ){
					
					return( RESP_OK );
				}
			}
			
			return( RESP_NOT_OK );
			
		}catch( Throwable e ){
			
			log(e);
			
			return( RESP_FAILED );
		}
	}
	
	protected void
	receiveBind(
		DHTTransportUDPContact	originator,
		Map						request,
		Map						response )
	{
		trace( "received bind request" );
		
		boolean	ok 	= true;
		boolean	log	= true;
		
		try{
			server_mon.enter();
		
			Object[]	entry = (Object[])rendezvous_bindings.get( originator.getAddress().toString());
			
			if ( entry == null ){
			
				if ( rendezvous_bindings.size() == RENDEZVOUS_SERVER_MAX ){
					
					ok	= false;
				}
			}else{
				
					// already present, no need to log again
				
				log	= false;
			}
			
			if ( ok ){
				
				long	now = plugin_interface.getUtilities().getCurrentSystemTime();
				
				rendezvous_bindings.put( originator.getAddress().toString(), new Object[]{ originator, new Long( now )});
				
				response.put( "port", new Long( originator.getAddress().getPort()));
			}
		}finally{
			
			server_mon.exit();
		}
		
		if ( log ){
			
			log( "Rendezvous request from " + originator.getString() + " " + (ok?"accepted":"denied" ));
		}
		
		response.put( "ok", new Long(ok?1:0));
	}
		
	public void
	destroy()
	{
		try{
			server_mon.enter();
			
			Iterator	it = rendezvous_bindings.values().iterator();
			
			while( it.hasNext()){
				
				Object[]	entry = (Object[])it.next();
				
				final DHTTransportUDPContact	contact = (DHTTransportUDPContact)entry[0];
				
				new AEThread( "DHTNATPuncher:destroy", true )
				{
					public void
					runSupport()
					{
						sendClose( contact );
					}
				}.start();
			}
		}catch( Throwable e ){
			
			log( e );
			
		}finally{
			
			server_mon.exit();
		}
	}
	
	protected int
	sendClose(
		DHTTransportContact	target )
	{
		try{
			Map	request = new HashMap();
			
			request.put("type", new Long( RT_CLOSE_REQUEST ));
			
			Map response = sendRequest( target, request, TRANSFER_TIMEOUT );
			
			if ( response == null ){
				
				return( RESP_FAILED );
			}
			
			if (((Long)response.get( "type" )).intValue() == RT_CLOSE_REPLY ){
				
				int	result = ((Long)response.get("ok")).intValue();
					
				trace( "received close reply: " + (result==0?"failed":"ok" ));
					
				if ( result == 1 ){
					
					return( RESP_OK );
				}
			}
			
			return( RESP_NOT_OK );
			
		}catch( Throwable e ){
			
			log(e);
			
			return( RESP_FAILED );
		}
	}
	
	protected void
	receiveClose(
		DHTTransportUDPContact	originator,
		Map						request,
		Map						response )
	{		
		trace( "received close request" );
					
		final DHTTransportContact	current_target = rendezvous_target;
	
		if ( current_target != null && Arrays.equals( current_target.getID(), originator.getID())){
			
			new AEThread( "DHTNATPuncher:close", true )
			{
				public void
				runSupport()
				{
					rendezvousFailed( current_target, true );
				}
			}.start();
		}
		
		response.put( "ok", new Long(1));
	}
	
	
	
	protected int
	sendQuery(
		DHTTransportContact	target )
	{
		try{
			Map	request = new HashMap();
			
			request.put("type", new Long( RT_QUERY_REQUEST ));
			
			Map response = sendRequest( target, request, TRANSFER_TIMEOUT );
			
			if ( response == null ){
				
				return( RESP_FAILED );
			}
			
			if (((Long)response.get( "type" )).intValue() == RT_QUERY_REPLY ){
				
				int	result = ((Long)response.get("ok")).intValue();
					
				trace( "received query reply: " + (result==0?"failed":"ok" ));
					
				if ( result == 1 ){
					
					return( RESP_OK );
				}
			}
			
			return( RESP_NOT_OK );
			
		}catch( Throwable e ){
			
			log(e);
			
			return( RESP_FAILED );
		}
	}
	
	protected void
	receiveQuery(
		DHTTransportUDPContact	originator,
		Map						request,
		Map						response )
	{		
		trace( "received query request" );
					
		InetSocketAddress	address = originator.getTransportAddress();
		
		response.put( "ip", address.getAddress().getHostAddress().getBytes());
		
		response.put( "port", new Long( address.getPort()));
		
		response.put( "ok", new Long(1));
	}
	
	protected Map
	sendPunch(
		DHTTransportContact 			rendezvous,
		final DHTTransportUDPContact	target,
		Map								originator_client_data,
		boolean							no_tunnel )
	{		
		AESemaphore	wait_sem 	= new AESemaphore( "DHTNatPuncher::sendPunch" );
		Object[]	wait_data 	= new Object[]{ target, wait_sem, new Integer(0)};
		
		try{

			try{
				punch_mon.enter();
			
				oustanding_punches.add( wait_data );
				
			}finally{
				
				punch_mon.exit();
			}
			
			Map	request = new HashMap();
			
			request.put("type", new Long( RT_PUNCH_REQUEST ));
			
			request.put("target", target.getAddress().toString().getBytes());
			
			if ( originator_client_data != null ){
				
				if ( no_tunnel ){
					
					originator_client_data.put( "_notunnel", new Long(1));
				}
				
				request.put( "client_data", originator_client_data );
			}
	
				// for a message payload (i.e. no_tunnel) we double the initiator timeout to give
				// more chance for reasonable size messages to get through as they have to go through
				// 2 xfer processes
			
			Map response = sendRequest( rendezvous, request, no_tunnel?TRANSFER_TIMEOUT*2:TRANSFER_TIMEOUT );
			
			if ( response == null ){
				
				return( null );
			}
			
			if (((Long)response.get( "type" )).intValue() == RT_PUNCH_REPLY ){
				
				int	result = ((Long)response.get("ok")).intValue();

				trace( "received " + ( no_tunnel?"message":"punch") + " reply: " + (result==0?"failed":"ok" ));
				
				if ( result == 1 ){
					
						// pick up port changes from the rendezvous
											
					Long	indirect_port = (Long)response.get( "port" );
					
					if ( indirect_port != null ){
						
						int transport_port	= indirect_port.intValue();
				
						if ( transport_port != 0 ){
						
							InetSocketAddress	existing_address = target.getTransportAddress();
						
							if ( transport_port != existing_address.getPort()){
							
								target.setTransportAddress(
										new InetSocketAddress(existing_address.getAddress(), transport_port ));
							}
						}
					}	
					
					if ( !no_tunnel ){
						
							// ping the target a few times to try and establish a tunnel
											
						UTTimerEvent	event = 
							timer.addPeriodicEvent(
									3000,
									new UTTimerEventPerformer()
									{
										private int	pings = 1;
										
										public void
										perform(
											UTTimerEvent		event )
										{
											if ( pings > 3 ){
												
												event.cancel();
											
												return;
											}
										
											pings++;
												
											if ( sendTunnelOutbound( target )){
												
												event.cancel();
											}
										}
									});
							
						if ( sendTunnelOutbound( target )){
							
							event.cancel();
						}
						
							// give the other end a few seconds to kick off some tunnel events to us
						
						if ( wait_sem.reserve(10000)){
							
							event.cancel();
						}
					}
											
						// routers often fiddle with the port when not mapped so we need to grab the right one to use
						// for direct communication
					
						// first priority goes to direct tunnel messages received
					
					int	transport_port = 0;
					
					try{
						punch_mon.enter();
					
						transport_port = ((Integer)wait_data[2]).intValue();
						
					}finally{
						
						punch_mon.exit();
					}
				
					if ( transport_port != 0 ){
						
						InetSocketAddress	existing_address = target.getTransportAddress();
						
						if ( transport_port != existing_address.getPort()){
							
							target.setTransportAddress(
								new InetSocketAddress(existing_address.getAddress(), transport_port ));
						}
					}
					
					Map	target_client_data = (Map)response.get( "client_data" );
					
					if ( target_client_data == null ){
						
						target_client_data = new HashMap();
					}
					
					return( target_client_data );
				}
			}
			
			return( null );
			
		}catch( Throwable e ){
			
			log(e);
			
			return( null );
			
		}finally{
			
			try{
				punch_mon.enter();
			
				oustanding_punches.remove( wait_data );
				
			}finally{
				
				punch_mon.exit();
			}
		}
	}
	
	protected void
	receivePunch(
		DHTTransportUDPContact		originator,
		Map							request,
		Map							response )
	{
		trace( "received punch request" );
		
		boolean	ok = false;
		
		try{
			server_mon.enter();
		
			String	target_str = new String((byte[])request.get( "target" ));
			
			Object[] entry = (Object[])rendezvous_bindings.get( target_str );
		
			if ( entry != null ){
				
				DHTTransportUDPContact	target = (DHTTransportUDPContact)entry[0];
				
				Map target_client_data = sendConnect( target, originator, (Map)request.get( "client_data" ));
				
				if ( target_client_data != null ){
					
					response.put( "client_data", target_client_data );
																	
					response.put( "port", new Long( target.getTransportAddress().getPort()));
					
					ok	= true;
				}
			}
			
			log( "Rendezvous punch request from " + originator.getString() + " to " + target_str + " " + (ok?"initiated":"failed"));

		}finally{
			
			server_mon.exit();
		}
		
		response.put( "ok", new Long(ok?1:0));
	}
	
	protected Map
	sendConnect(
		DHTTransportContact target,
		DHTTransportContact	originator,
		Map					originator_client_data )
	{
		try{
			Map	request = new HashMap();
			
			request.put("type", new Long( RT_CONNECT_REQUEST ));
			
			request.put("origin", encodeContact( originator ));
							
			request.put( "port", new Long( ((DHTTransportUDPContact)originator).getTransportAddress().getPort()));
			
			if ( originator_client_data != null ){
				
				request.put( "client_data", originator_client_data );
			}
			
			Map response = sendRequest( target, request, TRANSFER_TIMEOUT );
			
			if ( response == null ){
				
				return( null );
			}
			
			if (((Long)response.get( "type" )).intValue() == RT_CONNECT_REPLY ){
				
				int	result = ((Long)response.get("ok")).intValue();