diskmanagerchannelimpl.java

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

		}
		
		boolean	stop_force_start = false;
		
		synchronized( download ){
			
			Map	dl_state = (Map)download.getDownload().getData( channel_key );
			
			if ( dl_state != null ){
				
				dl_state.remove( "" + channel_id );
				
				if ( dl_state.size() == 0 ){
					
					stop_force_start	= true;
				}
			}
		}
		
		if ( stop_force_start ){
			
			download.setForceStart( false );
		}
	}
	
	protected class
	request 
		implements DiskManagerRequest
	{
		private int		request_type;
		private long	request_offset;
		private long	request_length;
		private List	listeners	= new ArrayList();
		
		private volatile boolean	cancelled;
		
		AESemaphore	wait_sem = new AESemaphore( "DiskManagerChannelImpl:wait" );
		
		public void
		setType(
			int			_type )
		{
			request_type		= _type;
		}
		
		public void
		setOffset(
			long		_offset )
		{
			request_offset	= _offset;
		}
		
		public void
		setLength(
			long		_length )
		{
			request_length	= _length;
		}
		
		public long
		getRemaining()
		{
			synchronized( data_written ){

				return( request_length - (current_position - request_offset ));
			}
		}
		
		public long
		getAvailableBytes()
		{
			if ( plugin_file.getDownloaded() == plugin_file.getLength()){
				
				return( getRemaining());
			}
			
			int	download_state = download.getState();
			
				// if the file is incomplete and the download isn't running then we don't have a view
				// of what's available or not (to do this we'd need to add stuff to access resume data) 
			
			if ( 	download_state != Download.ST_DOWNLOADING &&
					download_state != Download.ST_SEEDING ){
				
				return( -1 );
			}
			
			synchronized( data_written ){

				Iterator	it = data_written.iterator();
				
					// may not have been compacted to we need to aggregate contigous entry lengths 
				
				dataEntry	last_entry 	= null;
				
				while( it.hasNext()){
					
					dataEntry	entry = (dataEntry)it.next();
					
					long	entry_offset = entry.getOffset();
					long	entry_length = entry.getLength();

					if ( last_entry == null ){
						
						if ( entry_offset > current_position ){
							
							break;
						}
						
						if ( entry_offset <= current_position && current_position < entry_offset + entry_length ){

							last_entry = entry;
						}
					}else{
	
						if ( last_entry.getOffset() + last_entry.getLength() == entry.getOffset()){
							
							last_entry = entry;
							
						}else{
							
							break;
						}
					}	
				}
					
				if ( last_entry == null ){
					
					return( 0 );
					
				}else{
					
					return( last_entry.getOffset() + last_entry.getLength() - current_position );
				}
			}
		}
		
		public void
		run()
		{			
			long	rem = request_length;
			
			long	pos = request_offset;
			
			try{

				while( rem > 0 && !cancelled ){
					
					int	len = 0;
					
					synchronized( data_written ){
						
						current_position = pos;
						
						Iterator	it = data_written.iterator();
						
						while( it.hasNext()){
							
							dataEntry	entry = (dataEntry)it.next();
							
							long	entry_offset = entry.getOffset();
							
							if ( entry_offset > pos ){
																
								break;
							}
							
							long	entry_length = entry.getLength();
							
							long	available = entry_offset + entry_length - pos;
							
							if ( available > 0 ){
								
								len = (int)( available<MAX_READ_CHUNK?available:MAX_READ_CHUNK);
								
								break;
							}
						}
					}				
					
					if ( len > 0 ){
						
						DirectByteBuffer buffer = core_file.read( pos, len );
	
						inform( new event( new PooledByteBufferImpl( buffer ), pos, len ));
						
						pos += len;
						
						rem -= len;
						
						synchronized( data_written ){
							
							byte_rate.addValue( len );
							
							current_position = pos;
						}
					}else{
	
						inform( new event( pos ));
						
						synchronized( data_written ){
							
							waiters.add( wait_sem );
						}
						
						try{

							wait_sem.reserve();
							
						}finally{
							
							synchronized( data_written ){
								
								waiters.remove( wait_sem );
							}
						}
					}
				}
			}catch( Throwable e ){
				
				inform( e );
			}
		}
		
		public void
		cancel()
		{
			cancelled	= true;
						
			inform( new Throwable( "Request cancelled" ));

			wait_sem.release();
		}
		
		protected void
		inform(
			Throwable e )
		{
			inform( new event( e ));
		}
		
		protected void
		inform(
			event		ev )
		{
			for (int i=0;i<listeners.size();i++){
				
				try{
					((DiskManagerListener)listeners.get(i)).eventOccurred( ev );
					
				}catch( Throwable e ){
					
					Debug.printStackTrace(e);
				}
			}
		}
		
		public void
		addListener(
			DiskManagerListener	listener )
		{
			listeners.add( listener );
		}
	
		public void
		removeListener(
			DiskManagerListener	listener )
		{
			listeners.remove( listener );
		}
		
		protected class
		event
			implements DiskManagerEvent
		{
			private int					event_type;
			private Throwable			error;
			private PooledByteBuffer	buffer;
			private long				event_offset;
			private int					event_length;
			
			protected
			event(
				Throwable		_error )
			{
				event_type	= DiskManagerEvent.EVENT_TYPE_FAILED;
				error		= _error;
			}
			
			protected 
			event(
				long				_offset )
			{
				event_type		= DiskManagerEvent.EVENT_TYPE_BLOCKED;

				event_offset	= _offset;	
			}
			
			protected
			event(
				PooledByteBuffer	_buffer,
				long				_offset,
				int					_length )
			{
				event_type		= DiskManagerEvent.EVENT_TYPE_SUCCESS;
				buffer			= _buffer;
				event_offset	= _offset;
				event_length	= _length;
			}
			
			public int
			getType()
			{
				return( event_type );
			}
			
			public DiskManagerRequest
			getRequest()
			{
				return( request.this );
			}
			
			public long
			getOffset()
			{
				return( event_offset );
			}
			
			public int
			getLength()
			{
				return( event_length );
			}
			
			public PooledByteBuffer
			getBuffer()
			{
				return( buffer );
			}
			
			public Throwable
			getFailure()
			{
				return( error );
			}
		}
	}
	
	protected static class
	dataEntry
	{
		private long	offset;
		private long	length;
	
		protected
		dataEntry(
			long		_offset,
			long		_length )
		{
			offset	= _offset;
			length	= _length;
		}
		
		protected long
		getOffset()
		{
			return( offset );
		}
		
		protected long
		getLength()
		{
			return( length );
		}
		
		protected void
		setLength(
			long	_length )
		{
			length	= _length;
		}
		
		protected String
		getString()
		{
			return( "offset=" + offset + ",length=" + length );
		}
	}
}