Few critical bugs are fixed

src/com/esotericsoftware/kryonet/TcpConnection.java

@@ -46,10 +46,14 @@ class TcpConnection {
 	private volatile long lastWriteTime, lastReadTime;
 	private int currentObjectLength;
 	private final Object writeLock = new Object();
+	private final int objectBufferSize;
+	private final int lengthLength;
 
 	public TcpConnection (Serialization serialization, int writeBufferSize, int objectBufferSize) {
 		this.serialization = serialization;
-		writeBuffer = ByteBuffer.allocate(writeBufferSize);
+		this.objectBufferSize = objectBufferSize;
+		lengthLength = serialization.getLengthLength();
+		writeBuffer = ByteBuffer.allocate(Math.max(writeBufferSize, lengthLength + objectBufferSize));
 		readBuffer = ByteBuffer.allocate(objectBufferSize);
 		readBuffer.flip();
 	}
@@ -200,9 +204,21 @@ public int send (Connection connection, Object object) throws IOException {
 		SocketChannel socketChannel = this.socketChannel;
 		if (socketChannel == null) throw new SocketException("Connection is closed.");
 		synchronized (writeLock) {
+			int lengthLength = serialization.getLengthLength();
+
+			//wait while buffer has enough free space.
+			//thats because message length will be put only after the whole message will be serialized
+            while (writeBuffer.capacity() < writeBuffer.position() + lengthLength + objectBufferSize) {
+                try {
+                    Thread.sleep(10);
+                } catch (InterruptedException e) {
+                    break;
+                }
+                writeToSocket();
+	        }
+
 			// Leave room for length.
 			int start = writeBuffer.position();
-			int lengthLength = serialization.getLengthLength();
 			writeBuffer.position(writeBuffer.position() + lengthLength);
 
 			// Write data.

src/com/esotericsoftware/kryonet/rmi/ObjectSpace.java

@@ -68,7 +68,7 @@ public class ObjectSpace {
 
 	static private final Object instancesLock = new Object();
 	static ObjectSpace[] instances = new ObjectSpace[0];
-	static private final HashMap<Class, CachedMethod[]> methodCache = new HashMap();
+	static private final HashMap<Kryo, HashMap<Class, CachedMethod[]>> methodCache = new HashMap();
 	static private boolean asm = true;
 
 	final IntMap idToObject = new IntMap();
@@ -306,11 +306,13 @@ static private class RemoteInvocationHandler implements InvocationHandler {
 		private Byte lastResponseID;
 		private byte nextResponseId = 1;
 		private Listener responseListener;
+		private volatile int totalPendingResponsesCnt = 0;
 
 		final ReentrantLock lock = new ReentrantLock();
 		final Condition responseCondition = lock.newCondition();
-		final InvokeMethodResult[] responseTable = new InvokeMethodResult[64];
-		final boolean[] pendingResponses = new boolean[64];
+		final InvokeMethodResult[] responseTable = new InvokeMethodResult[responseIdMask+1];
+		final boolean[] pendingResponses = new boolean[responseIdMask+1];
+		final Object invokeLocker = new Object();
 
 		public RemoteInvocationHandler (Connection connection, final int objectID) {
 			super();
@@ -404,11 +406,25 @@ public Object invoke (Object proxy, Method method, Object[] args) throws Excepti
 			boolean needsResponse = !udp && (transmitReturnValue || transmitExceptions || !nonBlocking);
 			byte responseID = 0;
 			if (needsResponse) {
-				synchronized (this) {
-					// Increment the response counter and put it into the low bits of the responseID.
-					responseID = nextResponseId++;
-					if (nextResponseId > responseIdMask) nextResponseId = 1;
-					pendingResponses[responseID] = true;
+				synchronized (invokeLocker) {
+					while (true) {
+						synchronized (this) {
+							if (totalPendingResponsesCnt < responseIdMask) {
+								// Find the first non-pending responseID.
+								do {
+									responseID = nextResponseId++;
+									if (nextResponseId > responseIdMask) nextResponseId = 1;
+								} while (pendingResponses[responseID]);
+								pendingResponses[responseID] = true;
+								totalPendingResponsesCnt++;
+								break;
+							}
+						}
+						//This sleep is under lock "synchronized(invokeLocker)" but not under lock "synchronized(this)"
+						//So all sequenced invocations will wait (and totalPendingResponsesCnt could not increased),
+						// but invocation result can be handled (and totalPendingResponsesCnt could decreased)
+						Thread.sleep(10);
+					}
 				}
 				// Pack other data into the high bits.
 				byte responseData = responseID;
@@ -454,8 +470,11 @@ public Object invoke (Object proxy, Method method, Object[] args) throws Excepti
 				throw new TimeoutException("Response timed out: " + method.getDeclaringClass().getName() + "." + method.getName());
 			} finally {
 				synchronized (this) {
-					pendingResponses[responseID] = false;
-					responseTable[responseID] = null;
+					if (pendingResponses[responseID]) {
+						totalPendingResponsesCnt--;
+						pendingResponses[responseID] = false;
+						responseTable[responseID] = null;
+					}
 				}
 			}
 		}
@@ -562,7 +581,12 @@ static public class InvokeMethodResult implements FrameworkMessage {
 	}
 
 	static CachedMethod[] getMethods (Kryo kryo, Class type) {
-		CachedMethod[] cachedMethods = methodCache.get(type); // Maybe should cache per Kryo instance?
+		HashMap<Class, CachedMethod[]> cache = methodCache.get(kryo);
+		if (cache == null) {
+			cache = new HashMap<Class, CachedMethod[]>();
+			methodCache.put(kryo, cache);
+		}
+		CachedMethod[] cachedMethods = cache.get(type);
 		if (cachedMethods != null) return cachedMethods;
 
 		ArrayList<Method> allMethods = new ArrayList();
@@ -630,7 +654,7 @@ public int compare (Method o1, Method o2) {
 
 			cachedMethods[i] = cachedMethod;
 		}
-		methodCache.put(type, cachedMethods);
+		cache.put(type, cachedMethods);
 		return cachedMethods;
 	}
 

test/com/esotericsoftware/kryonet/SlowServerTest.java

@@ -0,0 +1,110 @@
+/* Copyright (c) 2008, Nathan Sweet
+ * All rights reserved.
+ * 
+ * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following
+ * conditions are met:
+ * 
+ * - Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
+ * - Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following
+ * disclaimer in the documentation and/or other materials provided with the distribution.
+ * - Neither the name of Esoteric Software nor the names of its contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING,
+ * BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
+ * SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */
+
+package com.esotericsoftware.kryonet;
+
+import com.esotericsoftware.kryo.Kryo;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SlowServerTest extends KryoNetTestCase {
+	AtomicInteger receivedBytes = new AtomicInteger();
+
+	public void testManyLargeMessages () throws IOException {
+		final int messageCount = 1024;
+		int objectBufferSize = 10250;
+		int writeBufferSize = 10250 * messageCount / 2;
+
+		Server server = new Server(writeBufferSize, objectBufferSize);
+		startEndPoint(server);
+		register(server.getKryo());
+		server.bind(tcpPort);
+		final AtomicInteger serverReceived = new AtomicInteger();
+		final AtomicInteger clientReceived = new AtomicInteger();
+
+		server.addListener(new Listener() {
+			AtomicInteger receivedBytes = new AtomicInteger();
+
+			public void received (Connection connection, Object object) {
+				if (object instanceof LargeMessage) {
+					System.out.println("Server sending message: " + serverReceived.get());
+					connection.sendTCP(object);
+
+					receivedBytes.addAndGet(((LargeMessage)object).bytes.length);
+
+					int count = serverReceived.incrementAndGet();
+					System.out.println("Server received " + count + " messages.");
+					if (count == messageCount) {
+						System.out.println("Server received all " + messageCount + " messages!");
+						System.out.println("Server received and sent " + receivedBytes.get() + " bytes.");
+					}
+				}
+			}
+		});
+
+		final Client client = new Client(writeBufferSize, objectBufferSize);
+		startEndPoint(client);
+		register(client.getKryo());
+		client.connect(5000, host, tcpPort);
+
+		client.addListener(new Listener() {
+			AtomicInteger receivedBytes = new AtomicInteger();
+
+			public void received (Connection connection, Object object) {
+				if (object instanceof LargeMessage) {
+					int count = clientReceived.incrementAndGet();
+					System.out.println("Client received " + count + " messages.");
+					if (count == messageCount) {
+						System.out.println("Client received all " + messageCount + " messages!");
+						System.out.println("Client received and sent " + receivedBytes.get() + " bytes.");
+						stopEndPoints();
+					}
+				}
+			}
+		});
+
+		byte[] b = new byte[1024 * 10];
+		for (int i = 0; i < messageCount; i++) {
+			System.out.println("Client sending: " + i);
+			client.sendTCP(new LargeMessage(b));
+		}
+		System.out.println("Client has queued " + messageCount + " messages.");
+
+		waitForThreads(5000);
+		assertEquals(messageCount, clientReceived.get());
+		assertEquals(messageCount, serverReceived.get());
+	}
+
+	private void register (Kryo kryo) {
+		kryo.register(byte[].class);
+		kryo.register(LargeMessage.class);
+	}
+
+	public static class LargeMessage {
+		public byte[] bytes;
+
+		public LargeMessage () {
+		}
+
+		public LargeMessage (byte[] bytes) {
+			this.bytes = bytes;
+		}
+	}
+}