init, thx MrExtremez

2006Redone file_server/src/org/apollo/jagcached/dispatch/ChannelRequest.java

@@ -0,0 +1,58 @@
+package org.apollo.jagcached.dispatch;
+
+import org.jboss.netty.channel.Channel;
+
+/**
+ * A specialised request which contains a channel as well as the request object
+ * itself.
+ * @author Graham
+ * @param <T> The type of request.
+ */
+public final class ChannelRequest<T> implements Comparable<ChannelRequest<T>> {
+	
+	/**
+	 * The channel.
+	 */
+	private final Channel channel;
+	
+	/**
+	 * The request.
+	 */
+	private final T request;
+	
+	/**
+	 * Creates a new channel request.
+	 * @param channel The channel.
+	 * @param request The request.
+	 */
+	public ChannelRequest(Channel channel, T request) {
+		this.channel = channel;
+		this.request = request;
+	}
+	
+	/**
+	 * Gets the channel.
+	 * @return The channel.
+	 */
+	public Channel getChannel() {
+		return channel;
+	}
+	
+	/**
+	 * Gets the request.
+	 * @return The request.
+	 */
+	public T getRequest() {
+		return request;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public int compareTo(ChannelRequest<T> o) {
+		if (request instanceof Comparable<?> && o.request instanceof Comparable<?>) {
+			return ((Comparable<T>) request).compareTo(o.request);
+		}
+		return 0;
+	}
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/HttpRequestWorker.java

@@ -0,0 +1,139 @@
+package org.apollo.jagcached.dispatch;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.Date;
+
+import org.apollo.jagcached.fs.IndexedFileSystem;
+import org.apollo.jagcached.resource.CombinedResourceProvider;
+import org.apollo.jagcached.resource.HypertextResourceProvider;
+import org.apollo.jagcached.resource.ResourceProvider;
+import org.apollo.jagcached.resource.VirtualResourceProvider;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
+import org.jboss.netty.handler.codec.http.HttpRequest;
+import org.jboss.netty.handler.codec.http.HttpResponse;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * A worker which services HTTP requests.
+ * @author Graham
+ */
+public final class HttpRequestWorker extends RequestWorker<HttpRequest, ResourceProvider> {
+
+	/**
+	 * The value of the server header.
+	 */
+	private static final String SERVER_IDENTIFIER = "JAGeX/3.1";
+	
+	/**
+	 * The directory with web files.
+	 */
+	private static final File WWW_DIRECTORY = new File("./data/www/");
+
+	/**
+	 * The default character set.
+	 */
+	private static final Charset CHARACTER_SET = Charset.forName("ISO-8859-1");
+
+	/**
+	 * Creates the HTTP request worker.
+	 * @param fs The file system.
+	 */
+	public HttpRequestWorker(IndexedFileSystem fs) {
+		super(new CombinedResourceProvider(new VirtualResourceProvider(fs), new HypertextResourceProvider(WWW_DIRECTORY)));
+	}
+
+	@Override
+	protected ChannelRequest<HttpRequest> nextRequest() throws InterruptedException {
+		return RequestDispatcher.nextHttpRequest();
+	}
+
+	@Override
+	protected void service(ResourceProvider provider, Channel channel, HttpRequest request) throws IOException {
+		String path = request.getUri();
+		ByteBuffer buf = provider.get(path);
+		
+		ChannelBuffer wrappedBuf;
+		HttpResponseStatus status = HttpResponseStatus.OK;
+		
+		String mimeType = getMimeType(request.getUri());
+		
+		if (buf == null) {
+			status = HttpResponseStatus.NOT_FOUND;
+			wrappedBuf = createErrorPage(status, "The page you requested could not be found.");
+			mimeType = "text/html";
+		} else {
+			wrappedBuf = ChannelBuffers.wrappedBuffer(buf);
+		}
+		
+		HttpResponse resp = new DefaultHttpResponse(request.getProtocolVersion(), status);
+		
+		resp.setHeader("Date", new Date());
+		resp.setHeader("Server", SERVER_IDENTIFIER);
+		resp.setHeader("Content-type", mimeType + ", charset=" + CHARACTER_SET.name());
+		resp.setHeader("Cache-control", "no-cache");
+		resp.setHeader("Pragma", "no-cache");
+		resp.setHeader("Expires", new Date(0));
+		resp.setHeader("Connection", "close");
+		resp.setHeader("Content-length", wrappedBuf.readableBytes());
+		resp.setChunked(false);
+		resp.setContent(wrappedBuf);
+		
+		channel.write(resp).addListener(ChannelFutureListener.CLOSE);
+	}
+
+	/**
+	 * Gets the MIME type of a file by its name.
+	 * @param name The file name.
+	 * @return The MIME type.
+	 */
+	private String getMimeType(String name) {
+		if (name.endsWith(".htm") || name.endsWith(".html")) {
+			return "text/html";
+		} else if (name.endsWith(".css")) {
+			return "text/css";
+		} else if (name.endsWith(".js")) {
+			return "text/javascript";
+		} else if (name.endsWith(".jpg") || name.endsWith(".jpeg")) {
+			return "image/jpeg";
+		} else if (name.endsWith(".gif")) {
+			return "image/gif";
+		} else if (name.endsWith(".png")) {
+			return "image/png";
+		} else if (name.endsWith(".txt")) {
+			return "text/plain";
+		}
+		return "application/octect-stream";
+	}
+
+	/**
+	 * Creates an error page.
+	 * @param status The HTTP status.
+	 * @param description The error description.
+	 * @return The error page as a buffer.
+	 */
+	private ChannelBuffer createErrorPage(HttpResponseStatus status, String description) {
+		String title = status.getCode() + " " + status.getReasonPhrase();
+		
+		StringBuilder bldr = new StringBuilder();
+		
+		bldr.append("<!DOCTYPE html><html><head><title>");
+		bldr.append(title);
+		bldr.append("</title></head><body><h1>");
+		bldr.append(title);
+		bldr.append("</h1><p>");
+		bldr.append(description);
+		bldr.append("</p><hr /><address>");
+		bldr.append(SERVER_IDENTIFIER);
+		bldr.append(" Server</address></body></html>");
+		
+		return ChannelBuffers.copiedBuffer(bldr.toString(), Charset.defaultCharset());
+	}
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/JagGrabRequestWorker.java

@@ -0,0 +1,46 @@
+package org.apollo.jagcached.dispatch;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apollo.jagcached.fs.IndexedFileSystem;
+import org.apollo.jagcached.net.jaggrab.JagGrabRequest;
+import org.apollo.jagcached.net.jaggrab.JagGrabResponse;
+import org.apollo.jagcached.resource.ResourceProvider;
+import org.apollo.jagcached.resource.VirtualResourceProvider;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFutureListener;
+
+/**
+ * A worker which services JAGGRAB requests.
+ * @author Graham
+ */
+public final class JagGrabRequestWorker extends RequestWorker<JagGrabRequest, ResourceProvider> {
+
+	/**
+	 * Creates the JAGGRAB request worker.
+	 * @param fs The file system.
+	 */
+	public JagGrabRequestWorker(IndexedFileSystem fs) {
+		super(new VirtualResourceProvider(fs));
+	}
+
+	@Override
+	protected ChannelRequest<JagGrabRequest> nextRequest() throws InterruptedException {
+		return RequestDispatcher.nextJagGrabRequest();
+	}
+
+	@Override
+	protected void service(ResourceProvider provider, Channel channel, JagGrabRequest request) throws IOException {
+		ByteBuffer buf = provider.get(request.getFilePath());
+		if (buf == null) {
+			channel.close();
+		} else {
+			ChannelBuffer wrapped = ChannelBuffers.wrappedBuffer(buf);
+			channel.write(new JagGrabResponse(wrapped)).addListener(ChannelFutureListener.CLOSE);
+		}
+	}
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/OnDemandRequestWorker.java

@@ -0,0 +1,61 @@
+package org.apollo.jagcached.dispatch;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+
+import org.apollo.jagcached.fs.FileDescriptor;
+import org.apollo.jagcached.fs.IndexedFileSystem;
+import org.apollo.jagcached.net.ondemand.OnDemandRequest;
+import org.apollo.jagcached.net.ondemand.OnDemandResponse;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+
+/**
+ * A worker which services 'on-demand' requests.
+ * @author Graham
+ */
+public final class OnDemandRequestWorker extends RequestWorker<OnDemandRequest, IndexedFileSystem> {
+	
+	/**
+	 * The maximum length of a chunk, in bytes.
+	 */
+	private static final int CHUNK_LENGTH = 500;
+
+	/**
+	 * Creates the 'on-demand' request worker.
+	 * @param fs The file system.
+	 */
+	public OnDemandRequestWorker(IndexedFileSystem fs) {
+		super(fs);
+	}
+
+	@Override
+	protected ChannelRequest<OnDemandRequest> nextRequest() throws InterruptedException {
+		return RequestDispatcher.nextOnDemandRequest();
+	}
+
+	@Override
+	protected void service(IndexedFileSystem fs, Channel channel, OnDemandRequest request) throws IOException {
+		FileDescriptor desc = request.getFileDescriptor();
+				
+		ByteBuffer buf = fs.getFile(desc);
+		int length = buf.remaining();
+		
+		for (int chunk = 0; buf.remaining() > 0; chunk++) {
+			int chunkSize = buf.remaining();
+			if (chunkSize > CHUNK_LENGTH) {
+				chunkSize = CHUNK_LENGTH;
+			}
+			
+			byte[] tmp = new byte[chunkSize];
+			buf.get(tmp, 0, tmp.length);
+			ChannelBuffer chunkData = ChannelBuffers.wrappedBuffer(tmp, 0, chunkSize);
+			
+			OnDemandResponse response = new OnDemandResponse(desc, length, chunk, chunkData);
+			channel.write(response);
+		}
+	}
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/RequestDispatcher.java

@@ -0,0 +1,112 @@
+package org.apollo.jagcached.dispatch;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+
+
+import org.apollo.jagcached.net.jaggrab.JagGrabRequest;
+import org.apollo.jagcached.net.ondemand.OnDemandRequest;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.handler.codec.http.HttpRequest;
+
+/**
+ * A class which dispatches requests to worker threads.
+ * @author Graham
+ */
+public final class RequestDispatcher {
+	
+	/**
+	 * The maximum size of a queue before requests are rejected.
+	 */
+	private static final int MAXIMUM_QUEUE_SIZE = 1024;
+	
+	/**
+	 * A queue for pending 'on-demand' requests.
+	 */
+	private static final BlockingQueue<ChannelRequest<OnDemandRequest>> onDemandQueue = new PriorityBlockingQueue<ChannelRequest<OnDemandRequest>>();
+
+	/**
+	 * A queue for pending JAGGRAB requests.
+	 */
+	private static final BlockingQueue<ChannelRequest<JagGrabRequest>> jagGrabQueue = new LinkedBlockingQueue<ChannelRequest<JagGrabRequest>>();
+	
+	/**
+	 * A queue for pending HTTP requests.
+	 */
+	private static final BlockingQueue<ChannelRequest<HttpRequest>> httpQueue = new LinkedBlockingQueue<ChannelRequest<HttpRequest>>();
+	
+	/**
+	 * Gets the next 'on-demand' request from the queue, blocking if none are
+	 * available.
+	 * @return The 'on-demand' request.
+	 * @throws InterruptedException if the thread is interrupted.
+	 */
+	static ChannelRequest<OnDemandRequest> nextOnDemandRequest() throws InterruptedException {
+		return onDemandQueue.take();
+	}
+	
+	/**
+	 * Gets the next JAGGRAB request from the queue, blocking if none are
+	 * available.
+	 * @return The JAGGRAB request.
+	 * @throws InterruptedException if the thread is interrupted.
+	 */
+	static ChannelRequest<JagGrabRequest> nextJagGrabRequest() throws InterruptedException {
+		return jagGrabQueue.take();
+	}
+	
+	/**
+	 * Gets the next HTTP request from the queue, blocking if none are
+	 * available.
+	 * @return The HTTP request.
+	 * @throws InterruptedException if the thread is interrupted.
+	 */
+	static ChannelRequest<HttpRequest> nextHttpRequest() throws InterruptedException {
+		return httpQueue.take();
+	}
+	
+	/**
+	 * Dispatches an 'on-demand' request.
+	 * @param channel The channel.
+	 * @param request The request.
+	 */
+	public static void dispatch(Channel channel, OnDemandRequest request) {
+		if (onDemandQueue.size() >= MAXIMUM_QUEUE_SIZE) {
+			channel.close();
+		}
+		onDemandQueue.add(new ChannelRequest<OnDemandRequest>(channel, request));
+	}
+
+	/**
+	 * Dispatches a JAGGRAB request.
+	 * @param channel The channel.
+	 * @param request The request.
+	 */
+	public static void dispatch(Channel channel, JagGrabRequest request) {
+		if (jagGrabQueue.size() >= MAXIMUM_QUEUE_SIZE) {
+			channel.close();
+		}
+		jagGrabQueue.add(new ChannelRequest<JagGrabRequest>(channel, request));
+	}
+
+	/**
+	 * Dispatches a HTTP request.
+	 * @param channel The channel.
+	 * @param request The request.
+	 */
+	public static void dispatch(Channel channel, HttpRequest request) {
+		if (httpQueue.size() >= MAXIMUM_QUEUE_SIZE) {
+			channel.close();
+		}
+		httpQueue.add(new ChannelRequest<HttpRequest>(channel, request));
+	}
+	
+	/**
+	 * Default private constructor to prevent instantiation.
+	 */
+	private RequestDispatcher() {
+		
+	}
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/RequestWorker.java

@@ -0,0 +1,90 @@
+package org.apollo.jagcached.dispatch;
+
+import java.io.IOException;
+
+import org.jboss.netty.channel.Channel;
+
+/**
+ * The base class for request workers.
+ * @author Graham
+ * @param <T> The type of request.
+ * @param <P> The type of provider.
+ */
+public abstract class RequestWorker<T, P> implements Runnable {
+	
+	/**
+	 * The resource provider.
+	 */
+	private final P provider;
+	
+	/**
+	 * An object used for locking checks to see if the worker is running.
+	 */
+	private final Object lock = new Object();
+	
+	/**
+	 * A flag indicating if the worker should be running.
+	 */
+	private boolean running = true;
+	
+	/**
+	 * Creates the request worker with the specified file system.
+	 * @param provider The resource provider.
+	 */
+	public RequestWorker(P provider) {
+		this.provider = provider;
+	}
+
+	/**
+	 * Stops this worker. The worker's thread may need to be interrupted.
+	 */
+	public final void stop() {
+		synchronized (lock) {
+			running = false;
+		}
+	}
+		
+	@Override
+	public final void run() {
+		while (true) {
+			synchronized (lock) {
+				if (!running) {
+					break;
+				}
+			}
+			
+			ChannelRequest<T> request;
+			try {
+				request = nextRequest();
+			} catch (InterruptedException e) {
+				continue;
+			}
+			
+			Channel channel = request.getChannel();
+			
+			try {
+				service(provider, channel, request.getRequest());
+			} catch (IOException e) {
+				e.printStackTrace();
+				channel.close();
+			}
+		}
+	}
+
+	/**
+	 * Gets the next request.
+	 * @return The next request.
+	 * @throws InterruptedException if the thread is interrupted.
+	 */
+	protected abstract ChannelRequest<T> nextRequest() throws InterruptedException;
+	
+	/**
+	 * Services a request.
+	 * @param provider The resource provider.
+	 * @param channel The channel.
+	 * @param request The request to service.
+	 * @throws IOException if an I/O error occurs.
+	 */
+	protected abstract void service(P provider, Channel channel, T request) throws IOException;
+
+}

2006Redone file_server/src/org/apollo/jagcached/dispatch/RequestWorkerPool.java

@@ -0,0 +1,75 @@
+package org.apollo.jagcached.dispatch;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apollo.jagcached.Constants;
+import org.apollo.jagcached.fs.IndexedFileSystem;
+
+/**
+ * A class which manages the pool of request workers.
+ * @author Graham
+ * @author Ryley Kimmel <ryley.kimmel@live.com>
+ */
+public final class RequestWorkerPool {
+	
+	/**
+	 * The number of threads per request type.
+	 */
+	private static final int THREADS_PER_REQUEST_TYPE = Runtime.getRuntime().availableProcessors();
+	
+	/**
+	 * The number of request types.
+	 */
+	private static final int REQUEST_TYPES = 3;
+	
+	/**
+	 * The executor service.
+	 */
+	private final ExecutorService service;
+	
+	/**
+	 * A list of request workers.
+	 */
+	private final List<RequestWorker<?, ?>> workers = new ArrayList<RequestWorker<?, ?>>();
+	
+	/**
+	 * The request worker pool.
+	 */
+	public RequestWorkerPool() {
+		int totalThreads = REQUEST_TYPES * THREADS_PER_REQUEST_TYPE;
+		service = Executors.newFixedThreadPool(totalThreads);
+	}
+
+	/**
+	 * Starts the threads in the pool.
+	 * @throws Exception if the file system cannot be created.
+	 */
+	public void start() throws Exception {
+		File base = new File(Constants.FILE_SYSTEM_DIR);
+		for (int i = 0; i < THREADS_PER_REQUEST_TYPE; i++) {
+			workers.add(new JagGrabRequestWorker(new IndexedFileSystem(base, true)));
+			workers.add(new OnDemandRequestWorker(new IndexedFileSystem(base, true)));
+			workers.add(new HttpRequestWorker(new IndexedFileSystem(base, true)));
+		}
+		
+		for (RequestWorker<?, ?> worker : workers) {
+			service.submit(worker);
+		}
+	}
+	
+	/**
+	 * Stops the threads in the pool.
+	 */
+	public void stop() {
+		for (RequestWorker<?, ?> worker : workers) {
+			worker.stop();
+		}
+		
+		service.shutdownNow();
+	}
+
+}