trnonblockingserver.java

java 文件下载器。可自定义

// Decompiled by Jad v1.5.8e2. Copyright 2001 Pavel Kouznetsov.
// Jad home page: http://kpdus.tripod.com/jad.html
// Decompiler options: packimports(3) fieldsfirst ansi space 
// Source File Name:   TRNonBlockingServer.java

package org.gudy.azureus2.core3.tracker.server.impl.tcp.nonblocking;

import com.aelitis.azureus.core.networkmanager.*;
import com.aelitis.azureus.core.networkmanager.admin.NetworkAdmin;
import java.net.*;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import org.gudy.azureus2.core3.config.COConfigurationManager;
import org.gudy.azureus2.core3.ipfilter.IpFilter;
import org.gudy.azureus2.core3.logging.*;
import org.gudy.azureus2.core3.tracker.server.TRTrackerServerException;
import org.gudy.azureus2.core3.tracker.server.impl.tcp.TRTrackerServerTCP;
import org.gudy.azureus2.core3.util.*;

// Referenced classes of package org.gudy.azureus2.core3.tracker.server.impl.tcp.nonblocking:
//			TRNonBlockingServerProcessor, TRNonBlockingServerProcessorFactory

public class TRNonBlockingServer extends TRTrackerServerTCP
	implements com.aelitis.azureus.core.networkmanager.VirtualServerChannelSelector.SelectListener
{

	private static final LogIDs LOGID;
	private static final int TIMEOUT_CHECK_INTERVAL = 10000;
	private static final int CLOSE_DELAY = 5000;
	private TRNonBlockingServerProcessorFactory processor_factory;
	private final VirtualChannelSelector read_selector;
	private final VirtualChannelSelector write_selector;
	private List connections_to_close;
	private List processors;
	private long total_timeouts;
	private long total_connections;
	public static final int MAX_CONCURRENT_CONNECTIONS = COConfigurationManager.getIntParameter("Tracker TCP NonBlocking Conc Max");
	private final AEMonitor this_mon;
	private VirtualServerChannelSelector accept_server;
	private volatile boolean closed;

	public TRNonBlockingServer(String _name, int _port, InetAddress _bind_ip, boolean _apply_ip_filter, TRNonBlockingServerProcessorFactory _processor_factory)
		throws TRTrackerServerException
	{
		this(_name, _port, _bind_ip, _apply_ip_filter, true, _processor_factory);
	}

	public TRNonBlockingServer(String _name, int _port, InetAddress _bind_ip, boolean _apply_ip_filter, boolean _start_up_ready, TRNonBlockingServerProcessorFactory _processor_factory)
		throws TRTrackerServerException
	{
		Exception exception;
		super(_name, _port, false, _apply_ip_filter, _start_up_ready);
		connections_to_close = new ArrayList();
		processors = new ArrayList();
		this_mon = new AEMonitor("TRNonBlockingServer");
		processor_factory = _processor_factory;
		read_selector = new VirtualChannelSelector((new StringBuilder()).append(_name).append(":").append(_port).toString(), 1, false);
		write_selector = new VirtualChannelSelector((new StringBuilder()).append(_name).append(":").append(_port).toString(), 4, true);
		boolean ok = false;
		if (_port == 0)
			throw new TRTrackerServerException("port of 0 not currently supported");
		try
		{
			InetSocketAddress address;
			if (_bind_ip == null)
			{
				_bind_ip = NetworkAdmin.getSingleton().getSingleHomedServiceBindAddress();
				if (_bind_ip == null)
					address = new InetSocketAddress(_port);
				else
					address = new InetSocketAddress(_bind_ip, _port);
			} else
			{
				address = new InetSocketAddress(_bind_ip, _port);
			}
			accept_server = VirtualServerChannelSelectorFactory.createBlocking(address, 0, this);
			accept_server.start();
			AEThread read_thread = new AEThread("TRTrackerServer:readSelector") {

				final TRNonBlockingServer this$0;

				public void runSupport()
				{
					selectLoop(read_selector);
				}

			
			{
				this$0 = TRNonBlockingServer.this;
				super(x0);
			}
			};
			read_thread.setDaemon(true);
			read_thread.start();
			AEThread write_thread = new AEThread("TRTrackerServer:writeSelector") {

				final TRNonBlockingServer this$0;

				public void runSupport()
				{
					selectLoop(write_selector);
				}

			
			{
				this$0 = TRNonBlockingServer.this;
				super(x0);
			}
			};
			write_thread.setDaemon(true);
			write_thread.start();
			AEThread close_thread = new AEThread("TRTrackerServer:closeScheduler") {

				final TRNonBlockingServer this$0;

				public void runSupport()
				{
					closeLoop();
				}

			
			{
				this$0 = TRNonBlockingServer.this;
				super(x0);
			}
			};
			close_thread.setDaemon(true);
			close_thread.start();
			Logger.log(new LogEvent(LOGID, (new StringBuilder()).append("TRTrackerServer: Non-blocking listener established on port ").append(getPort()).toString()));
			ok = true;
		}
		catch (Throwable e)
		{
			Logger.logTextResource(new LogAlert(false, 3, "Tracker.alert.listenfail"), new String[] {
				(new StringBuilder()).append("").append(getPort()).toString()
			});
			throw new TRTrackerServerException("TRTrackerServer: accept fails", e);
		}
		finally
		{
			if (ok) goto _L0; else goto _L0
		}
		if (!ok)
			destroySupport();
		break MISSING_BLOCK_LABEL_404;
		destroySupport();
		throw exception;
	}

	protected void selectLoop(VirtualChannelSelector selector)
	{
		long last_time = 0L;
		do
		{
			if (closed)
				break;
			try
			{
				selector.select(100L);
				if (selector == read_selector)
				{
					long now = SystemTime.getCurrentTime();
					if (now < last_time)
						last_time = now;
					else
					if (now - last_time >= 10000L)
					{
						last_time = now;
						checkTimeouts(now);
					}
				}
			}
			catch (Throwable e)
			{
				Debug.printStackTrace(e);
			}
		} while (true);
	}

	public void newConnectionAccepted(ServerSocketChannel server, SocketChannel channel)
	{
		final TRNonBlockingServerProcessor processor = processor_factory.create(this, channel);
		int num_processors;
		this_mon.enter();
		total_connections++;
		processors.add(processor);
		num_processors = processors.size();
		this_mon.exit();
		break MISSING_BLOCK_LABEL_73;
		Exception exception;
		exception;
		this_mon.exit();
		throw exception;
		if (MAX_CONCURRENT_CONNECTIONS != 0 && num_processors > MAX_CONCURRENT_CONNECTIONS)
			removeAndCloseConnection(processor);
		else
		if (isIPFilterEnabled() && ip_filter.isInRange(channel.socket().getInetAddress().getHostAddress(), "Tracker", null))
		{
			removeAndCloseConnection(processor);
		} else
		{
			com.aelitis.azureus.core.networkmanager.VirtualChannelSelector.VirtualSelectorListener read_listener = new com.aelitis.azureus.core.networkmanager.VirtualChannelSelector.VirtualSelectorListener() {

				private boolean selector_registered;
				final TRNonBlockingServerProcessor val$processor;
				final TRNonBlockingServer this$0;

				public boolean selectSuccess(VirtualChannelSelector selector, SocketChannel sc, Object attachment)
				{
					int read_result;
					read_result = processor.processRead();
					if (read_result == 0)
					{
						if (selector_registered)
							read_selector.pauseSelects(sc);
					} else
					if (read_result < 0)
						removeAndCloseConnection(processor);
					else
					if (!selector_registered)
					{
						selector_registered = true;
						read_selector.register(sc, this, null);
					}
					return read_result != 2;
					Throwable e;
					e;
					Debug.printStackTrace(e);
					removeAndCloseConnection(processor);
					return false;
				}

				public void selectFailure(VirtualChannelSelector selector, SocketChannel sc, Object attachment, Throwable msg)
				{
					removeAndCloseConnection(processor);
				}

			
			{
				this$0 = TRNonBlockingServer.this;
				processor = trnonblockingserverprocessor;
				super();
			}
			};
			read_listener.selectSuccess(read_selector, channel, null);
		}
		return;
	}

	protected void readyToWrite(final TRNonBlockingServerProcessor processor)
	{
		com.aelitis.azureus.core.networkmanager.VirtualChannelSelector.VirtualSelectorListener write_listener = new com.aelitis.azureus.core.networkmanager.VirtualChannelSelector.VirtualSelectorListener() {

			private boolean selector_registered;
			final TRNonBlockingServerProcessor val$processor;
			final TRNonBlockingServer this$0;

			public boolean selectSuccess(VirtualChannelSelector selector, SocketChannel sc, Object attachment)
			{
				int write_result;
				write_result = processor.processWrite();
				if (write_result > 0)
				{
					if (selector_registered)
					{
						write_selector.resumeSelects(sc);
					} else
					{
						selector_registered = true;
						write_selector.register(sc, this, null);
					}
				} else
				if (write_result == 0)
					removeAndCloseConnection(processor);
				else
				if (write_result < 0)
				{
					processor.failed();
					removeAndCloseConnection(processor);
				}
				return write_result != 2;
				Throwable e;
				e;
				Debug.printStackTrace(e);
				removeAndCloseConnection(processor);
				return false;
			}

			public void selectFailure(VirtualChannelSelector selector, SocketChannel sc, Object attachment, Throwable msg)
			{
				removeAndCloseConnection(processor);
			}

			
			{
				this$0 = TRNonBlockingServer.this;
				processor = trnonblockingserverprocessor;
				super();
			}
		};
		write_listener.selectSuccess(write_selector, processor.getSocketChannel(), null);
	}

	protected void removeAndCloseConnection(TRNonBlockingServerProcessor processor)
	{
		processor.completed();
		this_mon.enter();
		if (processors.remove(processor))
		{
			read_selector.cancel(processor.getSocketChannel());
			write_selector.cancel(processor.getSocketChannel());
			connections_to_close.add(processor);
		}
		this_mon.exit();
		break MISSING_BLOCK_LABEL_77;
		Exception exception;
		exception;
		this_mon.exit();
		throw exception;
	}

	public void checkTimeouts(long now)
	{
		this_mon.enter();
		List new_processors = new ArrayList(processors.size());
		for (int i = 0; i < processors.size(); i++)
		{
			TRNonBlockingServerProcessor processor = (TRNonBlockingServerProcessor)processors.get(i);
			if (now - processor.getStartTime() > PROCESSING_GET_LIMIT)
			{
				read_selector.cancel(processor.getSocketChannel());
				write_selector.cancel(processor.getSocketChannel());
				connections_to_close.add(processor);
				total_timeouts++;
			} else
			{
				new_processors.add(processor);
			}
		}

		processors = new_processors;
		this_mon.exit();
		break MISSING_BLOCK_LABEL_162;
		Exception exception;
		exception;
		this_mon.exit();
		throw exception;
	}

	public void closeLoop()
	{
		List pending_list;
		long default_delay;
		long delay;
		pending_list = new ArrayList();
		default_delay = 3333L;
		delay = default_delay;
_L3:
		if (closed) goto _L2; else goto _L1
_L1:
		long start;
		if (delay > 0L)
			try
			{
				Thread.sleep(delay);
			}
			catch (Throwable e)
			{
				Debug.printStackTrace(e);
			}
		start = SystemTime.getCurrentTime();
		for (int i = 0; i < pending_list.size(); i++)
			try
			{
				TRNonBlockingServerProcessor processor = (TRNonBlockingServerProcessor)pending_list.get(i);
				processor.closed();
				processor.getSocketChannel().close();
			}
			catch (Throwable e) { }

		this_mon.enter();
		pending_list = connections_to_close;
		connections_to_close = new ArrayList();
		this_mon.exit();
		break MISSING_BLOCK_LABEL_145;
		Exception exception;
		exception;
		this_mon.exit();
		throw exception;
		long duration = SystemTime.getCurrentTime() - start;
		if (duration < 0L)
			duration = 0L;
		delay = default_delay - duration;
		  goto _L3
_L2:
	}

	protected void closeSupport()
	{
		closed = true;
		accept_server.stop();
		destroySupport();
	}

	static 
	{
		LOGID = LogIDs.TRACKER;
	}


}