dhtnatpuncherimpl.java

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

				trace( "received connect reply: " + (result==0?"failed":"ok" ));

				if ( result == 1 ){
					
					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 );
		}
	}
	
	protected void
	receiveConnect(
		DHTTransportContact		rendezvous,
		Map						request,
		Map						response )
	{
		trace( "received connect request" );

		boolean	ok = false;
			
			// ensure that we've received this from our current rendezvous node
		
		DHTTransportContact	rt = rendezvous_target;
		
		if ( rt != null && rt.getAddress().equals( rendezvous.getAddress())){
					
			final DHTTransportUDPContact	target = decodeContact( (byte[])request.get( "origin" ));
			
			if ( target != null ){
													
				int	transport_port = 0;
				
				Long	indirect_port = (Long)request.get( "port" );
			
				if ( indirect_port != null ){
				
					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 ));
					}
				}
				
				Map	originator_client_data  = (Map)request.get( "client_data" );
				
				boolean	no_tunnel = false;
				
				if ( originator_client_data == null ){
					
					originator_client_data = new HashMap();
					
				}else{
					
					no_tunnel = originator_client_data.get( "_notunnel" ) != null;
				}
								
				if ( no_tunnel ){
					
					log( "Received message from " + target.getString());

				}else{
					
					log( "Received connect request from " + target.getString());

						// ping the origin a few times to try and establish a tunnel
									
					UTTimerEvent event = 
						timer.addPeriodicEvent(
								3000,
								new UTTimerEventPerformer()
								{
									private int pings = 1;
									
									public void
									perform(
										UTTimerEvent		ev )
									{
										if ( pings > 3 ){
											
											ev.cancel();
											
											return;
										}
											
										pings++;
										
										if ( sendTunnelInbound( target )){
											
											ev.cancel();
										}
									}
								});
						
					if ( sendTunnelInbound( target )){
						
						event.cancel();
					}
				}
				
				Map client_data = adapter.getClientData( target.getTransportAddress(), originator_client_data );
				
				if ( client_data == null ){
					
					client_data = new HashMap();
				}
				
				response.put( "client_data", client_data );
				
				ok	= true;
				
			}else{
				
				log( "Connect request: failed to decode target" );
			}
		}else{
			
			log( "Connect request from invalid rendezvous: " + rendezvous.getString());
		}
		
		response.put( "ok", new Long(ok?1:0));
	}
	
	protected boolean
	sendTunnelInbound(
		DHTTransportContact target )
	{
		log( "Sending tunnel inbound message to " + target.getString());
		
		try{
			Map	message = new HashMap();
			
			message.put( "type", new Long( RT_TUNNEL_INBOUND ));
			
			return( sendTunnelMessage( target, message ));
			
		}catch( Throwable e ){
			
			log(e);
			
			return( false );
		}
	}
	
	protected void
	receiveTunnelInbound(
		DHTTransportUDPContact		originator,
		Map							data )
	{
		log( "Received tunnel inbound message from " + originator.getString());
							
		try{
			punch_mon.enter();
		
			for (int i=0;i<oustanding_punches.size();i++){
				
				Object[]	wait_data = (Object[])oustanding_punches.get(i);
				
				DHTTransportContact	wait_contact = (DHTTransportContact)wait_data[0];
				
				if( originator.getAddress().getAddress().equals( wait_contact.getAddress().getAddress())){
					
					wait_data[2] = new Integer( originator.getTransportAddress().getPort());
											
					((AESemaphore)wait_data[1]).release();
				}
			}
			
		}finally{
			
			punch_mon.exit();
		}
	}
	
	protected boolean
	sendTunnelOutbound(
		DHTTransportContact target )
	{
		log( "Sending tunnel outbound message to " + target.getString());
		
		try{
			Map	message = new HashMap();
			
			message.put( "type", new Long( RT_TUNNEL_OUTBOUND ));
			
			return( sendTunnelMessage( target, message ));
			
		}catch( Throwable e ){
			
			log(e);
			
			return( false );
		}
	}
	
	protected void
	receiveTunnelOutbound(
		DHTTransportContact		originator,
		Map						data )
	{
		log( "Received tunnel outbound message from " + originator.getString());
	}
	
	public Map
	punch(
		String					reason,
		InetSocketAddress[]		target,
		DHTTransportContact[]	rendezvous_used,
		Map						originator_client_data )
	{
		try{
			DHTTransportUDP	transport = (DHTTransportUDP)dht.getTransport();

			DHTTransportUDPContact contact = transport.importContact( target[0], transport.getProtocolVersion());
			
			Map	result = punch( reason, contact, rendezvous_used, originator_client_data );
			
			target[0] = contact.getTransportAddress();
			
			return( result );
			
		}catch( Throwable e ){
			
			Debug.printStackTrace(e);
			
			return( null );
		}
	}
	
	public Map
	punch(
		String					reason,
		DHTTransportContact		_target,
		DHTTransportContact[]	rendezvous_used,
		Map						originator_client_data )
	{
		DHTTransportUDPContact	target = (DHTTransportUDPContact)_target;
		
		try{
			DHTTransportContact rendezvous = getRendezvous( reason, target );
			
			if ( rendezvous_used != null && rendezvous_used.length > 0 ){
				
				rendezvous_used[0] = rendezvous;
			}
			
			if ( rendezvous == null ){
				
				return( null );
			}
			
			Map	target_client_data = sendPunch( rendezvous, target, originator_client_data, false );
			
			if ( target_client_data != null ){
				
				log( "    punch to " + target.getString() + " succeeded" );
				
				return( target_client_data );
			}
			
		}catch( Throwable e ){
			
			log( e );
		}
		
		log( "    punch to " + target.getString() + " failed" );

		return( null );
	}
	
	public Map
	sendMessage(
		InetSocketAddress		rendezvous,
		InetSocketAddress		target,
		Map						message )
	{
		try{
			DHTTransportUDP	transport = (DHTTransportUDP)dht.getTransport();

			DHTTransportUDPContact rend_contact 	= transport.importContact( rendezvous, transport.getProtocolVersion());
			DHTTransportUDPContact target_contact 	= transport.importContact( target, transport.getProtocolVersion());
						
			Map	result = sendPunch( rend_contact, target_contact, message, true );

			return( result );
			
		}catch( Throwable e ){
			
			Debug.printStackTrace(e);
			
			return( null );
		}
	}
	
	
	public void
	setRendezvous(
		DHTTransportContact		target,
		DHTTransportContact		rendezvous )
	{
		explicit_rendezvous_map.put( target.getAddress(), rendezvous );
		
		if ( target.getAddress().equals( dht.getTransport().getLocalContact().getAddress())){
			
			publish( true );
		}
	}
	
	protected DHTTransportContact
	getRendezvous(
		String				reason,
		DHTTransportContact	target )
	{
		DHTTransportContact	explicit = (DHTTransportContact)explicit_rendezvous_map.get( target.getAddress());
		
		if ( explicit != null ){
			
			return( explicit );
		}
		
		byte[]	key = getPublishKey( target );
		
		final DHTTransportValue[]	result_value = {null};
		
		final Semaphore sem = plugin_interface.getUtilities().getSemaphore();
		
		dht.get( 	key, 
					reason + ": lookup for '" + target.getString() + "'",
					(byte)0,
					1,
					RENDEZVOUS_LOOKUP_TIMEOUT,
					false, true,
					new DHTOperationAdapter()
					{
						public void
						read(
							DHTTransportContact	contact,
							DHTTransportValue	value )
						{
							result_value[0] = value;
							
							sem.release();
						}
						
						public void
						complete(
							boolean				timeout )
						{
							sem.release();
						}
					});
		
		sem.reserve();
		
		DHTTransportContact result = null;
		
		if ( result_value[0] != null ){
			
			byte[]	bytes = result_value[0].getValue();
			
			try{
				ByteArrayInputStream	bais = new ByteArrayInputStream( bytes );
				
				DataInputStream	dis = new DataInputStream( bais );
				
				byte	version = dis.readByte();
				
				if ( version != 0 ){
					
					throw( new Exception( "Unsupported rendezvous version '" + version + "'" ));
				}
				
				result = dht.getTransport().importContact( dis );
				
			}catch( Throwable e ){
				
				log(e);
			}
		}
		
		log( "Lookup of rendezvous for " + target.getString() + " -> " + ( result==null?"None":result.getString()));

		return( result );
	}
	
	protected byte[]
	getPublishKey(
		DHTTransportContact	contact )
	{
		byte[]	id = contact.getID();
		byte[]	suffix = ":DHTNATPuncher".getBytes();
		
		byte[]	res = new byte[id.length + suffix.length];
		
		System.arraycopy( id, 0, res, 0, id.length );
		System.arraycopy( suffix, 0, res, id.length, suffix.length );
		
		return( res );
	}
	
	protected byte[]
   	encodePublishValue(
   		DHTTransportContact	contact )
   	{ 		
		try{
	   		ByteArrayOutputStream	baos = new ByteArrayOutputStream();
	   		
	   		DataOutputStream	dos = new DataOutputStream(baos);
	   		
	   		dos.writeByte( 0 );	// version 
	   		
	   		contact.exportContact( dos );
	   		
	   		dos.close();
	   		
	  		return( baos.toByteArray());

		}catch( Throwable e ){
			
			log( e );
		
			return( new byte[0]);
    	}
   	}
	
	protected byte[]
  	encodeContact(
  		DHTTransportContact	contact )
  	{ 		
		try{
	   		ByteArrayOutputStream	baos = new ByteArrayOutputStream();
	   		
	   		DataOutputStream	dos = new DataOutputStream(baos);
	   		   		
	   		contact.exportContact( dos );
	   		
	   		dos.close();
	   		
	  		return( baos.toByteArray());
	
		}catch( Throwable e ){
			
			log( e );
		
			return( null );
	   	}
  	}
	
	protected DHTTransportUDPContact
	decodeContact(
		byte[]		bytes )
	{
		try{
			ByteArrayInputStream	bais = new ByteArrayInputStream( bytes );
			
			DataInputStream	dis = new DataInputStream( bais );
						
			return((DHTTransportUDPContact)dht.getTransport().importContact( dis ));
			
		}catch( Throwable e ){
			
			log(e);
			
			return( null );
		}
	}
	
	protected void
	log(
		String	str )
	{
		if ( TRACE ){
			System.out.println( str );
		}
		
		logger.log( "NATPuncher: " + str );
	}
	
	protected void
	log(
		Throwable 	e )
	{
		if ( TRACE ){
			e.printStackTrace();
		}
		
		logger.log( "NATPuncher: error occurred" );
		
		logger.log(e);
	}
	
	protected void
	trace(
		String	str )
	{
		if ( TRACE ){
			System.out.println( str );
		}
	}
}