package net.spy.memcached;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Test queue overflow.
 */
public class QueueOverflowTest extends ClientBaseCase {

	@Override
	protected void initClient() throws Exception {
		initClient(new DefaultConnectionFactory(5, 1024) {
			@Override
			public MemcachedConnection createConnection(
					List<InetSocketAddress> addrs) throws IOException {
				MemcachedConnection rv = super.createConnection(addrs);
				rv.setGetOptimization(false);
				return rv;
			}
			@Override
			public long getOperationTimeout() {
				return 1000;
			}
		});
	}

	private void runOverflowTest(byte b[]) throws Exception {
		Collection<Future<Boolean>> c=new ArrayList<Future<Boolean>>();
		try {
			for(int i=0; i<1000; i++) {
				c.add(client.set("k" + i, 0, b));
			}
			fail("Didn't catch an illegal state exception");
		} catch(IllegalStateException e) {
			// expected
		}
		try {
			Thread.sleep(50);
			for(Future<Boolean> f : c) {
				f.get(1, TimeUnit.SECONDS);
			}
		} catch(TimeoutException e) {
			// OK, at least we got one back.
		} catch(ExecutionException e) {
			// OK, at least we got one back.
		}
		Thread.sleep(500);
		assertTrue(client.set("kx", 0, "woo").get(1, TimeUnit.SECONDS));
	}

	public void testOverflowingInputQueue() throws Exception {
		runOverflowTest(new byte[]{1});
	}

	public void testOverflowingWriteQueue() throws Exception {
		byte[] b=new byte[8192];
		Random r=new Random();
		r.nextBytes(b);
		runOverflowTest(b);
	}

	public void testOverflowingReadQueue() throws Exception {
		byte[] b=new byte[8192];
		Random r=new Random();
		r.nextBytes(b);
		client.set("x", 0, b);

		Collection<Future<Object>> c=new ArrayList<Future<Object>>();
		try {
			for(int i=0; i<1000; i++) {
				c.add(client.asyncGet("x"));
			}
			fail("Didn't catch an illegal state exception");
		} catch(IllegalStateException e) {
			// expected
		}
		try {
			Thread.sleep(50);
			for(Future<Object> f : c) {
				assertTrue(Arrays.equals(b,
					(byte[])f.get(5, TimeUnit.SECONDS)));
			}
		} catch(TimeoutException e) {
			// OK, just want to make sure the client doesn't crash
		} catch(ExecutionException e) {
			// OK, at least we got one back.
		}
		Thread.sleep(500);
		assertTrue(client.set("kx", 0, "woo").get(1, TimeUnit.SECONDS));
	}
}

@@ -214,16 +214,107 @@
 Random;%0A
+import java.util.concurrent.ArrayBlockingQueue;%0Aimport java.util.concurrent.BlockingQueue;%0A
 import j
@@ -345,32 +345,32 @@
 utionException;%0A
-
 import java.util
@@ -474,16 +474,57 @@
 ption;%0A%0A
+import net.spy.memcached.ops.Operation;%0A%0A
 /**%0A * T
@@ -662,16 +662,197 @@
 eption %7B
+%0A%0A%09%09// We're creating artificially constrained queues with the explicit%0A%09%09// goal of overrunning them to verify the client will still be%0A%09%09// functional after such conditions occur.
 %0A%09%09initC
@@ -1203,16 +1203,440 @@
 0;%0A%09%09%09%7D%0A
+%09%09%09@Override%0A%09%09%09public BlockingQueue%3COperation%3E createOperationQueue() %7B%0A%09%09%09%09return new ArrayBlockingQueue%3COperation%3E(getOpQueueLen());%0A%09%09%09%7D%0A%09%09%09@Override%0A%09%09%09public BlockingQueue%3COperation%3E createReadOperationQueue() %7B%0A%09%09%09%09return new ArrayBlockingQueue%3COperation%3E(%0A%09%09%09%09%09(int) (getOpQueueLen() * 1.1));%0A%09%09%09%7D%0A%09%09%09@Override%0A%09%09%09public BlockingQueue%3COperation%3E createWriteOperationQueue() %7B%0A%09%09%09%09return createOperationQueue();%0A%09%09%09%7D%0A
 %09%09%7D);%0A%09%7D
@@ -2205,32 +2205,32 @@
 ead.sleep(500);%0A
-
 %09%09assertTrue(cli
@@ -2218,32 +2218,80 @@
 );%0A%09%09assertTrue(
+%22Was not able to set a key after failure.%22,%0A%09%09%09%09
 client.set(%22kx%22,

Make QueueOverflowTest a bit more standalone.